ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Semantic cache для RAG

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

Спроектировать и реализовать слой семантического кэширования для RAG-системы на основе векторного поиска. Кэш будет сохранять ответы на часто задаваемые или семантически близкие вопросы, используя Redis для быстрого доступа и Qdrant для сравнения эмбеддингов запросов с порогом семантической близости 0.9. В результате система должна вернуть кэшированный ответ для ≥30% пользовательских запросов, ускоряя обслуживание и снижая нагрузку на LLM.

Ключевой результат Рабочий semantic cache, интегрированный в RAG-пайплайн, с измеренным cache hit rate ≥30% на тестовом наборе запросов.

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

Перед началом необходимо иметь:

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

1. Если нет готового RAG — создаём минимальный: FAISS + ChromaDB (заменить на Qdrant) + локальная LLM (через Ollama или transformers) + эмбеддер sentence-transformers/all-MiniLM-L6-v2.
2. Если нет набора вопросов — генерируем 150 вопросов с помощью LLM по теме документации или wiki (например, «Что такое …?», «Как настроить …?»).
3. Если нет эталонных ответов — для первых 50 вопросов вручную фиксируем ответы, остальные используем только для измерения hit rate (сравниваем по семантике).

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

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

Этап 1: Проектирование архитектуры и схемы данных (30 минут)

Действия

1. Нарисовать схему потоков (текстовое описание):
   • Пользователь отправляет запрос.
   • Система получает эмбеддинг запроса.
   • Выполняется поиск в Qdrant: query_vector → search(limit=1, score_threshold=0.9).
   • Если найдён похожий запрос (score ≥ 0.9) → взять ответ из Redis по ключу cache:<id>.
   • Если нет → выполнить RAG, сохранить ответ в Redis и записать эмбеддинг в Qdrant.
2. Спроектировать коллекции в Qdrant
   • Коллекция query_cache с полями:
     • id (integer, auto-increment)
     • query_text (string)
     • answer (string)
     • embedding (vector, 384 dims)
     • created_at (datetime)
3. Спроектировать структуру в Redis
   • Ключ: cache:{qdrant_id} → value: answer (string)
   • Возможно, также хранить timestamp для TTL.
4. Определить параметры конфигурации
   • SEMANTIC_THRESHOLD = 0.9
   • CACHE_TTL = 86400 (сутки)
   • QDRANT_COLLECTION_NAME = "query_cache"

Ожидаемый результат этапа Документ с архитектурой (или код инициализации коллекции Qdrant и Redis) готов.

Этап 2: Реализация модуля semantic cache (2–3 часа)

Действия

1. Создать класс SemanticCache

   import numpy as np
   from qdrant_client import QdrantClient
   from qdrant_client.http import models
   from redis import Redis
   from sentence_transformers import SentenceTransformer
   
   class SemanticCache:
       def __init__(self, qdrant_url, redis_url, threshold=0.9):
           self.qdrant = QdrantClient(url=qdrant_url)
           self.redis = Redis.from_url(redis_url)
           self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
           self.threshold = threshold
           self.collection_name = "query_cache"
   

2. Реализовать методы
   • _get_embedding(text: str) → np.ndarray
   • lookup(query: str) -> str | None: поиск в Qdrant, если найден → достать ответ из Redis.
   • store(query: str, answer: str): добавить эмбеддинг в Qdrant и ответ в Redis.
   • _ensure_collection(): создать коллекцию, если не существует.
3. Написать unit-тесты (pytest) для методов lookup (пустой кэш, точное совпадение, семантически близкий запрос, miss).
4. Протестировать порог убедиться, что при threshold=0.9 синонимичные запросы (например, «Как открыть файл» и «Как запустить документ») попадают.

Ожидаемый результат этапа Модуль SemanticCache с протестированными базовыми сценариями.

Этап 3: Интеграция с RAG-пайплайном (1–2 часа)

Действия

1. Создать обёртку RAGWithCache

   class RAGWithCache:
       def __init__(self, rag_pipeline, cache: SemanticCache):
           self.rag = rag_pipeline
           self.cache = cache
   
       def answer(self, query: str) -> str:
           cached = self.cache.lookup(query)
           if cached:
               return cached
           answer = self.rag.generate(query)
           self.cache.store(query, answer)
           return answer
   

