Заставьте свой RAG не использовать устаревшие документы: отслеживание актуальности и старения

Предположим, что система RAG уже парсит PDF-файлы, разбивает их на фрагменты, встраивает их, извлекает контекст и даже сохраняет память о разговоре между вопросами. Однако внутри неё может скрываться более тонкая проблема. Векторный индекс относится ко всем фрагментам одинаково. Руководство по прошивке 2022 года и бюллетень с исправлениями от прошлого вторника имеют одинаковую вероятность быть найденными, если оба семантически соответствуют запросу. Агент не имеет представления о том, что документ был заменён или что новая версия должна иметь приоритет.

На практике это приводит к реальному ущербу: устаревшие процедуры соблюдения норм, устаревшие ссылки на API, старые шаги диагностики, которые больше не применимы. Агент с полной уверенностью предоставляет устаревшую информацию, поскольку ничто не указывает ему на обратное.

Чтобы исправить это, требуются три элемента, все из которых могут быть включены в пайплайн Databricks с использованием Vector Search и Delta Sync.

Три проблемы на трёх уровнях

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

1. Прием изменений. Новые и отредактированные документы должны попадать в индекс без необходимости полной пересборки.
2. Устаревание данных. Каждый фрагмент требует записи его текущего состояния — актуального, устаревшего или замененного, чтобы система могла их различать.
3. Ранжирование. Когда два фрагмента имеют примерно одинаковую значимость, новый должен иметь приоритет.

Проблема 1: внесение нового контента

Благодаря инструменту Databricks Vector Search в составе Delta Sync уже существует инфраструктура для постепенных обновлений. Delta Sync следует за потоком изменений исходной таблицы и добавляет в индекс новые данные, обновления и удаления.

Основной выбор касается типа обработки. В режиме триггеров вы сами определяете момент синхронизации:

# Manual control — you decide when to sync
vsc.create_delta_sync_index(
    ...
    pipeline_type="TRIGGERED",
)

# After appending new chunks:
index = vsc.get_index(index_name="<YOUR_INDEX>")
index.sync()  # Only processes CDF deltas, not full re-index

В режиме непрерывной работы индекс обновляется сам в течение нескольких минут после изменений в таблице:

# Automatic — syncs within minutes of table changes
vsc.create_delta_sync_index(
    ...
    pipeline_type="CONTINUOUS",
)
# No manual sync() calls needed

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

Процесс загрузки новых или изменённых документов выглядит следующим образом:

from pyspark.sql import functions as F
from pyspark.sql.functions import expr

def ingest_new_documents(volume_path, chunked_table, parsed_table, doc_version="2.0"):
    """Parse new PDFs, chunk them, append to the chunks table.
    Existing chunks from updated docs are soft-deleted first."""

    # Read new files - binaryFile gives us modificationTime
    new_docs_df = spark.read.format("binaryFile").load(volume_path)

    # Check which files already exist in the table
    existing_paths = (
        spark.read.table(chunked_table)
        .filter("is_active = true")
        .select("path").distinct().collect()
    )
    existing_set = {row["path"] for row in existing_paths}

    all_paths = [r["path"] for r in new_docs_df.select("path").distinct().collect()]
    updated_paths = [p for p in all_paths if p in existing_set]

    # Soft-delete chunks from updated documents
    if updated_paths:
        path_list = ", ".join([f"'{p}'" for p in updated_paths])
        spark.sql(f"""
            UPDATE {chunked_table}
            SET is_active = false, superseded_by = '{doc_version}'
            WHERE path IN ({path_list}) AND is_active = true
        """)

    # Parse with ai_parse_document v2.0
    parsed_df = new_docs_df.withColumn(
        "parsed_content",
        expr(f"""ai_parse_document(content, map(
            "version", "2.0",
            "imageOutputPath", "{volume_path}/parsed_images/"
        ))""")
    ).drop("content")

    # ... your chunking logic here (RecursiveCharacterTextSplitter) ...
    # produces df_chunks with [path, chunk]

    # Enrich with ACTUAL file modification time (not current_timestamp!)
    mod_times_df = parsed_df.select("path", "modificationTime").distinct()

    df_chunks_enriched = (
        df_chunks
        .join(mod_times_df, on="path", how="left")
        .withColumn("id", F.monotonically_increasing_id()
                    + spark.read.table(chunked_table).count())
        .withColumn("ingested_at", F.current_timestamp())
        .withColumn("source_modified_at", F.col("modificationTime"))
        .withColumn("doc_version", F.lit(doc_version))
        .withColumn("is_active", F.lit(True))
        .withColumn("superseded_by", F.lit(None).cast("string"))
        .drop("modificationTime")
    )

    # Append - CDF picks up the changes automatically
    df_chunks_enriched.write.mode("append").saveAsTable(chunked_table)

    return df_chunks_enriched

