Как вы проектируете data lineage для RAG (от документа к ответу)?

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

Data lineage (происхождение данных) для RAG — это сквозная прослеживаемость пути от исходных документов через этапы чанкинга, эмбеддинга, retrieval до финального ответа LLM. Такая система позволяет в любой момент ответить на вопрос: «из каких конкретно фрагментов каких документов был сгенерирован данный ответ». Это критично для аудита, отладки неверных ответов, обеспечения соответствия регуляторным требованиям и непрерывного улучшения пайплайна. Рекомендуемое решение — использование стандарта OpenLineage в связке с визуализатором Marquez, который автоматически регистрирует события на каждом шаге и строит граф зависимостей]].

1. Термин: Data lineage (происхождение данных)

Data lineage (также provenance) — это метаданные, описывающие жизненный цикл данных: откуда они появились, какие преобразования прошли, куда были переданы. В контексте RAG lineage фиксирует:

• Какие документы были загружены в базу знаний.
• Как они были разбиты на чанки (чunks).
• Какие эмбеддинги (векторные представления) были созданы.
• Какой запрос пользователя поступил.
• Какие чанки были извлечены (retrieved) и с какими релевантностями.
• Какой контекст попал в промпт LLM.
• Какой ответ был сгенерирован.

Зачем это нужно в RAG

• Аудит и compliance: если ответ содержит неверную информацию, можно отследить, из какого документа она взялась.
• Отладка: понять, почему ответ плохой — из-за плохого retrieval, неверного чанка или ошибки LLM.
• Улучшение системы: анализируя lineage, можно оптимизировать стратегию чанкинга, эмбеддинги или механизм ранжирования.
• Объяснимость: для пользователя или заказчика можно показать, на основе каких источников сформирован ответ.

2. Компоненты RAG pipeline, подлежащие трекингу

Полный жизненный цикл данных в RAG состоит из нескольких этапов. Для каждого этапа нужно фиксировать уникальные идентификаторы, временные метки и входные/выходные данные.

Ключевой принцип каждый артефакт (чанк, эмбеддинг, ответ) должен иметь глобально уникальный идентификатор (UUID), чтобы можно было построить цепочку.

3. OpenLineage — стандарт для сбора lineage

OpenLineage — это открытый стандарт и набор библиотек для сбора метаданных о передаче данных между компонентами. Он определяет модель:

• Job — процесс, преобразующий данные (например, chunking_job, retrieval_job).
• Dataset — набор данных на входе или выходе job (например, document, chunks, responses).
• Run — конкретное выполнение job с временным интервалом и статусом.

В контексте RAG мы можем описывать каждый компонент как job, а чанки, документы, ответы — как datasets.

OpenLineage предоставляет REST API для отправки событий, а также клиентские библиотеки на Python, Java, Spark, Airflow и др.

Пример события для этапа retrieval:

from openlineage.client import OpenLineageClient
from openlineage.client.run import RunEvent, RunState, Run, Job
from openlineage.client.event import EventType

client = OpenLineageClient(url="http://marquez:5000")

event = RunEvent(
    eventType=EventType.COMPLETE,
    eventTime="2025-03-04T12:00:00Z",
    run=Run(runId="retrieval-run-uuid"),
    job=Job(namespace="rag-pipeline", name="retrieve"),
    inputs=[{"namespace": "rag", "name": "user_query"}],
    outputs=[{"namespace": "rag", "name": "retrieved_chunks", "facets": {
        "chunk_ids": ["chunk1", "chunk2", "chunk3"],
        "scores": [0.92, 0.85, 0.74]
    }}]
)
client.emit(event)

4. Marquez — визуализация lineage

Marquez — это open-source платформа для сбора, визуализации и управления data lineage. Она принимает события от OpenLineage, сохраняет их в БД (PostgreSQL) и строит интерактивный граф зависимостей.

Основные возможности:

• Просмотр lineage конкретного датасета (например, «из каких документов произошёл этот ответ»).
• Фильтрация по времени, namespace, job.
• API для программного доступа.
• Интеграция с Airflow, dbt, Spark и др.

Для RAG мы поднимаем Marquez вместе с PostgreSQL и OpenLineage-клиент внутри каждого компонента пайплайна. После выполнения запроса lineage автоматически регистрируется, и в интерфейсе можно кликнуть на ответ и увидеть цепочку:

Документ A → Чанк A1 → (эмбеддинг) → Retrieval → Context → Ответ

5. Практическая реализация data lineage для RAG

Проектируем микросервисную или модульную архитектуру, где каждый этап (ingestion, chunking, retrieval, generation) представляет собой отдельный job в OpenLineage.

