Verhindern Sie, dass Ihr RAG veraltete Dokumente bereitstellt: Überwachung von Aktualität und Veraltung

Nehmen wir an, ein RAG-System analysiert bereits PDFs, teilt sie in Blöcke auf, integriert sie, holt Kontext ab und behält sogar das Gesprächsgedächtnis über verschiedene Interaktionen hinweg. Dennoch kann sich ein weiteres Problem darin verbergen. Der Vektorindex behandelt alle Blöcke gleich – ein Firmware-Leitfaden aus dem Jahr 2022 und ein Hotfix-Bulletin vom vorigen Dienstag haben denselben Suchgewicht, solange beide semantisch zur Abfrage passen. Das System hat keine Ahnung davon, dass ein Dokument ersetzt wurde oder dass eine neuere Version Vorrang haben sollte.

In der Praxis führt dies zu erheblichen Schäden: veraltete Konformitätsverfahren, nicht mehr gültige API-Referenzen sowie alte Fehlerbehebungsschritte, die nicht mehr anwendbar sind. Das System liefert Informationen, die veraltet sind, mit voller Zuversicht – weil ihm nichts anderes mitgeteilt wurde.

Um dieses Problem zu beheben, sind drei Komponenten erforderlich, die alle in einen Databricks-Pipeline mit Vector Search und Delta Sync passen.

Drei Probleme auf drei Ebenen

Es scheint sich um ein einziges Problem zu handeln, doch es handelt sich um getrennte Herausforderungen an verschiedenen Stellen im Pipeline-Prozess:

1. Einbezug von Änderungen. Neue und bearbeitete Dokumente müssen zum Index gelangen, ohne dass eine vollständige Neuberechnung erforderlich ist.
2. Veraltetheit. Jeder Datenblock benötigt einen dokumentierten Zustand – aktuell, veraltet oder ersetzt – damit das System sie voneinander unterscheiden kann.
3. Rangfolge. Wenn zwei Datenblöcke etwa gleich relevant sind, sollte der neuere Vorrang haben.

Problem 1: Einfügung neuer Inhalte

Mit Databricks Vector Search auf Delta Sync existiert bereits die Infrastruktur für inkrementelle Aktualisierungen. Delta Sync folgt dem Change Data Feed der Quelltabelle und überträgt Einfüge, Aktualisierungen und Löschungen in den Index.

Die wichtigste Entscheidung betrifft den Typ des Pipelines. Im auslösbaren Modus entscheiden Sie selbst, wann synchronisiert werden soll:

# 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

Im kontinuierlichen Modus aktualisiert sich der Index innerhalb weniger Minuten nach Änderungen in der Tabelle:

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

Bei den meisten Produktionsumgebungen bietet der auslösende Modus mehr Kontrolle. Sie können eine Gruppe von Dokumenten aufnehmen und einmal synchronisieren, anstatt für jede Zeile einen Index-Update auszulösen.

Der Ablauf bei der Aufnahme neuer oder geänderter Dokumente sieht wie folgt aus:

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

Betrachten Sie genau die Spalte source_modified_at. Es handelt sich dabei nicht um current_timestamp(), sondern um die tatsächliche Dateiänderungszeit, die vom binaryFile-Leser bereitgestellt wird. Dieser Unterschied ist für die Bewertung der Aktualität von großer Bedeutung, denn ein Dokument, das vor drei Monaten bearbeitet wurde, sollte anders bewertet werden als eines, das gestern bearbeitet wurde – unabhängig davon, wann der Eingabeprozess ausgeführt wurde.

Problem 2: Überwachung der Veraltetheit

Bei einer reinen Ähnlichkeitssuche wird nicht erkannt, ob ein Datensatz veraltet ist. Die Tabelle mit den Datensätzen benötigt Metadatumsäulen, die den Lebenszyklus jedes Inhaltsstücks erfassen.

Das Schema

Die Tabelle wächst von drei auf acht Spalten an:

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

• ingested_at dokumentiert, wann der Datensatz in den Pipeline eingefügt wurde, als Prüfungsverlauf.
• source_modified_at dokumentiert, wann die Quelldatei tatsächlich geändert wurde, und dient zur Bewertung der Aktualität.
• doc_version ist eine Versionsbezeichnung für das Dokument.
• is_active ist ein Flag für eine sanfte Löschung, wobei „false“ bedeutet, dass der Datensatz von der Abfrage ausgeschlossen wird.
• superseded_by gibt den neueren Datensatz an, der diesen ersetzt hat.

Hinzufügen der Spalten zu einer vorhandenen Tabelle

Eine bereits vorhandene Tabelle, wie z. B. docs_chunked, kann diese Spalten ohne Neubau aufnehmen, und source_modified_at kann aus den tatsächlichen Zeitstempeln der Dateien in der Tabelle „parsed-documents“ nachgefüllt werden:

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
""")

Lebenszyklusverwaltung

Sobald diese Spalten vorhanden sind, ist die Verwaltung der Frische ein einfacher SQL-Befehl:

# 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
"""))

Index neu erstellen

Nach dem Hinzufügen von Spalten muss der Index neu erstellt werden, damit die neuen Metadaten synchronisiert werden:

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"
    ],
)

Von nun an enthält die Vektorabfrage die Aktualitätsfelder und kann inaktivere Zeilen bei einem Eingabefrage ausschließen.

Problem 3: Bevorzugung neuerer Inhalte