Внимательно посмотрите на столбец source_modified_at. Это не current_timestamp(); это реальное время изменения файла, которое отображается при чтении из binaryFile. Эта разница имеет огромное значение для оценки актуальности, поскольку документ, отредактированный три месяца назад, должен получать другой балл, чем документ, отредактированный вчера, независимо от того, когда именно выполнялась процедура обработки.

Проблема 2: отслеживание устаревания

Чистый поиск по сходству не может определить, устарел ли тот или иной фрагмент. В таблице с фрагментами необходимы столбцы метаданных, которые фиксируют жизненный цикл каждого элемента контента.

Схема

Количество столбцов в таблице увеличивается с трех до восьми:

id | path | chunk | ingested_at | source_modified_at | doc_version | is_active | superseded_by

• ingested_at фиксирует момент, когда данные попали в систему обработки, в качестве записи для аудита.
• source_modified_at отмечает момент фактической модификации исходного файла и используется для оценки свежести данных.
• doc_version — это метка версии документа.
• is_active — флаг для «мягкого» удаления; значение false означает исключение документа из результатов поиска.
• superseded_by указывает название более нового документа, который заменил этот.

Добавление столбцов в существующую таблицу

Существующая таблица, такая как docs_chunked, может принять эти столбцы без необходимости пересоздания, а значение source_modified_at можно восстановить на основе реальных временных меток файлов из таблицы parsed-documents:

parsed_table = f"{catalog}.{schema}.docs_parsed"
chunked_table = f"{catalog}.{schema}.docs_chunked"

# Add columns
spark.sql(f"""
    ALTER TABLE {chunked_table}
    ADD COLUMNS (
        ingested_at TIMESTAMP,
        source_modified_at TIMESTAMP,
        doc_version STRING,
        is_active BOOLEAN,
        superseded_by STRING
    )
""")
# Backfill source_modified_at from actual file timestamps
# docs_parsed has modificationTime from the binaryFile reader
spark.sql(f"""
    MERGE INTO {chunked_table} AS chunks
    USING (
        SELECT DISTINCT path, modificationTime
        FROM {parsed_table}
    ) AS parsed
    ON chunks.path = parsed.path
    WHEN MATCHED AND chunks.ingested_at IS NULL THEN
        UPDATE SET
            chunks.ingested_at = current_timestamp(),
            chunks.source_modified_at = parsed.modificationTime,
            chunks.doc_version = '1.0',
            chunks.is_active = true,
            chunks.superseded_by = null
""")

Операции в течение жизненного цикла

При наличии этих столбцов управление свежестью осуществляется с помощью обычного SQL:

# Soft-delete an outdated document
spark.sql(f"""
    UPDATE {chunked_table}
    SET is_active = false
    WHERE path LIKE '%Firmware_Guide_v5%'
""")

# Supersede v6 with v7
spark.sql(f"""
    UPDATE {chunked_table}
    SET is_active = false, superseded_by = 'Firmware_Guide_v7.pdf'
    WHERE path LIKE '%Firmware_Guide_v6%' AND is_active = true
""")
# Staleness report
display(spark.sql(f"""
    SELECT path, doc_version, source_modified_at,
        datediff(current_date(), source_modified_at) as age_days,
        CASE
            WHEN datediff(current_date(), source_modified_at) < 30
                THEN '🟢 Fresh'
            WHEN datediff(current_date(), source_modified_at) < 90
                THEN '🟡 Aging'
            ELSE '🔴 Stale'
        END as freshness
    FROM {chunked_table}
    WHERE is_active = true
    ORDER BY source_modified_at DESC
"""))

