Как вы делаете data quality monitoring для RAG корпуса?

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

Quality Quality Quality|Data quality monitoring для RAG корпуса — это непрерывный процесс проверки документов на полноту, отсутствие дубликатов, корректность языка, длину, наличие PII и соответствие бизнес-правилам. Основные инструменты — Great Expectations (Python) и Deequ (Spark). При падении качества более чем на 5% за день система должна отправлять alert. Мониторинг позволяет предотвратить деградацию retrieval и генерации ответов.

1. Термин: Data Quality Monitoring (DQM) для RAG корпуса

Data Quality Monitoring — это набор автоматических проверок, которые выполняются на каждом этапе жизненного цикла данных: при индексации новых документов, при обновлении корпуса и периодически на всём датасете. В контексте RAG корпус]] — это коллекция чанков (текстовых фрагментов) с метаданными (source_id, дата, язык и т.д.). DQM гарантирует, что retrieval получает на вход только валидные, релевантные и безопасные данные.

Почему это критично для RAG

• Мусор на входе → мусор на выходе. Если в корпусе есть дубликаты, пустые документы или PII, LLM может выдать неверный или опасный ответ.
• Качество retrieval напрямую зависит от чистоты корпуса. Например, дубликаты искажают ранжирование, а слишком длинные чанки выходят за лимит контекста.
• Без мониторинга невозможно отследить деградацию данных со временем (например, устаревшие документы, изменение формата).

2. Зачем нужен DQM в RAG

DQM позволяет:

• Обнаруживать проблемы на ранней стадии (до того, как они повлияют на пользователей).
• Автоматически блокировать загрузку дефектных документов.
• Собирать метрики для анализа трендов качества.

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

3.1 Полнота (Completeness)

• Проверка отсутствие null в обязательных полях (text, source_id, chunk_id).
• Порог 0% null.

3.2 Уникальность (Uniqueness)

• Проверка дубликаты по хэшу содержимого (SHA256 текста) или по комбинации source_id + chunk_id.
• Порог 0% дубликатов.

3.3 Язык (Language)

• Проверка детекция языка через langdetect или fastText. Ожидаемый язык — русский, английский и т.д.
• Порог допустимо <1% документов на других языках.

3.4 Длина (Length)

• Проверка длина текста в символах (или токенах). Минимум 100 символов (не пустой), максимум 100k символов (не слишком большой).
• Порог 0% вне диапазона.

3.5 PII (Personally Identifiable Information)

• Проверка NER-модель (например, spaCy, HuggingFace) для поиска email, телефонов, паспортных данных.
• Порог 0% документов с PII (или блокировка таких документов).

3.6 Актуальность (Freshness)

• Проверка дата последнего обновления документа. Если документ старше N дней — помечается как устаревший.
• Порог доля устаревших документов не должна превышать 10% (настраивается).

3.7 Консистентность (Consistency)

• Проверка формат метаданных (например, дата в ISO 8601, source_id не пустой).
• Порог 0% нарушений.

4. Инструменты для DQM

4.1 Great Expectations (GX)

• Описание Open-source библиотека на Python для определения, валидации и документирования ожиданий (expectations) о данных.
• Особенности
  • Поддержка pandas, Spark, SQL.
  • Встроенные expectations (expect_column_values_to_not_be_null, expect_column_values_to_be_unique и т.д.).
  • Генерация HTML-отчётов с визуализацией.
  • Интеграция с Airflow, Prefect, DBT.

4.2 Deequ

• Описание Библиотека для Spark, разработанная Amazon. Позволяет вычислять метрики качества данных на больших объёмах.
• Особенности
  • Масштабируемость (Spark).
  • Встроенные проверки (completeness, uniqueness, compliance).
  • Constraint suggestion (автоматическое предложение правил).
  • Интеграция с AWS Glue, EMR.

4.3 Pandera

• Описание Библиотека для валидации pandas DataFrame'ов с помощью схем (schemas).
• Особенности
  • Статическая и динамическая валидация.
  • Поддержка типов, диапазонов, пользовательских функций.
  • Лёгкая интеграция в пайплайны.

4.4 Кастомные скрипты

• Для специфических проверок (PII, язык, хэширование) можно написать собственные функции на Python и запускать их как часть пайплайна.

5. Реализация проверок (пример с Great Expectations)

import great_expectations as gx
from great_expectations.dataset import PandasDataset
import pandas as pd

# Загружаем данные корпуса (пример)
df = pd.read_parquet("rag_corpus.parquet")

# Создаём датасет GX
ds = PandasDataset(df)