Шаги:

1. Установить Marquez (через Docker Compose) с PostgreSQL.
2. Добавить клиент OpenLineage в каждый модуль RAG.
3. На этапе загрузки документа создаём dataset для исходного файла, генерируем id документа.
4. На этапе чанкинга job chunking принимает на вход id документа, создаёт набор чанков с их собственными id (например, {doc_id}_chunk_N). Каждый чанк регистрируется как dataset или фацет.
5. На этапе эмбеддинга job embedding принимает на вход чанки, создаёт векторы (можно ссылаться на чанки как на вход, но не обязательно регистрировать каждый вектор как отдельный dataset — достаточно отметить, что чанк стал доступен для поиска).
6. На этапе retrieval job retrieve принимает запрос, возвращает список id чанков с метаданными (score). Выходной dataset retrieved_chunks содержит эти id.
7. На этапе генерации job generate принимает контекст (набор чанков) и запрос, выдаёт ответ. Выходной dataset llm_response может дополнительно содержать ссылку на ответ.

Пример кода для этапа генерации (fastapi):

from openlineage.client import OpenLineageClient, RunEvent, Run, Job, Dataset
from uuid import uuid4

client = OpenLineageClient(url="http://marquez:5000")

def log_generation(query: str, chunks: list[str], response: str, run_id: str):
    event = RunEvent(
        eventType="COMPLETE",
        eventTime=datetime.utcnow().isoformat() + "Z",
        run=Run(runId=run_id),
        job=Job(namespace="rag", name="llm_generation"),
        inputs=[
            Dataset(namespace="rag", name="user_query"),
            Dataset(namespace="rag", name="retrieved_chunks", facets={"chunks": chunks})
        ],
        outputs=[
            Dataset(namespace="rag", name="llm_response", facets={"response_text": response})
        ]
    )
    client.emit(event)

Важно не стоит логировать полное содержимое каждого чанка или ответа — достаточно идентификаторов и кратких метаданных. Полные данные хранятся в базе знаний отдельно.

6. Альтернативы OpenLineage + Marquez

Рекомендация для RAG-системы до среднего масштаба OpenLineage + Marquez — оптимальный выбор: лёгкий, с открытым кодом, простой интеграцией.

7. Best practices при проектировании data lineage для RAG

• Уникальные id на каждом уровне: документы, чанки, ответы — глобально уникальные UUID.
• Версионирование: при обновлении документа создавайте новую версию, чанки с новыми id. В lineage указывайте, какая версия использовалась.
• Стандартизация метаданных: используйте фацеты OpenLineage для хранения дополнительных атрибутов (score, модель, параметры).
• Минимизация размера событий: не пихайте в событие весь текст чанка — только id и hash.
• Асинхронная отправка: отправляйте события через background tasks или message queue, чтобы не замедлять пайплайн.
• Совместимость с observability: объедините lineage с трейсингом (OpenTelemetry) для полной картины.
• Соглашение об именовании: namespace rag или rag-{project}, имена job — retrieve, generate и т.д.

8. Вызовы и их решения

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

Задача построить мини-RAG на Python с отслеживанием data lineage через OpenLineage + Marquez, визуализировать путь от загрузки документа до ответа.

Инструменты Python 3.10+, FastAPI, SentenceTransformers, ChromaDB, OpenLineage Python SDK, Docker Compose.

Шаги:

1. Поднять Marquez + PostgreSQL через Docker Compose (официальный репозиторий).
2. Создать RAG-пайплайн:
   • ingestion.py: загружает текст из файла, chunking (фиксированный размер 200 символов, overlap 20), сохраняет в ChromaDB.
   • retrieval.py: принимает запрос, вызывает ChromaDB similarity search, возвращает id чанков и scores.
   • generation.py: формирует промпт из конкатенации чанков и запроса, вызывает LLM (можно заглушку, возвращающую "Ответ на основе чанков: ...").
3. В каждый шаг добавить логирование OpenLineage: для ingestion — job document_ingestion, для retrieval — retrieve, для generation — llm_generation.
4. Создать API-эндпоинт /ask, который принимает вопрос и возвращает ответ, плюс в ответе добавляет ссылку на lineage uuid.
5. Посмотреть в UI Marquez граф: щёлкнуть на ответ, увидеть предков.

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

• После первого запроса в Marquez появится граф из трёх нодов (document_ingestion → retrieve → llm_generation).
• Кликнув на llm_generation, можно увидеть входные датасеты: user_query, retrieved_chunks (с id чанков).
• Кликнув на retrieved_chunks, увидеть document_ingestion с id исходного документа.

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

Навигация

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