ТЕХНИЧЕСКОЕ ЗАДАНИЕ: RAG с векторной БД на CPU (Chroma/Qdrant)

1. Цель задачи

Разработать локальную Retrieval-Augmented Generation (RAG) систему на ограниченных ресурсах (8GB RAM, CPU). Необходимо загрузить 100 000 текстовых документов (новости, wiki-статьи или синтезированные данные), построить векторную базу данных (Chroma или Qdrant в in-memory режиме) и реализовать пайплайн поиска + генерации ответа на русском языке. Система должна отвечать на вопросы по корпусу за время не более 500 мс (latency поиск + генерация без учёта первого прогрева). Ключевой результат работающее RAG-приложение (CLI или FastAPI), которое обрабатывает запрос пользователя и возвращает ответ с цитатами из документов при latency <500 мс на CPU.

2. Исходные данные

Если нет реального инструмента — симулируем:

1. Если нет HuggingFace — скачать датасет вручную через wget с зеркала или сгенерировать синтетику с помощью numpy и python:

   import random, json
   subjects = ["котики", "финансы", "IT", "спорт", "наука"]
   docs = []
   for i in range(100000):
       s = random.choice(subjects)
       text = f"Документ {i}: {s}. " + " ".join(random.choices(["факт", "новость", "исследование"], k=5))
       docs.append({"id": i, "text": text})
   with open("corpus.json", "w") as f:
       json.dump(docs, f)
   

2. Если нет GPU — все шаги выполняются на CPU (только CPU-модели).

3. Технологический стек

4. Этапы выполнения

Этап 1: Подготовка корпуса и окружения (2 часа)

Действия

1. Создать виртуальное окружение Python 3.10+ и установить зависимости:

   pip install chromadb sentence-transformers langchain langchain-community llama-cpp-python fastapi uvicorn
   

2. Загрузить или сгенерировать 100k документов (пункт 2). Сохранить в corpus.json.
3. Загрузить модель эмбеддингов:

   from sentence_transformers import SentenceTransformer
   model = SentenceTransformer('intfloat/multilingual-e5-small', device='cpu')
   

4. Скачать LLM в GGUF (например, Qwen2.5-1.5B-Instruct-q4_k_m.gguf) в папку models/.

Ожидаемый результат этапа Рабочее окружение, файл corpus.json, модель эмбеддингов и LLM готовы к использованию.

Этап 2: Индексация документов в векторную БД (3 часа)

Действия

1. Разбить корпус на батчи по 500 документов для избежания переполнения RAM.
2. Для каждого документа вычислить эмбеддинг с помощью model.encode(doc["text"]).
3. Загрузить векторы в Chroma:

   import chromadb
   client = chromadb.PersistentClient(path="./chroma_db")
   collection = client.create_collection(name="docs", metadata={"hnsw:space": "cosine"})
   for batch in batches:
       ids = [str(d["id"]) for d in batch]
       embeddings = [d["embedding"] for d in batch]  # предварительно закэшировать
       metadatas = [{"source": d["text"][:50]} for d in batch]
       collection.add(ids=ids, embeddings=embeddings, metadatas=metadatas)
   

4. Для Qdrant (in-memory): запустить qdrant_client.QdrantClient(":memory:") и загрузить через upsert.
5. Выполнить тестовый поиск: collection.query(query_embeddings=[...], n_results=5) — проверить скорость (ожидаем <50ms на CPU).

Ожидаемый результат этапа Хранящаяся на диске векторная БД (Chroma) с 100k записей, эмбеддинги кэшированы.

Этап 3: Построение RAG пайплайна (2 часа)

Действия

1. Реализовать функцию поиска:

   def retrieve(query: str, k=5):
       emb = model.encode(query).tolist()
       return collection.query(query_embeddings=[emb], n_results=k)
   

2. Загрузить LLM через llama-cpp-python:

   from llama_cpp import Llama
   llm = Llama(model_path="models/qwen-1.5b.gguf", n_ctx=2048, n_threads=4)
   

