ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать семантический кэш

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

Разработать семантический кэш для LLM-приложения, который на основе эмбеддингов запросов и порога косинусного сходства позволяет повторно использовать ранее сгенерированные ответы. Кэш хранит векторные представления запросов в Qdrant, а для быстрого доступа к метаданным использует Redis. В результате ожидается cache hit rate 30–50% на репрезентативном наборе запросов (не менее 1 000 уникальных). Ключевой результат рабочий семантический кэш с hit rate ≥ 30% при latency < 200 мс на поиск.

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

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

1. Скачать датасет вопросов (например, 2 000 строк из SQuAD) в CSV.
2. Создать скрипт generate_variations.py, который для каждого вопроса генерирует 3 синонимичных варианта с помощью nlpaug.ContextualWordEmbsAug(model='bert-base-uncased').
3. Установить Qdrant локально (docker-compose) или использовать in-memory Qdrant из qdrant-client (режим :memory:).
4. Redis заменить на fakeredis для тестов в памяти — это позволит не разворачивать сервисы.

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

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

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

Действия

1. Установить зависимости: pip install sentence-transformers qdrant-client fakeredis pandas nlpaug scikit-learn.
2. Скачать датасет (SQuAD) или сгенерировать синтетический набор пар (вопрос‑ответ).
3. Написать скрипт prepare_data.py, который:
   • загружает датасет в DataFrame;
   • генерирует для каждого вопроса 3 вариации (если нужна симуляция попаданий);
   • сохраняет train/test split (80/20) в data/queries.parquet.
4. Инициализировать in-memory Qdrant и fakeredis:

   from qdrant_client import QdrantClient
   from fakeredis import FakeStrictRedis
   
   qdrant = QdrantClient(":memory:")
   redis = FakeStrictRedis()
   

Ожидаемый результат этапа Настроенное окружение, файл data/queries.parquet с колонками query, response, variations, готовые клиенты Qdrant и Redis без развёртывания контейнеров.

Этап 2: Разработка модуля эмбеддингов (1 ч)

Действия

1. Создать embedder.py с классом Embedder:

   from sentence_transformers import SentenceTransformer
   
   class Embedder:
       def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
           self.model = SentenceTransformer(model_name)
       
       def encode(self, text: str) -> list[float]:
           return self.model.encode(text, normalize_embeddings=True).tolist()
   

2. Добавить кэширование запросов в Redis: перед вызовом модели проверять, есть ли хэш запроса (SHA256) в Redis. Если есть — возвращать закэшированный эмбеддинг, иначе вычислять и сохранять с TTL (1 час).
3. Протестировать скорость: один вызов encode не должен превышать 50 мс на CPU (опционально GPU).

Ожидаемый результат этапа Модуль embedder.py с методом encode, использующий Redis для кэширования эмбеддингов. Юнит-тест проверяет, что повторный вызов того же запроса не пересчитывает эмбеддинг.

Этап 3: Реализация семантического кэша (2 ч)

Действия

1. Создать semantic_cache.py с классом SemanticCache:

   class SemanticCache:
       def __init__(self, embedder, qdrant_client, redis, collection="cache", threshold=0.85):
           self.embedder = embedder
           self.qdrant = qdrant_client
           self.redis = redis
           self.collection = collection
           self.threshold = threshold
       def get(self, query: str) -> str | None:
           # 1. Получить эмбеддинг запроса
           # 2. Поиск в Qdrant по cosine distance (limit=1)
           # 3. Если distance >= threshold, вернуть ответ из payload точки
           # 4. Иначе None
       def put(self, query: str, response: str):
           # 1. Вычислить эмбеддинг
           # 2. Вставить точку в Qdrant: id = hash(query), vector, payload={query, response}
           # 3. (Опционально) сохранить метаданные в Redis (time, frequency)
   

2. Настроить Qdrant коллекцию:

   from qdrant_client import models
   
   qdrant.recreate_collection(
       collection_name="cache",
       vectors_config=models.VectorParams(
           size=384,  # all-MiniLM-L6-v2
           distance=models.Distance.COSINE
       )
   )
   

