Как проектировать schema registry для метаданных RAG?

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

Schema registry — это централизованное хранилище схем данных с версионированием, которое обеспечивает согласованность метаданных в RAG-пайплайне. Проектирование включает выбор формата сериализации (Avro, JSON Schema, Protobuf), определение обязательных и опциональных полей, настройку правил эволюции схем (backward/forward/full совместимость) и интеграцию с этапом ingestion. Правильно спроектированный schema registry предотвращает рассинхронизацию между компонентами, упрощает отладку и позволяет безопасно обновлять структуру метаданных без остановки системы.

1. Термины: Schema Registry, Metadata, Avro

Schema registry — сервис, который хранит версии схем данных и предоставляет API для их регистрации, запроса и проверки совместимости. В контексте RAG он управляет структурой метаданных — дополнительной информации о документах (source_id, timestamp, doc_type, tags, author, embedding_version и т.д.).

Avro — один из популярных форматов сериализации данных с поддержкой схем и эволюции. Альтернативы: JSON Schema, Protocol Buffers (Protobuf). Avro часто используется в streaming-пайплайнах (Kafka) и хорошо подходит для RAG-ингредиентов благодаря компактности и строгой типизации.

2. Зачем schema registry в RAG-системе

RAG-система состоит из нескольких микросервисов: ingestion (разбивка, эмбеддинг, запись в векторную БД), retrieval, генерация ответа. Каждый сервис может независимо обрабатывать метаданные.

Проблемы без schema registry

• Разные компоненты могут интерпретировать поле timestamp как строку, число или объект — возникают ошибки сериализации.
• При добавлении нового поля (например, language) старые записи ломают логику валидации.
• Отсутствие версионирования — невозможно откатить схему, если новое поле оказалось некорректным.

Решение schema registry — единственный источник истины о структуре метаданных. Каждый компонент при чтении/записи получает актуальную схему из registry и проверяет свои данные на соответствие.

3. Ключевые требования к схеме метаданных

При проектировании схемы для RAG учтите:

Пример начального набора:

• Обязательные source_id (string), timestamp (long)
• Опциональные doc_type (string|null), tags (array of string), embedding_model (string|null)

4. Выбор формата сериализации

Рекомендация для RAG с Kafka — Avro, для REST-ориентированных систем — JSON Schema.

5. Проектирование схемы: пример на Avro

Зафиксируем версию v1:

{
  "type": "record",
  "name": "DocumentMetadata",
  "namespace": "com.myorg.rag",
  "fields": [
    {"name": "source_id", "type": "string"},
    {"name": "timestamp", "type": "long"},
    {"name": "doc_type", "type": ["string", "null"], "default": null},
    {"name": "tags", "type": {"type": "array", "items": "string"}, "default": []},
    {"name": "author", "type": ["string", "null"], "default": null}
  ]
}

Ключевые моменты

• Поле doc_type объявлено как union ["string", "null"] с default: null — это обеспечивает backward-совместимость при добавлении.
• tags — массив строк с пустым значением по умолчанию.
• namespace помогает избежать конфликтов в большой организации.

6. Версионирование и эволюция схемы

Schema registry хранит каждую версию схемы как неизменяемый артефакт. Эволюция схемы — это процесс создания новой версии на основе правил совместимости.

Типы эволюции

• Backward‑compatible — новая схема может читать данные, записанные старой схемой (разрешены: добавление поля с default, расширение типа union, удаление поля с default).
• Forward‑compatible — старая схема может читать данные, записанные новой схемой (разрешены: добавление поля, удаление поля, сужение типа union).
• Full compatibility — одновременно backward и forward.

Пример эволюции v1 → v2 В v2 решили добавить поле language (string с default "en") и удалить поле author (backward-совместимо, так как author был nullable с default null — удаление разрешено только если поле имеет default).

// v2
{
  "name": "DocumentMetadata",
  "fields": [
    {"name": "source_id", "type": "string"},
    {"name": "timestamp", "type": "long"},
    {"name": "doc_type", "type": ["string", "null"], "default": null},
    {"name": "tags", "type": {"type": "array", "items": "string"}, "default": []},
    {"name": "language", "type": "string", "default": "en"}
  ]
}

Проверка совместимости выполняется при регистрации v2: registry сравнивает v1 и v2 по правилу, заданному для субъекта (subject). Если несовместимость — регистрация отклоняется.

7. Проверка совместимости (backward / forward / full)

