Zatrzymaj swoje RAG przed używaniem przestarzałych dokumentów: monitorowanie aktualności i przestarzałości

Załóżmy, że system RAG już analizuje pliki PDF, dzieli je na fragmenty, włącza je do swojej struktury, pobiera kontekst i nawet przechowuje pamięć rozmowy między poszczególnymi krokami. Mimo to w jego strukturze może istnieć inne, mniej oczywiste problemy. Indeks wektorowy traktuje wszystkie fragmenty jednakowo – przewodnik dotyczący oprogramowania z 2022 roku oraz komunikat z poprawką z zeszłego wtorku mają taką samą wagę przy wyszukiwaniu, o ile oba semantycznie odpowiadają zapytaniu. Agent nie ma pojęcia, że dany dokument został zastąpiony lub że powinna przeważyć nowsza wersja.

W warunkach produkcyjnych to powoduje poważne konsekwencje: przestarzałe procedury zgodności, nieaktualne odniesienia do API, stare kroki naprawcze, które już nie są stosowane. Agent z pełnym przekonaniem dostarcza przestarzałe informacje, ponieważ nic mu nie wskazuje inaczej.

Ustąpienie tego problemu wymaga trzech elementów, które wszystkie pasują do pipeline’a w Databricks wykorzystującego Vector Search i Delta Sync.

Trzy problemy na trzech poziomach

Wyglądają jak jeden problem, ale są to odrębne wyzwania w różnych etapach procesu:

1. Przyjmowanie zmian. Nowe i zmodyfikowane dokumenty muszą trafić do indeksu bez konieczności całkowitej jego ponownej budowy.
2. Starość danych. Każdy fragment musi mieć zapisany stan – aktualny, przestarzały lub zastąpiony – aby system mógł je odróżnić.
3. Klasyfikacja. Gdy dwa fragmenty są mniej więcej równie istotne, ten nowszy powinien mieć wyższą pozycję.

Problem 1: wprowadzanie nowej treści

Dzięki Databricks Vector Search w połączeniu z Delta Sync infrastruktura do aktualizacji inkrementalnych już istnieje. Delta Sync śledzi Change Data Feed źródłowej tabeli i wprowadza do indeksu nowe dane, aktualizacje oraz usunięcia.

Głównym wyborem jest typ procesu. W trybie wyzwalanym sam decydujesz, kiedy dokonać synchronizacji:

# 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

W trybie ciągłym indeks aktualizuje się w ciągu kilku minut po zmianach w tabeli:

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

Dla większości konfiguracji produkcyjnych tryb wyzwalany zapewnia większą kontrolę. Możesz zaimportować partię dokumentów i zsynchronizować je raz, zamiast uruchamiać aktualizację indeksu dla każdego wiersza.

Proces importowania nowych lub zmienionych dokumentów wygląda następująco:

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

Przyjrzyj się uważnie kolumnie source_modified_at. To nie jest current_timestamp(). To rzeczywisty czas modyfikacji pliku, który ujawnia odczytujnik binaryFile. Ta różnica ma ogromne znaczenie dla oceny aktualności, ponieważ dokument edytowany trzy miesiące temu powinien otrzymać inną ocenę niż ten edytowany wczoraj, niezależnie od tego, kiedy akurat uruchomiono proces przetwarzania.

Problem 2: śledzenie przestarzałości

Czyste wyszukiwanie podobieństw nie wie, że dany fragment jest przestarzały. Tabela z fragmentami wymaga kolumn metadanych, które rejestrują cykl życia każdego elementu treści.

Schemat

Tabela rozszerza się z trzech kolumn do ośmiu:

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

• ingested_at rejestruje moment, w którym fragment trafił do procesu przetwarzania, służąc jako ślad audytowy.
• source_modified_at odnotowuje moment faktycznej modyfikacji pliku źródłowego i jest wykorzystywany do oceny aktualności.
• doc_version to etykieta wersji dokumentu.
• is_active to flaga służąca do delikatnego usuwania, przy czym wartość false oznacza wykluczenie z wyników wyszukiwania.
• superseded_by określa nowszy dokument, który zastąpił ten obecny.

Dodawanie kolumn do istniejącej tabeli

Tabela, która już istnieje, np. docs_chunked, może przyjąć te kolumny bez konieczności jej ponownego budowania, a wartość source_modified_at można uzupełnić na podstawie rzeczywistych dat z plików w tabeli 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
""")

Operacje w cyklu życia

Gdy te kolumny są już dostępne, zarządzanie świeżością odbywa się za pomocą zwykłego 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
"""))

Odbudowa indeksu

Po dodaniu kolumn odbuduj indeks, aby nowe metadane zostały zsynchronizowane:

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

Od teraz Vector Search przechowuje pola aktualności i może wykluczać nieruchome wiersze po otrzymaniu zapytania.

Problem 3: faworyzowanie nowszego treści

Tutaj sytuacja staje się ciekawsza. Same flagi sprawiają, że ranking aktywnych fragmentów jest wyłącznie semantyczny. Dokument zgodności z 2023 roku oraz zaktualizowany dokument z zeszłego miesiąca, oba oznaczone jako aktywne, otrzymują identyczne traktowanie.