3. Сформировать промпт с контекстом:

   def generate(query, docs):
       context = "\n\n".join([doc["text"] for doc in docs])
       prompt = f"На основе документов:\n{context}\n\nВопрос: {query}\nОтвет:"
       return llm(prompt, max_tokens=256, temperature=0.2)["choices"][0]["text"]
   

4. Объединить в одну функцию rag_pipeline(query).
5. Написать CLI-скрипт rag_cli.py с циклом ввода.

Ожидаемый результат этапа Рабочий RAG пайплайн, принимающий вопрос и выводящий ответ.

Этап 4: Оптимизация производительности и latency (2 часа)

Действия

1. Измерить latency каждого этапа: эмбеддинг запроса, поиск, генерация.
2. Оптимизировать:
   • Кэшировать эмбеддинг запроса при повторных запросах (LRU кэш).
   • Уменьшить n_results до 3 (снижает контекст).
   • Использовать llama_cpp с n_gpu_layers=0 (CPU) и n_threads=8.
   • Включить бенчмарк: timeit для 10 запросов.
3. Если latency >500 мс — переключиться на более лёгкую LLM (TinyLlama-1.1B GGUF) или уменьшить контекст (512 токенов).
4. Запустить нагрузочный тест: 20 последовательных запросов, среднее время.

Ожидаемый результат этапа Замеры latency (цель <500 мс), применённые оптимизации.

Этап 5: Создание API (опционально, 1 час)

Действия

1. Реализовать FastAPI endpoint /ask:

   @app.post("/ask")
   async def ask(query: str):
       result = rag_pipeline(query)
       return {"answer": result}
   

2. Запустить сервер: uvicorn main:app --host 0.0.0.0 --port 8000.
3. Протестировать через curl -X POST -d "query=Какой-то вопрос" http://localhost:8000/ask.

Ожидаемый результат этапа FastAPI endpoint, готовый к демонстрации.

5. Критерии приемки (Definition of Done)

• Корпус из 100 000 документов загружен и проиндексирован в векторную БД (Chroma или Qdrant).
• Среднее время ответа (latency) на CPU не превышает 500 мс на 10 различных запросах (измерено после прогрева).
• Система корректно возвращает ответ на русском языке, содержащий информацию из корпуса (оценка экспертом 2 примеров).
• Код пайплайна без GPU-зависимостей, использование только CPU.
• Наличие скрипта rag_cli.py или FastAPI сервера с рабочим примером.
• Присутствует файл с метриками производительности (latency_table.csv).
• Модель эмбеддингов и LLM загружены локально без обращений к внешним API.
• Возможность добавить новые документы без переиндексации всех данных (Chroma поддерживает incremental).
• Использование не более 8GB RAM в процессе работы (мониторинг через psutil).
• Документация (README.md) с инструкцией по запуску.

6. Ожидаемый результат

• Основной артефакт папка проекта с файлами:
  • rag_cli.py – CLI-интерфейс.
  • rag_api.py – FastAPI сервер (опционально).
  • indexer.py – скрипт индексации.
  • corpus.json – исходный корпус.
  • chroma_db/ – persistent хранилище.
  • latency_report.csv – таблица с замерами.
  • README.md – описание, установка, примеры.
• Содержание результата программа, которая принимает вопрос на русском, выводит ответ и цитаты из документов.
• Дополнительно результаты нагрузочного теста, код оптимизации.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание При первом выполнении время может увеличиться в 1.5 раза из-за необходимости настройки окружения и отладки.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я проверил, что после запуска rag_cli.py и ввода тестового вопроса получаю осмысленный ответ за <500 мс.
• Я убедился, что использовал только CPU-модели и никакие данные не отправляются в облако.
• Я проверил, что при повторном запуске сервера или CLI индекс не перестраивается заново (persistent mode).
• Я измерил RAM потребление при индексации (не превышает 7.5GB) и при генерации (не более 6GB).
• Я подготовил README с примерами команд и требованиями к системе (Python 3.10, 8GB RAM, без GPU).