Backward‑compatibility (BC): Новый продюсер может писать данные, а старый консьюмер всё равно их прочитает.

• Если в v2 добавили поле language с default → OK.
• Если добавили поле без default → FAIL (старые данные не содержат это поле, невозможно заполнить).

Forward‑compatibility (FC): Старый продюсер пишет данные по старой схеме, новый консьюмер читает по новой схеме.

• Если в v2 удалили поле author → OK (консьюмер просто игнорирует отсутствующее поле, если оно было nullable с default).
• Если удалили поле без default → FAIL.

Full compatibility = BC + FC — самый строгий режим. Позволяет менять продюсеров и консьюмеров независимо без остановки.

8. Интеграция с ingestion pipeline

Ingestion (этап загрузки документов) — ключевое место для проверки метаданных. Обычно pipeline состоит из:

1. Parser — извлекает сырые метаданные (например, из PDF, HTML).
2. Schema validation — перед отправкой в векторную БД вызывается schema registry для проверки.
3. Producer (Kafka) — отправляет документ с метаданными в Avro‑формате, используя идентификатор схемы из registry.

Код на Python (используя confluent-kafka и avro):

from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer

# Загружаем схему из файла (или из registry по subject)
value_schema = avro.loads(""" ... """ )

producer = AvroProducer({
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
    }, default_value_schema=value_schema)

metadata = {
    "source_id": "doc-123",
    "timestamp": 1715000000,
    "doc_type": "report",
    "tags": ["finance", "Q1"],
    "language": "en"
}

# AvroProducer автоматически проверяет согласно последней версии схемы
producer.produce(topic='rag-documents', value=metadata)
producer.flush()

Ошибки валидации (например, поле language отсутствует, а схема требует его без default) — AvroProducer выбросит исключение, и запись не произойдёт.

9. Инструменты для schema registry

Выбор для on‑prem и гибкости — Confluent; для мультиформатных схем (REST + Avro) — Apicurio; если уже используете Redpanda — берите их.

10. Мониторинг и управление жизненным циклом

Schema registry — критичный компонент, поэтому:

• Мониторинг: отслеживайте количество регистраций/ошибок совместимости, latency ответов registry.
• TTL для устаревших схем можно удалять версии, к которым никто не обращается (но осторожно — лучше деактивировать, а не удалять).
• Backup экспортируйте все схемы в Git (например, раз в день) для возможности восстановления.
• CI/CD регистрируйте новые версии схем через пайплайн, а не вручную, с автоматическими тестами совместимости.

11. Преимущества и ограничения

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

• Гарантированная согласованность метаданных во всех сервисах.
• Безопасная эволюция схем без остановки системы.
• Упрощение онбординга новых команд (единый контракт).
• Интеграция с Kafka Streams, Flink и другими системами.

Ограничения

• Дополнительная инфраструктура (ZooKeeper/Redpanda, кластер registry).
• Overhead при записи каждого документа (один вызов к registry для получения ID схемы, кэшируется).
• Сложность при работе с динамическими метаданными (например, пользовательские теги, которые меняются без контроля).

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

Задача построить минимальный schema registry для RAG-системы, обрабатывающей внутренние документы компании. Метаданные: source_id, department (отдел), confidentiality (уровень доступа), tags, version (версия документа).

Инструменты

• Python + confluent-kafka (AvroProducer, AvroConsumer)
• Apache Kafka (local, docker-compose)
• Confluent Schema Registry (тоже docker-compose)
• Docker Compose

Шаги:

1. Docker Compose поднимает Kafka (1 брокер) + Schema Registry.
2. Создание схемы v1 (source_id, department, confidentiality, tags). Зарегистрируйте через REST API.
3. Продюсер (Python) — читает CSV с документами, валидирует по v1, шлёт в топик metadata.
4. Консюмер (Python) — подписывается, десериализует, сохраняет в JSON‑файл для анализа.
5. Эволюция схемы v2 — добавляем поле version (int, default 1), регистрируем с правилом BACKWARD.
6. Тест — запускаем продюсера, который использует v2, старый консюмер (с v1) должен без ошибок прочитать сообщение (backward compatibility).
7. Мониторинг — через /subjects/ endpoint смотрим историю схем.

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

• Консюмер успешно обрабатывает сообщения v1 и v2.
• В логах schema registry видны обе версии с правилами совместимости.
• При попытке зарегистрировать несовместимую схему (удаление поля без default) — ошибка 409 Conflict.

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

Навигация

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