Rozwiązaniem jest pobieranie większej ilości danych, ponowne rankowanie według aktualności i zwracanie tylko najlepszych kilku wyników.

Ocena

Do tego celu służy wykładnicze spadanie z konfigurowalnym okresem połowicznym:

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

Wartość half_life_days jest głównym parametrem do dostosowywania i powinna odpowiadać charakterystyce domeny. Użyj około 30 dni dla treści szybko zmieniających się, takich jak notatki o wydaniu czy raporty incydentów, 90 dni dla materiałów zmieniających się w umiarkowanym tempie, jak dokumentacja techniczna, oraz 365 dni dla treści wolno zmieniających się, takich jak specyfikacje architektury i standardy zgodności.

Spersonalizowane narzędzie do wyszukiwania

Zastąp standardowe VectorSearchRetrieverTool spersonalizowanym narzędziem LangChain, które implementuje tę logikę aktualności:

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)

Następnie przekaż to agentowi:

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

Konfiguracja za pomocą YAML

Zamiast kodować wagę bezpośrednio, przechowuj je w pliku konfiguracyjnym agenta:

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

Część ustawień domyślnych odpowiednich dla różnych dziedzin:

# 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

Jak warstwy ze sobą współpracują

Gdy nowy plik PDF trafia do wolumenu Unity Catalog, funkcja ingest_new_documents() odczytuje go za pomocą binaryFile, który rejestruje czas modyfikacji. Funkcja ai_parse_document() wydobywa tekst, tabele i rysunki. Procedura rozdzielania w sposób rekurencyjny dzieli tekst na fragmenty. Starsze fragmenty z tego samego ścieżki są „miękko usuwane” poprzez ustawienie is_active = false. Nowe fragmenty są dodawane z rzeczywistym wartością source_modified_at, a funkcja index.sync() przekazuje te zmiany przez Change Data Feed.

Gdy użytkownik zadaje pytanie, specjalistyczne narzędzie wysyła zapytanie do Vector Search z filtrem na is_active, pobiera 15 kandydatów, ocenia każdego jako iloczyn podobieństwa i 0,7 oraz aktualności i 0,3, a następnie przekazuje pięć najlepszych wyników modelowi wraz z etykietami źródła i wersji. Model następnie udziela odpowiedzi, preferując nowszy treść.

Administratorzy zarządzają cyklem życia dokumentów za pomocą SQL: UPDATE ... SET is_active = false w celu usunięcia przestarzałych dokumentów, superseded_by do rejestrowania łańcuchów wersji oraz raportu aktualności, aby zobaczyć rozkład wieku treści.

Jakie są zmiany w praktyce

Przed wdrożeniem wagowania według aktualności zapytanie takie jak „Jaka jest procedura aktualizacji oprogramowania?” mogło zwrócić fragmenty z trzech wersji przewodnika po oprogramowaniu. Wszystkie są semantycznie istotne, ale tylko najnowsza zawiera aktualne kroki, więc agent mógłby połączyć wersje v5, v6 i v7 w jedną odpowiedź.

Po dodaniu filtru is_active oraz oceny na podstawie source_modified_at:

• Fragmenty wersji v5 są nieaktywne i są filtrowane przed pobraniem.
• Fragmenty wersji v6 są nieaktywne, oznaczone jako zastąpione przez v7 i również są filtrowane.
• Fragmenty wersji v7 są aktywne, mają najwyższą ocenę aktualności i trafiają do modelu.

Agent teraz odpowiada na podstawie bieżącej wersji i pokazuje dokładnie, który dokument został użyty.

Lekcje, które warto zapamiętać

• Używaj rzeczywistego czasu modyfikacji pliku, a nie czasu jego przyjęcia. Dokument zmodyfikowany trzy miesiące temu powinien wyglądać na trzy miesiące stary, a nie „dopiero co przyjęty dziś”, tylko dlatego, że proces został uruchomiony dziś rano.
• Zastosuj delikatne usuwanie zamiast trwałego. Ustawienie is_active = false umożliwia sprawdzenie, co i kiedy zostało usunięte, podczas gdy trwałe usuwanie niszczy historię.
• Dostosuj okres ważności do danej dziedziny. Trzydzieści dni nadaje się do raportów z incydentów, 90 do dokumentacji technicznej, a 365 do specyfikacji architektury. Nie istnieje uniwersalny standard, ponieważ zależy to od szybkości zmian w treści.
• Wybór wag jest prosty. Trudności pojawiają się przy poprawnym ustawieniu pola source_modified_at oraz przy ustaleniu spójnego cyklu życia: kto decyduje, kiedy dokument zostaje zastąpiony, jak śledzone są wersje oraz kiedy wykonywany jest raport o przestarzałości danych. Sam kod jest prosty, ale to procesy towarzyszące sprawiają, że funkcjonuje on w środowisku produkcyjnym.

Dokumentacja Databricks dotycząca Vector Search Delta Sync, ai_parse_document, Change Data Feed oraz jakości wyszukiwania omawia podstawowe funkcje.