ТЕХНИЧЕСКОЕ ЗАДАНИЕ: RAG с кэшированием ответов

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

Спроектировать и реализовать слой кэширования для RAG-системы с использованием Redis. Кэш должен хранить ответы на повторяющиеся запросы пользователей с TTL 1 час, чтобы снизить latency для повторных запросов на 80% по сравнению с полным RAG-циклом (retrieval + генерация). В процессе вы отработаете стратегию инвалидации кэша, выбор ключа и измерение производительности.

Ключевой результат Рабочий прототип RAG-системы с Redis-кэшем, в котором для повторных запросов время ответа снижено не менее чем на 80%.

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

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

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

1. Установите Redis через Docker: docker run --name redis-rag -p 6379:6379 -d redis:7
2. Напишите простой RAG-пайплайн на LangChain: FAISS-индекс + OpenAI API (или локальная модель через Ollama)
3. Создайте 20–30 уникальных запросов и 50 их копий (повторов) для измерения cache-hit.

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

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

Этап 1: Подготовка и настройка Redis (30 минут)

Действия

1. Установите Redis (Docker или локально). Проверьте подключение через redis-cli ping → PONG.
2. Установите Python-клиент: pip install redis.
3. Создайте файл cache.py с классом RedisCache:

   import redis
   import json
   
   class RedisCache:
       def __init__(self, host='localhost', port=6379, db=0, ttl=3600):
           self.client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
           self.ttl = ttl
   
       def get(self, key: str) -> dict | None:
           data = self.client.get(key)
           if data:
               return json.loads(data)
           return None
   
       def set(self, key: str, value: dict) -> None:
           self.client.setex(key, self.ttl, json.dumps(value))
   
       def flush(self):
           self.client.flushdb()
   

4. Напишите тест подключения:

   cache = RedisCache()
   cache.set("hello", {"answer": "world"})
   print(cache.get("hello"))   # {'answer': 'world'}
   

Ожидаемый результат этапа Рабочий класс для чтения/записи в Redis с TTL 1 час.

Этап 2: Интеграция кэша в RAG-пайплайн (1.5 часа)

Действия

1. Имея базовый RAG-пайплайн (создайте его, если нет), оберните вызов генерации в функцию get_rag_answer(question: str) -> dict.
2. Модифицируйте функцию: перед retrieval проверьте наличие ключа в Redis.
   Определите стратегию ключа используйте нормализованный (lowercase, strip) и хэшированный (MD5) текст вопроса для компактности.

   import hashlib
   
   def make_cache_key(question: str) -> str:
       normalized = question.strip().lower()
       return hashlib.md5(normalized.encode()).hexdigest()
   

3. Реализуйте двухуровневую логику:

   def get_answer(question: str) -> dict:
       key = make_cache_key(question)
       cached = cache.get(key)
       if cached:
           return {"source": "cache", "answer": cached["answer"], "latency": 0}
       
       # Полный RAG-процесс
       start = time.time()
       docs = retriever.get_relevant_documents(question)
       context = "\n".join([d.page_content for d in docs])
       prompt = f"Context: {context}\nQuestion: {question}\nAnswer:"
       answer = llm.invoke(prompt)
       elapsed = time.time() - start
       
       # Сохранить в кэш
       cache.set(key, {"answer": answer, "context": context})
       return {"source": "rag", "answer": answer, "latency": elapsed}
   

4. Протестируйте на одном и том же вопросе дважды — первый ответ должен приходить с source=rag, второй с source=cache.
5. Добавьте обработку ошибок Redis (если Redis недоступен, система должна работать в режиме fallback без кэша).

Ожидаемый результат этапа Функция get_answer корректно возвращает кэшированные ответы и измеряет latency.

Этап 3: Нагрузочное тестирование и замеры (1 час)

Действия

1. Подготовьте тестовый набор: 10 уникальных вопросов, каждый повторить 5 раз (всего 50 запросов). Сохраните в test_queries.py.
2. Напишите скрипт для прогона всех запросов последовательно и сбора статистики по latency. Разделите на две фазы:
   • Фаза 1 (cold): первый прогон всех уникальных вопросов (все попадают в RAG).
   • Фаза 2 (hot): повторный прогон тех же вопросов (все должны быть из кэша).
3. Посчитайте среднюю latency для каждой фазы.
4. Вычислите процент снижения: (cold_latency - hot_latency) / cold_latency * 100.
5. Подтвердите, что снижение >= 80%.
6. Проверьте TTL: подождите 1 час (или для теста установите TTL=10 секунд) и убедитесь, что запросы снова выполняются через RAG.

Ожидаемый результат этапа Таблица с latency для cold/hot и итоговый процент ускорения.

Этап 4: Оптимизация и документирование (30 минут)

Действия

1. Если ускорение меньше 80%, проанализируйте причины: возможно, кэш-ключ не захватывает синонимы, или Redis overhead высок. Попробуйте использовать pipeline для массового чтения.
2. Добавьте инвалидацию кэша по событию (например, при обновлении базы знаний) — симулируйте очистку кэша через flush при загрузке новых документов.
3. Напишите краткую документацию в README: архитектура, как запустить, результаты тестов.
4. Подготовьте скрипт для воспроизведения метрик (run_benchmark.py).

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

Этап 5: Интеграция мониторинга (опционально, 30 минут)

Действия

1. Добавьте логирование каждого запроса: время, source (cache/rag), latency.
2. Экспортируйте метрики в Prometheus-формат (через prometheus_client) или просто в CSV.
3. Нарисуйте график latency по времени (например, в Excel или Python matplotlib).

Ожидаемый результат этапа Дашборд или график, наглядно показывающий снижение latency для повторных запросов.

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

• Redis-сервер запущен, Python-клиент успешно подключается.
• Каждый запрос сначала проверяется в кэше; при наличии ответ возвращается из кэша.
• TTL для каждой записи установлен ровно 1 час (3600 секунд).
• После истечения TTL запрос повторно попадает в RAG.
• Средняя latency для повторных запросов снижена на 80% относительно первого запроса.
• При падении Redis система падает корректно (fallback на полный RAG без кэша).
• Кэш-ключ строится на основе нормализованного текста вопроса (без учёта регистра и лишних пробелов).
• Нагрузочный тест выполнен, результаты зафиксированы в таблице.
• Код размещён в репозитории с README и комментариями.
• В README описаны метрики (cold latency vs hot latency) и инструкция по запуску.

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

Основной артефакт

• Репозиторий с кодом (Python-модули rag_cache.py, cache.py, test_queries.py, run_benchmark.py).

Содержание

• Реализация кэширующего RAG с использованием Redis.
• Скрипт нагрузочного тестирования, выводящий cold_latency, hot_latency, speedup_percent.
• README с инструкцией и результатами теста.

Дополнительные результаты

• График/таблица сравнения latency (опционально).
• Описание стратегии инвалидации кэша.

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

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

Примечание Для первого раза закладывайте +1 час на отладку интеграции с RAG и Redis.

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

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

• Я проверил, что Redis запущен и отвечает на ping.
• Я убедился, что кэш-ключи детерминированы (один и тот же вопрос → один ключ).
• Я протестировал TTL: через 1 час (или тестовый интервал) запрос снова прошёл RAG.
• Я замерил latency для 10+ запросов в cold и hot режиме и получил ускорение >= 80%.
• Я обработал случай, когда Redis падает, и система не ломается.
• Я написал README, в котором описаны шаги запуска и результаты.