Hier wird es interessant. Allein die Flags lassen die Rangfolge der aktiven Blöcke rein semantisch bestimmen. Ein Konformitätsdokument aus dem Jahr 2023 und ein aktualisiertes Dokument vom letzten Monat, beide als aktiv markiert, erhalten dieselbe Behandlung.

Die Lösung besteht darin, mehr Daten zu holen, mithilfe der Aktualität erneut zu rangieren und nur die besten Ergebnisse zurückzugeben.

Die Bewertung

Ein exponentieller Abfall mit einer einstellbaren Halbwertszeit erledigt die Aufgabe:

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

Der Wert half_life_days ist die wichtigste Einstellmöglichkeit und sollte dem jeweiligen Bereich entsprechen. Verwenden Sie etwa 30 Tage für schnell ändernden Inhalt wie Release-Notes und Incidents-Berichte, 90 Tage für mäßig häufig geänderten Material wie technische Dokumentationen sowie 365 Tage für langsam ändernden Inhalt wie Architekturspezifikationen und Konformitätsstandards.

Eine benutzerdefinierte Abrufwerkzeug

Ersetzen Sie das standardmäßige VectorSearchRetrieverTool durch ein benutzerdefiniertes LangChain-Tool, das diese Aktualitätslogik umsetzt:

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)

Dann übergeben Sie es dem Agenten:

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

Konfigurieren über YAML

Anstatt die Gewichte fest einzuschreiben, speichern Sie sie im Konfigurationsfile des Agents:

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

Einige Vorlagen, die für verschiedene Bereiche geeignet sind:

# 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

Wie die Schichten zusammenpassen

Wenn ein neuer PDF in einem Unity Catalog-Volumen ankommt, liest ingest_new_documents() ihn mit binaryFile ein, wodurch die Änderungszeit erfasst wird. ai_parse_document() extrahiert Text, Tabellen und Abbildungen. Ein rekursiver Splitter teilt den Text in Abschnitte auf. Ältere Abschnitte aus demselben Pfad werden durch Setzen von is_active = false sanft gelöscht. Neue Abschnitte werden mit der tatsächlichen source_modified_at-Zeit hinzugefügt, und index.sync() überträgt die Änderung über den Change Data Feed.

Wenn ein Benutzer eine Anfrage stellt, fragt das benutzerdefinierte Tool Vector Search mit einem Filter auf is_active ab, holt 15 Kandidaten, bewertet jeden als Ähnlichkeitswert mal 0,7 plus Aktualitätswert mal 0,3 und übermittelt die fünf besten an das Modell zusammen mit Quell- und Versionsbezeichnungen. Das Modell antwortet anschließend und gibt neueren Inhalten Vorrang.

Administratoren verwalten den Lebenszyklus mit SQL: UPDATE ... SET is_active = false zur Außerbetriebnahme veralteter Dokumente, superseded_by zur Erfassung von Versionsketten sowie einem Aktualitätsbericht, um die Altersverteilung des Inhalts zu sehen.

Was ändert sich in der Praxis?

Vor der Gewichtung nach Aktualität könnte eine Abfrage wie „Wie verläuft das Verfahren zur Firmware-Update?“ Teile aus drei Versionen der Firmware-Anleitung zurückgeben. Alle sind semantisch relevant, doch nur die neueste enthält die aktuellen Schritte, wodurch der Agent v5, v6 und v7 zu einer einzigen Antwort kombinieren könnte.

Nach Hinzufügen des is_active-Filters und der Bewertung nach source_modified_at:

• v5-Teile sind inaktiv und werden vor der Abrufung ausgefiltert.
• v6-Teile sind inaktiv, als durch v7 ersetzt markiert und ausgefiltert.

v7-Chunks sind aktiv, haben die höchste Aktualitätsbewertung und werden an das Modell weitergeleitet.

Der Agent antwortet nun mit der aktuellen Version und zeigt genau an, welches Dokument er verwendet hat.

Lektionen, die man beibehalten sollte

• Verwenden Sie die tatsächliche Änderungszeit der Datei, nicht die Zeit des Einreichens. Ein Dokument, das vor drei Monaten geändert wurde, sollte wie drei Monate alt aussehen und nicht als „heute erst eingereicht“ gelten, nur weil der Pipeline-Prozess heute Morgen ausgeführt wurde.
• Verwenden Sie eine sanfte Löschung statt einer vollständigen Löschung. Durch das Setzen von is_active = false können Sie nachverfolgen, was und wann entfernt wurde, während eine vollständige Löschung die Historie verliert.
• Passen Sie die Halbwertszeit an den jeweiligen Anwendungsbereich an. Dreißig Tage eignen sich für Incidentsberichte, 90 Tage für technische Dokumente und 365 Tage für Architekturspezifikationen. Es gibt keinen universellen Standard, da dies von der Geschwindigkeit, mit der der Inhalt sich ändert, abhängt.
• Die Auswahl der Gewichte ist einfach. Herausfordernd ist es, source_modified_at korrekt einzustellen und einen konsistenten Lebenszyklus zu etablieren: Wer entscheidet, wann ein Dokument ersetzt wird, wie Versionen verfolgt werden und wann der Veraltungsbericht ausgeführt wird. Der Code selbst ist einfach, doch der um ihn herumliegende Prozess sorgt dafür, dass er in der Produktion funktioniert.

Die Databricks-Dokumentation zu Vector Search Delta Sync, ai_parse_document, Change Data Feed sowie der Suchqualität behandelt die zugrundeliegenden Funktionen.