3. Реализовать get с учётом порога: если расстояние (косинусная дистанция = 1 - similarity) меньше 1 - threshold, считаем попаданием.
4. Добавить обработку ошибок: пустой запрос, сбой Qdrant, отсутствие коллекции.

Ожидаемый результат этапа Рабочий класс SemanticCache, который при put сохраняет запрос-ответ, а при get возвращает ответ на семантически похожий запрос. Unit-тест: после put запроса возвращается ответ на его вариацию с порогом 0.85.

Этап 4: Интеграция и тестирование (2 ч)

Действия

1. Написать скрипт test_cache.py, который:
   • загружает 1 000 запросов из датасета (только оригиналы) и сохраняет их в кэш;
   • затем прогоняет 1 000 вариаций (похожих, но не идентичных);
   • замеряет hit rate = (число попаданий) / 1 000.
2. Экспериментально подобрать порог threshold для достижения 30–50% hit rate (перебор от 0.7 до 0.95 с шагом 0.05).
3. Замерить среднюю и p99 latency для get и put.
4. Проверить, что кэш корректно работает при повторных вставках (обновление ответа, если запрос уже есть).

Ожидаемый результат этапа Таблица с hit rate для разных порогов, итоговый выбранный порог, метрики latency (get < 200 мс, put < 300 мс). График зависимости hit rate от порога (опционально).

Этап 5: Оптимизация и финальная упаковка (2 ч)

Действия

1. Оптимизировать Qdrant: добавить индексирование по payload-полям (например, query_text для fallback точного совпадения), использовать upsert для пакетной загрузки.
2. Реализовать fallback: если Qdrant возвращает точное текстовое совпадение — отдавать ответ без проверки порога.
3. Написать CLI-интерфейс cli.py с командами put, get, stats.
4. Оформить код в виде пакета semantic_cache/ с __init__.py.
5. Подготовить README.md: описание, пример использования, результаты экспериментов.

Ожидаемый результат этапа Готовый пакет, CLI, документация. Код проходит flake8 и pytest с покрытием > 70%.

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

• Реализован класс SemanticCache с методами get и put.
• Кэш использует Qdrant (in-memory или Docker) и Redis (или fakeredis).
• При одинаковом/похожем запросе с порогом 0.85 возвращается ответ из кэша.
• На тестовом наборе из 1 000 пар (оригинал + вариация) hit rate ≥ 30%.
• Среднее время get не превышает 200 мс без GPU.
• Код покрыт юнит-тестами (pytest) — минимум 5 тестов.
• Весь код опубликован в репозитории (GitHub) с файлами зависимостей (requirements.txt).
• Есть CLI или скрипт для воспроизведения теста hit rate.
• В README приведена таблица hit rate при разных порогах.
• Обработка крайних случаев: пустой запрос, отсутствие коллекции, ошибки Redis.

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

Основной файл semantic_cache/semantic_cache.py — реализация кэша.
Скрипт тестирования evaluate.py — замер hit rate, генерация отчёта.
Артефакты

• Таблица hit rate для порогов 0.70–0.95 (CSV или Markdown в results/).
• График зависимости (опционально, PNG).
• Юнит-тесты в tests/.
• requirements.txt.

Дополнительно Докер-композ для production-развёртывания (если использовался не in-memory).

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

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

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

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

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

• Я развернул Qdrant и Redis (или использовал in-memory) и проверил подключение.
• Мой Embedder загружает модель и выдаёт нормализованные векторы.
• Я реализовал get и put в SemanticCache с учётом порога сходства.
• Я прогнал тест на 1 000 вариаций и получил hit rate ≥ 30%.
• Я замерил среднее время ответа get — оно не превышает 200 мс.
• Я написал минимум 5 юнит-тестов (точное совпадение, семантическое, пустой запрос, ошибка Qdrant, ошибка Redis).
• Я оформил код согласно PEP8 и добавил type hints.
• В README я привёл пример использования и таблицу hit rate для разных порогов.
• Я убедился, что кэш корректно обрабатывает дубликаты (обновление ответа).
• Я проверил, что падение Redis не ломает get (использую try‑except и fallback на Qdrant только).