Зупиніть ваш 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 та якості пошуку охоплює основні функції.