ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Настроить Bloom filter для retrieval

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

Разработать и внедрить Bloom filter для фильтрации часто искомых запросов в RAG-системе, чтобы снизить нагрузку на векторную базу данных и ускорить ответ. Фильтр будет кэшировать "популярные" запросы (например, топ-1000 за последний час) и возвращать предварительно вычисленные результаты, минуя дорогой retrieval. Ключевой результат Ускорение среднего времени ответа на retrieval в 2 раза для повторяющихся запросов.

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

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

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

1. Создать синтетический набор запросов: 80% повторяющихся (топ-100 популярных), 20% уникальных.
2. Использовать pybloom-live для создания Bloom filter.
3. Написать простой HTTP-сервер на FastAPI, который имитирует retrieval (задержка 200-500ms) и кэш.

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

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

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

Действия

1. Установить зависимости:

   pip install pybloom-live fastapi uvicorn numpy pandas locust
   

2. Собрать или сгенерировать набор запросов:
   • Из логов: извлечь 1000 уникальных запросов за последний день.
   • Если логов нет: сгенерировать 200 запросов, где 80% — повторения из списка popular_queries = ["погода", "курс доллара", ...], 20% — случайные строки.
3. Сохранить в queries.csv:

   query,timestamp,is_popular
   погода,2025-01-01 10:00:00,1
   курс доллара,2025-01-01 10:01:00,1
   ...
   

Ожидаемый результат этапа Файл queries.csv с 200+ запросами, Python-окружение готово.

Этап 2: Реализация Bloom filter (1 час)

Действия

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

   from pybloom_live import ScalableBloomFilter
   from functools import lru_cache
   import time
   
   class BloomCache:
       def __init__(self, capacity=1000, error_rate=0.01):
           self.bloom = ScalableBloomFilter(
               initial_capacity=capacity,
               error_rate=error_rate
           )
           self.cache = {}
           self.hits = 0
           self.misses = 0
   
       def add_popular(self, query, result):
           self.bloom.add(query)
           self.cache[query] = result
   
       def get(self, query):
           if query in self.bloom:
               result = self.cache.get(query)
               if result is not None:
                   self.hits += 1
                   return result
           self.misses += 1
           return None
   
       def stats(self):
           return {
               "hits": self.hits,
               "misses": self.misses,
               "hit_rate": self.hits / (self.hits + self.misses + 1e-10)
           }
   

2. Настроить параметры:
   • capacity=1000 (топ-1000 популярных запросов).
   • error_rate=0.01 (1% ложных срабатываний).
3. Протестировать на синтетических данных:

   cache = BloomCache()
   for q in popular_queries:
       cache.add_popular(q, f"result_{q}")
   # Проверить hit/miss
   

Ожидаемый результат этапа Рабочий Bloom filter с hit rate > 80% на популярных запросах.

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

Действия

1. Создать FastAPI-сервер с двумя эндпоинтами:
   • POST /search — основной поиск с Bloom filter.
   • POST /admin/popular — добавление популярных запросов в кэш.
2. Реализовать логику:

   from fastapi import FastAPI
   import time
   
   app = FastAPI()
   bloom_cache = BloomCache()
   
   @app.post("/search")
   async def search(query: str):
       # 1. Проверить Bloom filter
       cached = bloom_cache.get(query)
       if cached:
           return {"result": cached, "source": "cache", "latency_ms": 1}
       
       # 2. Если нет — retrieval (симуляция)
       start = time.time()
       result = simulate_retrieval(query)  # задержка 200-500ms
       latency = (time.time() - start) * 1000
       
       # 3. Добавить в кэш, если запрос популярный (опционально)
       bloom_cache.add_popular(query, result)
       
       return {"result": result, "source": "retrieval", "latency_ms": latency}
   
   def simulate_retrieval(query):
       time.sleep(0.3)  # симуляция 300ms
       return f"retrieved_{query}"
   

3. Добавить эндпоинт для мониторинга:

   @app.get("/stats")
   async def stats():
       return bloom_cache.stats()
   

4. Запустить сервер: uvicorn main:app --reload

Ожидаемый результат этапа API работает, кэш возвращает результаты для популярных запросов за <10ms.

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

Действия

1. Написать скрипт для locust:

   from locust import HttpUser, task, between
   
   class CacheUser(HttpUser):
       wait_time = between(1, 3)
       popular_queries = ["погода", "курс доллара", "новости", ...]
       rare_queries = ["уникальный запрос 1", "уникальный запрос 2", ...]
   
       @task(8)  # 80% популярных
       def search_popular(self):
           query = self.popular_queries[0]
           self.client.post("/search", json={"query": query})
   
       @task(2)  # 20% редких
       def search_rare(self):
           query = self.rare_queries[0]
           self.client.post("/search", json={"query": query})
   

2. Запустить тест: locust -f locustfile.py --host=http://localhost:8000
3. Собрать метрики:
   • Средняя задержка (p50, p95) для популярных vs редких запросов.
   • Hit rate Bloom filter.
4. Оптимизировать:
   • Увеличить capacity до 2000, если hit rate < 80%.
   • Уменьшить error_rate до 0.005, если ложные срабатывания > 5%.

Ожидаемый результат этапа Ускорение в 2+ раза для популярных запросов (с 300ms до <10ms).

Этап 5: Мониторинг и деплой (30 минут)

Действия

1. Добавить Prometheus-метрики:

   from prometheus_client import Counter, Histogram, generate_latest
   import prometheus_client
   
   cache_hits = Counter('bloom_cache_hits_total', 'Total cache hits')
   cache_misses = Counter('bloom_cache_misses_total', 'Total cache misses')
   latency_histogram = Histogram('bloom_cache_latency_seconds', 'Latency of cache lookup')
   
   @app.get("/metrics")
   async def metrics():
       return generate_latest()
   

2. Настроить Grafana-дашборд (опционально):
   • Панель: hit rate (cache_hits / (cache_hits + cache_misses)).
   • Панель: latency p50/p95.
3. Деплой на production:
   • Обернуть в Docker.
   • Добавить health check эндпоинт /health.

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

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

• Bloom filter реализован и интегрирован с RAG-системой.
• Среднее время ответа для популярных запросов < 10ms.
• Ускорение в 2+ раза для 80% запросов (популярных).
• Hit rate Bloom filter > 80% (на тестовых данных).
• Ложные срабатывания < 5% (проверено на 1000 запросов).
• Нагрузочное тестирование показало стабильность при 100 RPS.
• Prometheus-метрики экспортируются и доступны.
• Код покрыт unit-тестами (минимум 3 теста).
• Документация (README) с описанием архитектуры и параметров.

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

Основной артефакт Python-модуль bloom_cache.py с классом BloomCache и FastAPI-сервером.

Содержание

• Реализация Bloom filter с настраиваемыми параметрами.
• API для поиска и администрирования кэша.
• Prometheus-метрики.
• Unit-тесты.

Дополнительно

• locustfile.py для нагрузочного тестирования.
• queries.csv с тестовыми данными.
• Дашборд Grafana (JSON-экспорт).

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

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

Примечание Для первого раза заложите +2 часа на отладку и настройку параметров.

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

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

• Я проверил, что Bloom filter корректно фильтрует популярные запросы (hit rate > 80%).
• Я убедился, что ускорение составляет 2x+ для повторяющихся запросов.
• Я протестировал систему под нагрузкой (100 RPS) и не получил ошибок.
• Я добавил Prometheus-метрики и проверил их в Grafana.
• Я написал unit-тесты для класса BloomCache (минимум 3 теста).
• Я задокументировал параметры (capacity, error_rate) в README.