Как обрабатывать late-arriving data в ingestion?

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

Late-arriving data — это документы или события, которые поступают в пайплайн индексации с задержкой относительно их собственного временно́го штампа (event time). В контексте RAG-систем такая задержка приводит к неконсистентности индекса: устаревшие версии чанков, нарушение хронологического порядка, конфликты при обновлении. Для корректной обработки применяют три основные стратегии: Ignore (пропустить), Reprocess (переобработать с обновлением метаданных) и Window + Watermark (принять в пределах допустимого окна, иначе отбросить). Выбор стратегии зависит от критичности задержки, SLA и допустимой сложности инфраструктуры.

1. Термин: Late-arriving data (запаздывающие данные)

Late-arriving data — это записи (документы, логи, события), которые достигают системы обработки спустя значительное время после того момента, который они описывают (event time). Типичный пример: документ, созданный в 11:00, но доставленный в пайплайн индексации только в 13:00 из-за сбоя продюсера или сетевой задержки.

В пайплайнах streaming ingestion различают два временных понятия:

• Event time — время, когда документ был создан или изменение в источнике зафиксировано.
• Processing time — время, когда документ фактически обрабатывается на стороне индексатора.

Разница между ними и есть величина опоздания.

2. Причины возникновения late-arriving data в RAG-пайплайне

3. Влияние на качество векторного индекса

Late-arriving данные нарушают предположение о линейной упорядоченности изменений. Основные проблемы:

• Устаревшие чанки — новая версия документа приходит после того, как старая уже проиндексирована и, возможно, использована для ответов пользователям.
• Конфликт версий — если документ обрабатывается в параллельных потоках, возможна «гонка» между поздним и обычным сообщением.
• Нарушение хронологии — в сценариях, где важен порядок (например, лог транзакций), опоздавшие сообщения могут привести к неверному контексту.

4. Стратегия 1: Ignore (пропустить)

Идея все документы, пришедшие с задержкой больше допустимого порога, молча отбрасываются.

# Пример на Python (псевдокод)
LATENCY_THRESHOLD = timedelta(hours=1)

def should_ingest(document):
    delay = datetime.utcnow() - document.event_time
    if delay > LATENCY_THRESHOLD:
        log.warning(f"Dropping late document {document.id}, delay={delay}")
        return False
    return True

Плюсы минимальные накладные расходы, простота реализации.
Минусы потеря данных — потенциально критичная для задач, где требуется полная история (финансовые транзакции, логи аудита).
Когда применять некритичные данные (например, кэши общих фактов, где свежесть не важна) или когда источник подтверждает, что опоздавшие сообщения дублируют уже обработанные.

5. Стратегия 2: Reprocess (переобработать с обновлением метаданных)

Идея поздний документ всё же обрабатывается, но при этом обновляются метаданные существующего чанка (например, перезаписывается вектор и обновляется last_updated). Если более новая версия уже есть — конфликт разрешается через версионирование (например, MVCC — multi-version concurrency control).

# Схема на Redis + FAISS
def upsert_chunk(chunk_id, new_vector, new_metadata):
    current_version = redis.get(f"version:{chunk_id}") or 0
    if new_metadata.event_time > current_time - WINDOW:
        # Обновляем только если новое событие «свежее» (по версии, не по времени)
        if new_metadata.version > current_version:
            index.update(chunk_id, new_vector)
            redis.set(f"version:{chunk_id}", new_metadata.version)

Плюсы сохраняется консистентность — индекс всегда отражает последнюю версию документа.
Минусы увеличивается нагрузка на запись (особенно при частых обновлениях).
Когда применять системы, где важна полная актуальность, но допустима операция перезаписи (чат-боты с документацией, FAQ).

6. Стратегия 3: Window + Watermark (окно и водяной знак)

Идея определяется временно́е окно (например, 5 минут), в пределах которого поздние данные принимаются. Данные с задержкой меньше окна обрабатываются в нормальном порядке; данные с задержкой больше окна — либо отбрасываются, либо отправляются в side output (отдельный канал) для дальнейшего анализа.

Ключевой механизм — watermark (водяной знак). Это временной порог, который гарантирует, что все события с event time <= watermark уже поступили. В Apache Flink он задаётся так:

// Flink Java
DataStream<Document> stream = env
    .fromSource(kafkaSource, 
        WatermarkStrategy
            .<Document>forBoundedOutOfOrderness(Duration.ofMinutes(5))
            .withTimestampAssigner((doc, ts) -> doc.getEventTime().toEpochMilli()),
        "Kafka Source");