Пересоздание индекса

После добавления столбцов пересоздайте индекс, чтобы новые метаданные были синхронизированы:

vsc.create_delta_sync_index_and_wait(
    endpoint_name="<YOUR_VS_ENDPOINT>",
    index_name="<YOUR_INDEX>",
    source_table_name=chunked_table,
    primary_key="id",
    embedding_source_column="chunk",
    embedding_model_endpoint_name="databricks-gte-large-en",
    pipeline_type="TRIGGERED",
    columns_to_sync=[
        "id", "chunk", "path",
        "ingested_at", "source_modified_at",
        "doc_version", "is_active", "superseded_by"
    ],
)

Начиная с этого момента, Vector Search сохраняет поля свежести и может исключать неактивные строки при поступлении запроса.

Проблема 3: предпочтение более новому контенту

Здесь всё становится интереснее. Одни только флаги делают ранжирование среди активных блоков исключительно семантическим. Документ о соответствии от 2023 года и обновлённый документ от прошлого месяца, оба помеченные как активные, получают одинаковое обращение.

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

Оценка

Для этой цели подходит экспоненциальное затухание с настраиваемым периодом полураспада:

import math
from datetime import datetime

def recency_score(timestamp_str, half_life_days=90):
    """
    Exponential decay scoring based on content age.
    - Age 0 days → score 1.0
    - Age = half_life → score 0.5
    - Age = 2× half_life → score 0.25
    """
    try:
        if isinstance(timestamp_str, str):
            parsed_ts = datetime.fromisoformat(
                timestamp_str.replace("Z", "+00:00")
            )
        else:
            parsed_ts = timestamp_str
        age_days = (datetime.now(parsed_ts.tzinfo) - parsed_ts).days
        return math.exp(-0.693 * age_days / half_life_days)
    except Exception:
        return 0.5

Значение half_life_days является основным параметром настройки и должно соответствовать характеру контента. Для быстро обновляющегося контента, такого как примечания к выпуску и отчёты о инцидентах, используйте примерно 30 дней; для материалов с умеренной частотой изменений, таких как техническая документация — 90 дней; а для медленно обновляющегося контента, такого как спецификации архитектуры и стандарты соответствия — 365 дней.

Собственный инструмент поиска

Замените стандартный VectorSearchRetrieverTool на собственный инструмент LangChain, который будет реализовывать эту логику свежести:

from langchain.tools import tool
from databricks.vector_search.client import VectorSearchClient

@tool
def knowledge_search(query: str) -> str:
    """Search knowledge base with recency-weighted ranking.
    Newer documents are prioritized over older ones."""

    vsc = VectorSearchClient(disable_notice=True)
    index = vsc.get_index(index_name="<YOUR_INDEX>")

    # Over-fetch: get 15 results for re-ranking
    results = index.similarity_search(
        query_text=query,
        columns=["chunk", "path", "source_modified_at", "doc_version"],
        filters={"is_active": True},  # Only active chunks
        num_results=15,
    )

    data = results.get("result", {}).get("data_array", [])
    if not data:
        return "No relevant documents found."

    # Re-rank: combined = 70% similarity + 30% recency
    scored = []
    for row in data:
        chunk, path, modified_at, version, similarity = (
            row[0], row[1], row[2], row[3], row[4]
        )
        recency = recency_score(modified_at, half_life_days=90)
        combined = similarity * 0.7 + recency * 0.3
        scored.append({
            "chunk": chunk, "path": path,
            "doc_version": version, "combined_score": combined,
        })

    scored.sort(key=lambda x: x["combined_score"], reverse=True)

    # Return top 5 with source attribution
    parts = []
    for r in scored[:5]:
        parts.append(
            f"[Source: {r['path']} | Version: {r['doc_version']}]\n{r['chunk']}"
        )
    return "\n\n---\n\n".join(parts)

Затем передайте его агенту:

agent = create_agent(
    model=model,
    tools=[knowledge_search],  # Custom tool instead of VectorSearchRetrieverTool
    system_prompt=system_prompt,
    checkpointer=checkpointer,
)

Настройка с помощью YAML

Вместо того чтобы задавать веса вручную, храните их в файле конфигурации агента:

llm_endpoint_name: databricks-gpt-oss-120b
vector_search:
  index_name: <catalog>.<schema>.docs_chunked_index
lakebase:
  host: <your-lakebase-host>
  endpoint: projects/<project>/branches/production/endpoints/primary
  user: <your-email>
retrieval:
  overfetch_count: 15
  final_count: 5
  similarity_weight: 0.7
  recency_weight: 0.3
  half_life_days: 90

Некоторые предустановки, подходящие для разных областей применения:

# Release notes / changelogs (fast-changing)
retrieval:
  similarity_weight: 0.5
  recency_weight: 0.5
  half_life_days: 30
# Architecture / compliance docs (slow-changing)
retrieval:
  similarity_weight: 0.85
  recency_weight: 0.15
  half_life_days: 365
# Mixed knowledge base (balanced)
retrieval:
  similarity_weight: 0.7
  recency_weight: 0.3
  half_life_days: 90

Как соединяются слои

Когда новый PDF попадает в том Unity Catalog, функция ingest_new_documents() считывает его с использованием binaryFile, который фиксирует время изменения. Функция ai_parse_document() извлекает текст, таблицы и рисунки. Рекурсивный разделитель делит текст на части. Старые части с того же пути удаляются «мягко» путем установки значения is_active = false. Новые части добавляются с реальным значением source_modified_at, а функция index.sync() передаёт эти изменения через Change Data Feed.

Когда пользователь задаёт вопрос, специальный инструмент выполняет запрос в Vector Search с фильтром по параметру is_active, получает 15 кандидатов, оценивает каждого по формуле «сходство умноженное на 0,7 плюс свежесть умноженная на 0,3» и передаёт пять лучших результатов модели вместе с метками исходного файла и версии. Затем модель отвечает, отдавая предпочтение более свежему контенту.

Администраторы управляют жизненным циклом с помощью SQL: команда UPDATE ... SET is_active = false используется для устаревания документов, атрибут superseded_by — для записи цепочек версий, причем существует отчет о старости контента, позволяющий увидеть распределение по возрасту.

Какие изменения на практике

До введения весования по свежести запрос вроде «Как производится обновление прошивки?» мог возвращать фрагменты из трех версий руководства по прошивке. Все они были семантически релевантны, но только в последней версии содержались актуальные инструкции, и агент мог объединить информацию из версий v5, v6 и v7 в один ответ.

После добавления фильтра is_active и метрики овеществления source_modified_at:

• фрагменты версии v5 считаются неактивными и исключаются до их получения.
• фрагменты версии v6 также считаются неактивными, отмечаются как замененные версией v7 и исключаются.

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

Теперь агент отвечает на основе текущей версии и указывает точно, какой документ он использовал.

Уроки, которые стоит запомнить

• Используйте реальное время изменения файла, а не время его загрузки. Документ, изменённый три месяца назад, должен выглядеть как старый на три месяца, а не как «только что загруженный сегодня», потому что обработка прошла сегодня утром.
• Применяйте мягкое удаление вместо жёсткого. Установка is_active = false позволяет отслеживать, что и когда было удалено, тогда как жёсткое удаление уничтожает историю.
• Настройте период полураспада в соответствии с областью применения. Тридцать дней подходят для отчётов о инцидентах, 90 — для технической документации, а 365 — для спецификаций архитектуры. Универсального стандарта не существует, поскольку это зависит от скорости изменения контента.
• Выбор весов — простая задача. Сложнее всего правильно настроить source_modified_at и создать единый цикл жизни: определить, кто решает, когда документ устаревает, как отслеживаются версии и когда запускается отчет о старости данных. Сам код прост, но именно процессы, связанные с ним, обеспечивают его работоспособность в производственных условиях.

Документация Databricks по Vector Search Delta Sync, ai_parse_document, Change Data Feed и качеству поиска охватывает основные функции.