# Определяем ожидания (expectations)
expectations = [
    ds.expect_column_values_to_not_be_null("text"),
    ds.expect_column_values_to_not_be_null("source_id"),
    ds.expect_column_values_to_be_unique("chunk_id"),
    ds.expect_column_values_to_be_between("text_length", min_value=100, max_value=100_000),
    ds.expect_column_values_to_match_regex("language", r"^(ru|en)$"),
]

# Запускаем валидацию
results = ds.validate()

# Выводим результаты
for r in results.results:
    print(f"{r.expectation_config.expectation_type}: {'PASS' if r.success else 'FAIL'} (observed: {r.result['observed_value']})")

# Генерируем HTML-отчёт
ds.save_expectation_suite("rag_quality_suite.json")

Для PII и языка можно добавить пользовательские expectations:

from great_expectations.expectations.expectation import ColumnMapExpectation
import re
from langdetect import detect

class ExpectColumnValuesToNotContainPII(ColumnMapExpectation):
    map_metric = "column_values.not_contain_pii"
    success_keys = ("pii_patterns",)
    default_kwarg_values = {"pii_patterns": [r"\b\d{[[2 Как вы решаете проблему lost in the middle при работе с длинными контекстами|2]]}\.\d{[[2 Как вы решаете проблему lost in the middle при работе с длинными контекстами|2]]}\.\d{[[4. Какую векторную БД вы выберете для production-системы с больше 1 млн векторов|4]]}\b", r"\b[A-Za-z0-[[9. Как вы обновляете документы в существующей RAG-системе|9]]._%+-]+@[A-Za-z0-[[9. Как вы обновляете документы в существующей RAG-системе|9]].-]+\.[A-Z|a-z]{[[2 Как вы решаете проблему lost in the middle при работе с длинными контекстами|2]],}\b"]}

    def _validate(self, column, pii_patterns=None):
        def is_clean(value):
            return not any(re.search(p, str(value)) for p in pii_patterns)
        return column.map(is_clean)

# Аналогично для языка

6. Alerting и пороговые значения

• Порог если доля документов, не прошедших хотя бы одну проверку, превышает 5% за день → отправляется alert.
• Каналы Slack, email, PagerDuty, Telegram.
• Пример логики

def check_daily_quality(df, threshold=0.05):
    failed = 0
    total = len(df)
    # ... запуск проверок ...
    failure_rate = failed / total
    if failure_rate > threshold:
        send_alert(f"Quality dropped to {failure_rate:.2%} (threshold {threshold:.0%})")

• Действия при alert
  • Автоматическая блокировка новых документов.
  • Запуск пересчёта метрик.
  • Уведомление команды.

7. Интеграция в CI/CD пайплайн

DQM должен быть частью ETL-пайплайна индексации:

1. Pre-ingestion проверка сырых данных перед загрузкой в векторную БД.
2. Post-ingestion периодическая проверка всего корпуса (например, раз в сутки).
3. Continuous monitoring потоковая валидация при добавлении документов через API.

Пример с Airflow:

from airflow import DAG
from airflow.operators.python import PythonOperator

def validate_new_docs():
    # загрузить новые документы
    df = load_new_docs()
    # запустить Great Expectations
    suite = gx.data_context.load_expectation_suite("rag_quality_suite.json")
    results = suite.run(df)
    if not results["success"]:
        raise ValueError("Data quality check failed")

with DAG("rag_data_quality", schedule_interval="@daily") as dag:
    validate = PythonOperator(task_id="validate", python_callable=validate_new_docs)

8. Мониторинг дрейфа данных (Data Drift)

Со временем распределение данных в корпусе может меняться (например, добавляются документы на новом языке, меняется стиль). Это называется data drift. Для его обнаружения:

• Сравниваем статистики (средняя длина, частота слов, распределение языков) за разные периоды.
• Используем Population Stability Index (PSI) или Kullback-Leibler divergence.
• Если дрейф превышает порог → alert и пересмотр правил качества.

Также важен concept drift — изменение смысла запросов пользователей, но это уже относится к мониторингу retrieval, а не корпуса.

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

Задача Разработать систему мониторинга качества для корпуса из 10 000 новостных статей.

Инструменты Python, Great Expectations, Airflow, PostgreSQL (для хранения метрик), Slack webhook.

Шаги:

1. Собрать датасет новостей (например, из RSS-лент).
2. Разбить на чанки по 512 токенов, добавить метаданные (source, date, language).
3. Написать expectations: no null, уникальность chunk_id, длина 100–100k символов, язык только ru/en, отсутствие PII (email, телефоны).
4. Запустить валидацию через Great Expectations, сохранить отчёт.
5. Настроить ежедневный DAG в Airflow, который проверяет новые чанки.
6. Если доля брака >5% — отправить alert в Slack.
7. Визуализировать тренды качества в Grafana (метрики из PostgreSQL).

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

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

11. Навигация

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

Навигация

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