2. Встроить в существующий эндпоинт FastAPI (или консольный интерфейс):

   @app.post("/ask")
   async def ask(query: str):
       answer = rag_cache.answer(query)
       return {"query": query, "answer": answer, "from_cache": ...}
   

   Для отслеживания hit/miss добавить счётчик (можно через threading или redis incr).
3. Добавить логирование каждый запрос писать в лог с полем cache_hit: True/False.
4. Запустить интеграционный тест проверить, что повторный запрос с тем же смыслом попадает в кэш.

Ожидаемый результат этапа RAG-система работает с кэшем; в логах видно hit/miss.

Этап 4: Тестирование и настройка threshold (1–2 часа)

Действия

1. Подготовить тестовый набор
   • 150 запросов, из которых 50 — точные повторы, 50 — семантические парафразы, 50 — новые (не похожие).
2. Заполнить кэш первыми 50 запросами (прогнать через систему, чтобы сохранить).
3. Прогнать все 150 запросов, замерить:
   • Hit rate: количество попаданий / общее количество запросов.
   • Miss rate.
   • Средняя задержка для hit vs miss.
4. Поэкспериментировать с threshold
   • 0.8 → слишком много ложных попаданий (low precision).
   • 0.95 → слишком много misses.
   • Выбрать компромисс с hit rate ≥30% и accuracy (качество ответов) не ниже baseline.
5. Проверить качество кэширования для попаданий сравнить ответ из кэша с эталонным (если есть). Использовать semantic similarity (например, sentence-transformers) — разница не более 0.85.

Ожидаемый результат этапа Протокол экспериментов с таблицей метрик для разных threshold, выбран итоговый threshold.

Этап 5: Оптимизация и наблюдение (1 час)

Действия

1. Добавить TTL в Redis — автоматическое удаление старых записей.
2. Реализовать механизм инвалидации (например, при обновлении баз знаний удалить соответствующие записи по тегу).
3. Настроить метрики счётчики hits/misses, задержка lookup, размер кэша.
4. Написать пару графиков в Grafana (если есть время) или просто вывести статистику в консоль.
5. Провести небольшую нагрузочную проверку (locust, 10 concurrent users) — убедиться, что cache не увеличивает latency более чем на 50ms.

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

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

• Semantic cache реализован на основе Qdrant + Redis.
• Порог семантической близости настраивается через переменную окружения.
• Cache hit rate на тестовом наборе из 150 запросов ≥30% (при threshold=0.9).
• Среднее время ответа при cache hit ≤ 200ms (с учётом сети до Qdrant/Redis).
• Ответы из кэша семантически эквивалентны эталонным (cosine similarity ≥0.95).
• Кэш не вызывает ошибок на новых запросах (продолжает выполнять RAG).
• Написаны unit-тесты (минимум 3 для lookup: точное совпадение, семантическое совпадение, miss).
• Интеграция с RAG-пайплайном работает без сбоев.
• Есть документация (1–2 абзаца) по запуску и настройке threshold.
• Кэш корректно обрабатывает TTL (старые записи удаляются через 24 часа).

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

• Основной артефакт Репозиторий с кодом, содержащий:
  • semantic_cache.py — реализация класса SemanticCache.
  • rag_with_cache.py — интеграция с RAG.
  • tests/ — pytest-тесты.
  • docker-compose.yml — для Qdrant и Redis.
  • config.py — параметры (threshold, TTL).
  • README.md с инструкцией по запуску.
• Дополнительно протокол экспериментов (CSV/таблица) с hit rate при разных threshold.
• Опционально скрипт для генерации тестовых запросов, дашборд Grafana с метриками.

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

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

Примечание: для первого раза заложите 10–12 часов с учётом отладки и экспериментов.

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

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

• Я написал модуль SemanticCache с методами lookup и store.
• Я настроил Docker Compose для Qdrant и Redis.
• Я написал тесты для lookup: точное совпадение, парафраз, miss.
• Я интегрировал кэш в свой RAG-пайплайн (вызываю lookup до RAG).
• Я измерил cache hit rate на тестовом наборе и получил ≥30%.
• Я проверил, что время ответа при hit < 200 мс.
• Я настроил TTL для Redis и проверил, что старые записи удаляются.
• Я задокументировал в README, как запустить и изменить threshold.
• Я провёл хотя бы один эксперимент с другим threshold и записал результат.
• Весь код закоммичен в Git с понятными commit messages.