Застаўце ваш RAG ад выдачы застарэлых дакументаў: адстэйквэнція і ступень застарэласці

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

У практычным выкарыстоўванні гэта прыносіць рэальныя нашкоды: застарэлыя процедуры са адпаведнасцю, застарэлыя апісы API, старыя крокі дыягностыкі, якія больш не паслужаюць. Агент з упэўненасцю прадае застарэлую інфармацыю, таму што нічога не паведамляе яго пра іншае.

Ёжыце гэтую проблему можна за дапамою трох прыемоў, якія ўсе падходзяць у паўтарку Databricks з функцыямі Vector Search і Delta Sync.

Тры проблемы на трох роўнях

Яны выглядаюць як адна проблема, але ў рэальнасці яны — аднальныя задачы, якія выступаюць у разныя моменты працэсу апрацоўкі:

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

Проблема 1: адбіранне новага кантэнту

За дапамогою Databricks Vector Search на базе Delta Sync інфраструктура для інкрэментальных апдэйтаў вялікі час вучылася. Delta Sync следуе за Change Data Feed ад выходнай табелі і дадзеўляе індэксу новыя даны, змены чы выдалення.

Главны выбор — гэта тып працэсу апрацоўкі. У рэжыме запуску па сігнале вы самі выбіраеце, калі здзейсніць сінхронізацію:

# 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 перадае поля з інформацыяй прынадзейнасці і можа адзяліць няакtyвныя радкі, калі надходзі запіты.

Проблема 3: праблэма паследжвання новейшага кантэнту

Тут справы становяцца цікавымі. Толькі флагі робяць ранжыраванне акtyваўных частак чыста сэмантычным. Дакумент пра сарабітанне з 2023 года і апдэйтаваны дакумент з мінулага месяца, якія оба пазначаны як акtyваўныя, атрымваюць аднаковае стаўленне.

Рэшэнняя — збіраць больш меркаваіння, пераранжаваць іх за прынадзейнасцю і вярнуць толькі найлепшыя з іх.

Ранжыраванне

Для гэтага выкарыстоўваецца экспанэнцыйны заніжэнне з налаштавальным перыядам паловай жыццёвасці:

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 і якасці адзыскання дакументаў раскрываюць асновныя функцыі.