Как проектировать data contracts для RAG пайплайна?

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

Data contract — это формальное соглашение между producer (ингредиенты) и consumer (retrieval) в RAG пайплайне. Он определяет схему данных, обязательные поля, SLO (latency, freshness) и версионирование. Проектирование включает выбор формата схемы (JSON Schema, Avro, Protobuf), настройку CI-валидации и инструментов мониторинга (Great Expectations, dbt). Правильные data contracts обеспечивают надёжность, предсказуемость и упрощают эволюцию пайплайна.

1. Термин: Data Contract (контракт данных)

Data contract — это формальное описание ожиданий от данных, передаваемых между компонентами пайплайна. В контексте RAG речь идёт о взаимодействии между этапом индексации (producer) и этапом retrieval (consumer). В отличие от простой схемы, контракт включает не только структуру полей, но и:

• SLO (Service Level Objectives) — целевые показатели качества (например, максимальная задержка индексации).
• Обязательства по качеству — требования к полноте, свежести и согласованности.
• Версионирование — механизм управления изменениями схемы.
• Владельцы — кто отвечает за producer и consumer.

Термины producer и consumer заимствованы из событийно-ориентированной архитектуры: producer генерирует данные (чанки), consumer их потребляет (строит индекс, ищет).

2. Зачем Data Contract в RAG?

• Предотвращение "тихих" изменений Если индексатор внезапно перестанет передавать поле embedding, retrieval-сервис упадёт. Контракт делает такие изменения видимыми.
• Согласованность между командами В больших проектах индексацию и поиск могут разрабатывать разные группы. Контракт служит границей ответственности.
• Упрощение отладки Падение метрик retrieval (например, recall@k) часто связано с изменением данных. Контракт позволяет быстро проверить, изменилась ли схема или SLO.
• Эволюция без слома Версионирование контракта позволяет вводить новые поля или менять типы, не ломая существующий код, если соблюдать backward compatibility.

3. Ключевые элементы Data Contract

4. Проектирование схемы: выбор формата

Выбор формата схемы зависит от требований к производительности и экосистемы.

• JSON Schema: человекочитаемая, легко валидируется, подходит для REST API и streaming (например, через Kafka с JSON-сериализацией). Минус: большой размер данных (текстовые поля, эмбеддинги в base64).
• Apache Avro: компактный бинарный формат, встроенная поддержка эволюции схемы (forward/backward compatibility). Часто используется с Kafka и Hadoop.
• Protocol Buffers (Protobuf): строгая типизация, кодогенерация на множество языков, минимальный размер. Идеален для микросервисов и gRPC. Не поддерживает forward compatibility без дополнительных настроек.

Для RAG-пайплайна с чанками и эмбеддингами (обычно массив float32) Protobuf — хороший выбор: эффективность и типобезопасность.

Пример Protobuf схемы:

syntax = "proto3";

message Chunk {
    string source_id = 1;           // ID документа в источнике
    int32 chunk_index = 2;          // номер чанка в документе
    string content = 3;             // текст чанка
    int64 timestamp = 4;            // unix timestamp индексации
    bytes embedding = 5;            // float32 массив (сериализованный)
    map<string, string> metadata = 6; // опциональные метаданные
    int32 version = 7;              // версия документа (для дедупликации)
}

Обязательные поля: source_id, content, timestamp. embedding может быть обязательным или опциональным в зависимости от того, где вычисляются эмбеддинги.

5. SLO и мониторинг

SLO — это целевые показатели, которые producer обязуется соблюдать. Для RAG важно:

• Latency: время от загрузки документа до появления чанка в индексе. Типичное значение: < 5 секунд.
• Freshness: максимальное время, через которое новые/изменённые документы становятся доступны для поиска. Обычно < 1 минуты.
• Accuracy: доля чанков, прошедших валидацию схемы и не содержащих ошибок (например, отсутствующих обязательных полей). Цель: 100% или 99.9%.

Мониторинг SLO реализуется через:

• Prometheus + Grafana — метрики latency и freshness.
• Счётчики отбракованных чанков — для accuracy.

Пример кода для метрики latency (Python с prometheus_client):

from prometheus_client import Histogram, Counter
import time

index_latency = Histogram('rag_chunk_index_latency_seconds', 'Time to index a chunk')
failed_chunks = Counter('rag_failed_chunks_total', 'Number of chunks that failed validation')

def index_chunk(chunk):
    start = time.time()
    try:
        validate(chunk)  # валидация по контракту
        store_to_index(chunk)
    except ValidationError:
        failed_chunks.inc()
    finally:
        index_latency.observe(time.time() - start)

6. Версионирование и управление изменениями

Версионирование контракта следует принципам Semantic Versioning (major.minor.patch):

