Как организовать streaming feature pipelines для real-time RAG?

Краткий тезис

Streaming feature pipelines — это системы, которые непрерывно вычисляют признаки (features) на основе потоков событий (например, действий пользователя) и делают их доступными для RAG-запросов с задержкой менее 100 мс. Они сочетают источники вроде Kafka, процессоры вроде Apache Flink или Spark Structured Streaming и feature stores (Redis, Parquet). Такие пайплайны позволяют обогащать контекст поиска реальными пользовательскими метриками, повышая релевантность и персонализацию ответов.

1. Термин: feature pipeline в контексте real-time RAG

Feature pipeline — это набор этапов, которые преобразуют сырые данные (логи, клики) в структурированные признаки, используемые моделью или поисковым движком. В real-time RAG признаки могут влиять на:

• переранжирование результатов поиска;
• взвешивание чанков по релевантности;
• динамическое добавление контекстных подсказок.

Отличие от batch-пайплайнов: данные обрабатываются с нулевой задержкой (или минимальной), а не раз в день/час. Для этого нужны stream processing и online feature store.

2. Архитектура streaming feature pipeline

2.1 Источники данных

2.2 Stream processor

Популярные фреймворки:

• Apache Flink — лидер по производительности и возможностям (event-time, exactly-once, windowing).
• Spark Structured Streaming — удобен, если в компании уже есть Spark для batch-обработки.
• Kafka Streams — легковесный, встраиваемый в микросервисы.

Выбор зависит от:

• языка (Java/Scala vs Python);
• требований к задержке;
• сложности вычислений (например, скользящее окно по сложным ключам).

2.3 Feature store

Роль feature store:

• Унификация: один и тот же признак вычисляется в real-time и batch, доступен в online.
• Point-in-time correctness: гарантия, что признак соответствует времени запроса (avoiding data leakage).

2.4 Выгрузка и обратная связь

• Online features → записываются в Redis с TTL (например, 1 час).
• Offline features → партиционируются по дате и сохраняются в Parquet.
• Alerting: пайплайн может отправлять метрики в Prometheus/Grafana.

3. Feature computation: скользящие окна и агрегации

3.1 Виды окон

3.2 Типичные агрегации

• Среднее, сумма, количество для числовых признаков.
• Скользящее среднее (exponential moving average) — сглаживание шума.
• Embedding aggregation — усреднение эмбеддингов запросов пользователя за окно (см. пример ниже).

3.3 Пример: user_30min_embeddings

Этот feature агрегирует эмбеддинги последних N запросов пользователя за последние 30 минут.

# Псевдокод на Flink SQL
CREATE TABLE user_query_embeddings (
    user_id STRING,
    query_embedding ARRAY<FLOAT>,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '1' MINUTE
);

-- Скользящее окно: 30 минут, обновление каждые 1 минуту
INSERT INTO redis_features
SELECT
    user_id,
    AVG(F.col) AS avg_embedding,
    COUNT(*) AS query_count,
    WINDOW_END as window_end
FROM TABLE(
    HOP(TABLE user_query_embeddings, DESCRIPTOR(event_time), INTERVAL '1' MINUTE, INTERVAL '30' MINUTES))
GROUP BY user_id, window_start, window_end;

При поиске RAG читает из Redis:

# FastAPI endpoint
user_feature = redis_client.get(f"user:{user_id}:avg_embedding")
if user_feature:
    query_embedding = embed(user_query)
    combined_embedding = average(query_embedding, user_feature)
    results = vector_store.similarity_search(combined_embedding, k=10)

Зачем персонализация — если пользователь искал "футболки", следующий запрос "зимняя одежда" может смешивать контекст.

4. Обеспечение низкой латентности (< 100 ms)

Чтобы feature был доступен к моменту запроса пользователя, нужно соблюсти цепочку:

Событие → Kafka → Flink → Redis → RAG‑запрос

Ключевые практики:

• Минимизация сериализации: Avro/Protobuf вместо JSON.
• Асинхронное чтение из Redis (пишется в том же процессе или через sidecar).
• Locality: Flink и Redis на одном кластере или в одной зоне доступности.
• Batching: микропакетная обработка в Spark (микро-батч 100 мс) или event-at-a-time в Flink.
• TTL для признаков: хранить только актуальные данные (например, окно 30 минут → TTL = 40 минут).

5. Consistency и fault tolerance

5.1 Exactly-once semantics

• Flink: checkpointing + Kafka's transactional producer.
• Spark: Structured Streaming с WAL (Write-Ahead Log).

5.2 Watermark и event-time

• Event-time — время, когда событие произошло на клиенте.
• Watermark — задержка, позволяющая обрабатывать опоздавшие события.
• Для real-time RAG достаточно watermark 1-5 минут, так как пользователи могут ждать ответ не более 2 секунд.

5.3 Обработка сбоев

• State backend: RocksDB (Flink) или HDFS (Spark).
• Grace period для окон: если событие пришло после окончания окна, его можно игнорировать или обработать в специальном потоке.

6. Offline pipeline для обучения

Параллельно с online-вычислениями те же данные сохраняются в Parquet:

# Spark Structured Streaming: sink data to Parquet partitioned by date
query = df \
    .writeStream \
    .format("parquet") \
    .option("path", "s3://features/offline/") \
    .option("checkpointLocation", "s3://checkpoints/") \
    .partitionBy("event_date") \
    .trigger(processingTime="5 minutes") \
    .start()

Offline данные используются для:

• тренировки моделей ранжирования;
• ретроспективного анализа (почему retrieval показал низкий HR);
• A/B-тестирования новых окон.

7. Интеграция с query-time RAG

Фичи загружаются в момент запроса тремя способами:

1. Pre-join: на этапе retrieval ключ пользователя (user_id) используется для фильтрации кандидатов с учетом веса, вычисленного из features.
2. Post-merge: после получения top-k чанков, features используются как дополнительный сигнал для reranker (XGBoost или небольшой LLM).
3. Feature as context: усреднённый эмбеддинг истории запросов конкатенируется с текущим запросом.

Пример архитектуры:

Client → API Gateway → RAG Orchestrator
                          ├─> Redis (online features) ──> Retrieval (weighted)
                          ├─> LLM (with context)
                          └─> Response

8. Мониторинг и тестирование

8.1 Метрики пайплайна

8.2 Тестирование

• Backfill: запуск пайплайна на исторических данных для проверки корректности вычислений.
• Shadow mode: параллельный выключенный пайплайн, сверяющий результаты с реальным.
• Chaos engineering: симуляция сбоев Kafka, Redis, рестарта Flink-джобы.

Пет-проект для закрепления

Задача: Реализовать простую streaming feature pipeline для real-time RAG, который подсказывает следующий товар на основе последних кликов пользователя.

Инструменты:

• Kafka (локально через Docker)
• PyFlink или Kafka Streams (Python-бинальный поток)
• Redis (docker)
• Vector store (FAISS или Chroma)
• LLM (GPT-4 или Mistral через API)

Шаги:

1. Сгенерировать синтетический поток событий: {user_id, product_id, timestamp}.
2. Написать Flink-задачу, которая в скользящем окне 5 минут считает топ-5 product_id по количеству кликов.
3. Записать результат в Redis: ключ user:{id}:top_products, значение — список product_id.
4. Настроить RAG-эндпоинт: при запросе "похожие товары" читать из Redis топ продукта, получать их эмбеддинги, искать в векторной БД ближайшие.
5. Оценить latency (от клика до появления в ответе).

Ожидаемый результат: система, которая в реальном времени подсказывает товары с учётом последних действий, latency < 200 мс (включая LLM).

Связь с другими вопросами

Навигация

• Предыдущий: 858
• Следующий: 860
• Индекс: 00. Индекс разборов