Сценарий документ с event_time = 1:00 приходит в 2:00. Если водяной знак уже пересёк 1:05, то документ считается опоздавшим и отправляется в side output.

# Идея на Python с Flink-подобным API (синтетический пример)
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class Doc:
    id: int
    text: str
    event_time: datetime

def generate_side_output(doc, watermark):
    if doc.event_time < watermark - timedelta(minutes=5):
        # в side output
        pass
    else:
        # нормальная обработка
        pass

Плюсы гибкость — можно настроить допустимую степень опоздания; side output позволяет не терять данные, а лишь отложить их анализ.
Минусы сложнее реализация (требуется stream processing engine); при очень высокой задержке всё равно приходится решать, что делать с side output.
Когда применять потоковые пайплайны с реальными временными ограничениями (например, индексация логов или событий IoT).

7. Техническая реализация водяного знака в среде Python

Для небольших проектов можно эмулировать watermarks на уровне потребителя Kafka с confluent-kafka-python:

from confluent_kafka import Consumer, TopicPartition
from datetime import datetime, timezone
import time

consumer = Consumer(...)
consumer.assign([TopicPartition('documents', 0)])

WATERMARK_DELAY = 300  # 5 minutes in seconds
watermark = None

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    event_time = datetime.fromtimestamp(msg.timestamp()[1], tz=timezone.utc)
    now = datetime.now(timezone.utc)
    delay = (now - event_time).total_seconds()
    
    # Обновление водяного знака (минимальное время среди обработанных сообщений + DELAY)
    if watermark is None or event_time.timestamp() < watermark:
        watermark = event_time.timestamp() + WATERMARK_DELAY
    
    if delay > WATERMARK_DELAY:
        print(f"LATE: {msg.value()} (delay={delay}s) -> side output")
        continue
    # нормальная индексация
    process_document(msg.value())

8. Связь с метаданными векторов

При обработке late-arriving данных в RAG важно хранить реальный event_time в метаданных каждого чанка. Это позволяет:

• при обновлении версий сравнивать не только version, но и временную метку;
• при построении контекста для LLM отфильтровывать документы, которые «старее» момента запроса пользователя;
• при агрегации по времени (аналитика корректности) знать истинный временной ряд.

Пример схемы метаданных в pgvector или Qdrant:

{
  "chunk_id": "456",
  "document_id": "doc_123",
  "event_time": "2025-03-24T11:00:00Z",
  "ingestion_time": "2025-03-24T13:15:00Z",
  "version": 3,
  "late_arrival_processed": true
}

9. Интеграция с агентной архитектурой Agentic RAG

В Agentic RAG агент может принимать решение о том, как обрабатывать каждый поздний документ динамически:

• Агент-планировщик оценивает критичность опоздания: если документ от авторитетного источника и его нет в индексе — агент инициирует переиндексацию даже с высокой задержкой.
• Агент-аудитор периодически просматривает side output и решает, какие документы всё же стоит добавить (ручное подтверждение).
• Агент-версионировщик поддерживает историю версий и генерирует diff-обновления, чтобы не перезаписывать векторы целиком.

Пример агентного правила на LangGraph:

class LateArrivalDecision(BaseModel):
    action: Literal["ignore", "reprocess", "side_output"]
    reason: str

def decide_on_late_document(doc):
    # агент анализирует метаданные и текущее состояние
    if doc.source == "critical":
        return LateArrivalDecision(action="reprocess", reason="critical source")
    elif doc.age_in_hours > 24:
        return LateArrivalDecision(action="ignore", reason="too old")
    else:
        return LateArrivalDecision(action="side_output", reason="moderate delay")

10. Сравнительная таблица стратегий

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

Задача Разработать мини-пайплайн индексации документов с поддержкой late-arriving данных. Реализовать две стратегии: Reprocess (через обновление в векторной БД) и Window+Watermark (через очередь с приоритетом по event_time).

Инструменты Python, FAISS (in-memory), Apache Kafka (через confluent-kafka-python), SQLite для метаданных. Для водяного знака — эмуляция без Flink.

Шаги:

1. Настроить продюсера, который отправляет документы с искусственной задержкой (random delay от 1 до 30 минут).
2. На стороне консьюмера реализовать выбор стратегии через конфигурацию.
3. Для стратегии Reprocess: при получении документа с id, который уже есть в индексе, обновлять вектор и метаданные (версия, last_updated).
4. Для стратегии Watermark: определить окно 5 минут, для опоздавших документов печатать WARN в лог, но не индексировать.
5. Сравнить hit rate индекса до и после внедрения обработки late-arriving данных (для тестового датасета).

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

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

Навигация

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