Как вы делаете incremental ingestion для часто меняющихся документов?

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

ingestion|Incremental ingestion (ingestion|инкрементальная загрузка) для часто меняющихся документов строится на CDC (Change Data Capture) — захвате изменений из источника (база данных, файловое хранилище, очередь событий). Изменения передаются через Kafka (или аналогичную шину), consumer обрабатывает каждое событие: удаляет старые чанки документа, создаёт новые, пересчитывает эмбеддинги и атомарно обновляет векторную БД. Критичны транзакционность, версионирование и latency|низкая задержка (цель — <5 секунд от изменения до доступности в поиске).

1. Термин: Incremental ingestion (инкрементальная загрузка)

Что это Процесс обновления поискового индекса (векторной БД) только для изменившихся документов, вместо полной переиндексации всего корпуса.

Зачем нужна

• Скорость полная переиндексация 10⁶ документов может занимать часы, инкрементальная — секунды.
• Ресурсы экономия вычислительных мощностей (CPU/GPU для эмбеддингов, IO для БД).
• Актуальность пользователь получает свежие данные без задержек на полный ребилд.

Противоположность batch ingestion (ingestion|пакетная загрузка) — периодический полный пересчёт всех документов (например, раз в сутки). Для часто меняющихся данных (финансовые котировки, новости, логи) batch неприемлем.

2. CDC (Change Data Capture) — источник изменений

CDC — механизм, который отслеживает изменения в источнике данных (INSERT, UPDATE, DELETE) и публикует их как события.

Типы источников и CDC-инструменты

Пример конфигурации Debezium для PostgreSQL

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.dbname": "inventory",
    "database.server.name": "dbserver1",
    "table.include.list": "public.documents",
    "plugin.name": "pgoutput",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

Ключевое CDC даёт логи репликации — поток изменений в реальном времени.

3. Pipeline: Kafka как шина событий

Kafka — распределённая платформа потоковой передачи событий. В контексте incremental ingestion выполняет роль буфера и транспортного слоя.

Архитектура

Источник (БД) → Debezium → Kafka Topic (documents-changes) → Consumer (Ingestion Worker) → Векторная БД

Преимущества Kafka

• At-least-once delivery — гарантия, что каждое изменение будет обработано (с возможными дубликатами, которые нужно идемпотентно обрабатывать).
• Партиционирование — можно распределить нагрузку по ключу документа (source_id).
• Retention — хранить историю изменений для отладки и повторной обработки.

Consumer (на Python с использованием aiokafka):

import asyncio
from aiokafka import AIOKafkaConsumer
from vector_db import VectorDB  # условный клиент
from chunker import chunk_document
from embedder import embed_chunks

async def process_change(event):
    source_id = event['after']['id']
    new_text = event['after']['content']
    operation = event['op']  # 'c' (create), 'u' (update), 'd' (delete)

    if operation == 'd':
        # Удалить все чанки документа
        await VectorDB.delete_chunks_by_source_id(source_id)
        return

    # Для 'c' и 'u': удалить старые чанки, создать новые
    await VectorDB.delete_chunks_by_source_id(source_id)
    chunks = chunk_document(new_text)
    embeddings = await embed_chunks(chunks)
    await VectorDB.insert_chunks(source_id, chunks, embeddings)