• Breaking changes (major): удаление поля, изменение типа существующего поля, добавление нового обязательного поля. Требуют согласования и новой версии контракта.
• Non-breaking changes (minor): добавление опционального поля, расширение enum, изменение описания. Могут быть введены без явной смены версии, но лучше фиксировать.
• Patch: исправление опечаток, уточнение документации.

При изменении контракта producer должен:

1. Опубликовать новую версию в Schema Registry (Confluent Schema Registry, Apicurio).
2. Запустить CI-проверку backward compatibility.
3. Уведомить consumer команду.

Consumer, в свою очередь, должен поддерживать предыдущую версию некоторое время (grace period).

Backward compatibility означает, что данные, созданные со старой версией схемы, могут быть прочитаны новым consumer (обычно это про добавление полей со значениями по умолчанию). Forward compatibility — что данные, созданные с новой версией, могут быть прочитаны старым consumer (сложнее, требует игнорирования неизвестных полей).

7. CI/CD валидация контракта

В CI-пайплайне при изменении схемы должны выполняться следующие шаги:

1. Генерация кода из Protobuf/Avro (например, protoc или avrogen).
2. Компиляция и юнит-тесты consumer с новой схемой.
3. Проверка совместимости с предыдущей версией (backward compatibility).
   • Для Protobuf: утилита buf или ручное сравнение.
   • Для Avro: avro-tools compatibility check.
4. Запуск интеграционных тестов с реальным producer.

Пример .github/workflows/contract-check.yml:

name: Contract Check
on:
  pull_request:
    paths:
      - 'schemas/**'

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: bufbuild/buf-action@v1
        with:
          version: latest
          input: 'proto/'
          breaking: true
      - run: python -m pytest tests/

8. Инструменты для реализации

Great Expectations можно интегрировать в пайплайн индексации для мониторинга качества чанков:

import great_expectations as ge

def validate_chunk(chunk_dict):
    df = ge.dataset.PandasDataset([chunk_dict])
    df.expect_column_values_to_not_be_null('source_id')
    df.expect_column_values_to_be_in_set('chunk_index', list(range(1000)))
    df.expect_column_pair_values_to_be_equal('source_id', 'source_id')  # не меняется
    results = df.validate()
    if not results["success"]:
        raise ValueError(f"Validation failed: {results['statistics']['unexpected_percent']}")

9. Пример полного Data Contract для RAG

Документ контракта (например, в Confluence или YAML-файле):

contract_version: 1.0
name: ChunkContract
producer: Ingestion Team
consumer: Search Team

schema:
  format: protobuf
  file: proto/chunk.proto
  version: 1

obligatory_fields:
  - source_id
  - content
  - timestamp
  - embedding

slo:
  latency_p99_seconds: 5
  freshness_seconds: 60
  accuracy_percent: 99.9

versioning:
  strategy: semantic
  breaking_change_policy: "new major version required, 1 month grace period"
  compatibility: backward

monitoring:
  metrics_endpoint: https://metrics.internal/producer
  dashboards: ["RAG Ingestion SLO"]

10. Проблемы и best practices

• Проблема: Producer меняет схему без уведомления.
  • Решение: Запретить деплой без прохождения CI-проверки контракта; обязательный code review изменений схемы.
• Проблема: Слишком много обязательных полей.
  • Решение: Оставить обязательными только критически важные (source_id, content, timestamp), остальные — опциональные с reasonable default.
• Проблема: SLO не достигаются, контракт теряет доверие.
  • Решение: Начинать с консервативных значений (latency < 10 sec), затем ужесточать; мониторинг error budget.
• Best practice: Документировать контракт в читаемом виде с примерами (JSON-сниппет).
• Best practice: Использовать code-first подход: из Protobuf генерировать не только классы, но и документацию (proto-doc).

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

Задача: Создать простой RAG пайплайн с data contract между модулем chunking и векторным поиском.

Инструменты: Python, Protobuf, FastAPI, FAISS (in-memory), Great Expectations, GitHub Actions.

Шаги:

1. Определить Protobuf схему Chunk (source_id, content, timestamp, embedding).
2. Сгенерировать Python классы с помощью protoc.
3. Написать producer (FastAPI endpoint POST /chunks), который принимает JSON, валидирует и публикует Protobuf-сообщение в Redis Stream (или просто сохраняет в FAISS).
4. Написать consumer (скрипт), читающий из Redis, десериализующий Protobuf, вычисляющий эмбеддинг (если не пришёл) и добавляющий в FAISS.
5. Добавить Great Expectations для проверки полей перед сохранением.
6. Настроить GitHub Actions: при PR с изменениями схемы запускать buf breaking и прогонять тесты consumer.

Ожидаемый результат: Рабочий пайплайн, где любое изменение схемы (breaking) блокирует merge, а non-breaking изменения проходят автоматически. Дополнительно — графана дашборд с latency и accuracy.

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

13. Навигация

Навигация

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