async def main():
    consumer = AIOKafkaConsumer(
        'documents-changes',
        bootstrap_servers='kafka:9092',
        group_id='ingestion-group',
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    await consumer.start()
    try:
        async for msg in consumer:
            await process_change(msg.value)
    finally:
        await consumer.stop()

4. Обработка изменений: удаление старых чанков и создание новых

Проблема при обновлении документа старые чанки становятся неактуальными. Если их не удалить, retrieval может вернуть устаревшую информацию.

Алгоритм для операции UPDATE

1. Получить source_id (уникальный идентификатор документа).
2. Удалить все чанки с этим source_id из векторной БД.
3. Разбить новый текст на чанки (используя ту же стратегию chunking, что и при первичной индексации).
4. Сгенерировать эмбеддинги для каждого чанка.
5. Вставить новые чанки с тем же source_id и новым version (см. раздел 6).

Важно операция должна быть атомарной — либо все чанки обновлены, либо ни один. Иначе возможна ситуация, когда часть старых чанков удалена, а новые ещё не вставлены → временная потеря данных.

5. Транзакционность (атомарность обновления)

Транзакционность — гарантия, что группа операций (удаление + вставка) выполняется как единое целое: либо все успешно, либо откат к исходному состоянию.

Подходы

Рекомендация для большинства RAG-систем достаточно идентификации по source_id + версионирование. Если векторная БД поддерживает транзакции (например, Qdrant, Weaviate), используйте их. Если нет — применяйте idempotent consumer с повторной обработкой.

Пример транзакции в Qdrant (Python):

from qdrant_client import QdrantClient
from qdrant_client.http import models

client = QdrantClient("localhost", port=6333)

# Начинаем транзакцию (Qdrant поддерживает точку сохранения)
with client.transaction():
    # Удаляем старые точки по фильтру source_id
    client.delete(
        collection_name="documents",
        points_selector=models.Filter(
            must=[models.FieldCondition(key="source_id", match=models.MatchValue(value=source_id))]
        )
    )
    # Вставляем новые точки
    client.upsert(
        collection_name="documents",
        points=[models.PointStruct(id=chunk_id, vector=embedding, payload=payload) for ...]
    )

6. Versioning (версионирование) и TTL

Versioning — хранение метаданных о версии документа (timestamp, номер версии) в payload каждого чанка.

Зачем

• Retrieval только актуальных при поиске можно фильтровать по version <= current_version или timestamp < TTL.
• Откат при ошибке можно восстановить предыдущую версию.
• Аудит отслеживание истории изменений.

Реализация

• В payload чанка добавляем поля: source_id, version, updated_at.
• При retrieval добавляем фильтр: updated_at >= now - TTL или version = latest_version.
• TTL (Time-To-Live) — максимальное время, в течение которого чанк считается актуальным. Если документ не обновлялся дольше TTL, он исключается из поиска (или помечается как устаревший).

Пример payload

{
  "source_id": "doc_123",
  "version": 5,
  "updated_at": "2025-03-15T10:30:00Z",
  "chunk_index": 2
}

Фильтр при retrieval (Qdrant):

client.search(
    collection_name="documents",
    query_vector=query_emb,
    query_filter=models.Filter(
        must=[
            models.FieldCondition(key="updated_at", range=models.Range(gte=datetime.utcnow() - timedelta(days=7)))
        ]
    )
)

7. Latency (задержка) — цель <5 секунд

Latency — время от момента изменения в источнике до момента, когда новый чанк становится доступен для поиска.

Компоненты задержки

1. CDC capture (Debezium → Kafka): обычно <100 мс.
2. Kafka propagation (topic → consumer): <10 мс.
3. Consumer processing (chunking + embedding + DB write): основная часть.

Как достичь <5 секунд

• Параллельная обработка несколько consumer'ов в группе, каждый обрабатывает свой partition.
• Асинхронный I/O использовать asyncio (как в примере выше).
• Кэширование эмбеддингов если документ меняется незначительно, можно переиспользовать эмбеддинги старых чанков (но осторожно — риск устаревания).
• Batch-вставка накапливать несколько чанков и вставлять пачкой.
• Выбор быстрой векторной БД Qdrant (Rust) или Milvus (C++) быстрее, чем Pinecone (облачный) для write-heavy сценариев.

Мониторинг: метрики (latency p50/p99, throughput, error rate) через Prometheus + Grafana.

8. Проблемы и решения

9. Инструменты и альтернативы

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

Задача Реализовать инкрементальную загрузку для блога на Django (PostgreSQL). При создании/редактировании/удалении поста — обновлять векторный индекс (Qdrant) в реальном времени.

Инструменты

• PostgreSQL + Debezium (через Docker Compose)
• Kafka (один брокер)
• Python consumer (aiokafka, sentence-transformers, qdrant-client)
• Qdrant (векторная БД)
• Streamlit (UI для демонстрации)

Шаги:

1. Развернуть PostgreSQL, Kafka, Debezium, Qdrant через Docker Compose.
2. Создать таблицу posts (id, title, content, updated_at).
3. Настроить Debezium connector для таблицы posts.
4. Написать consumer, который:
   • Получает событие из Kafka.
   • Для INSERT/UPDATE: удаляет старые чанки по source_id=post.id, разбивает title + content на чанки (по 512 токенов), генерирует эмбеддинги (all-MiniLM-L6-v2), вставляет в Qdrant.
   • Для DELETE: удаляет все чанки по source_id.
5. Добавить версионирование: хранить updated_at в payload, при retrieval фильтровать по updated_at > now - 1 day.
6. Написать Streamlit-приложение, которое показывает текущие посты и позволяет искать по ним (через Qdrant).

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

• При добавлении поста через Django admin — через <2 секунды он становится доступен для поиска.
• При редактировании — старые чанки заменяются новыми.
• При удалении — чанки исчезают из поиска.

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

12. Навигация

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

Навигация

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