Arrêtez votre RAG de servir des documents obsolètes : suivi de la fraîcheur et de l’obsolescence

Supposons qu’un système RAG analyse déjà les PDF, les divise en blocs, les intègre, récupère le contexte et conserve même la mémoire des conversations au fil des échanges. Un problème moins évident peut cependant exister en interne. L’index vectoriel traite tous les blocs de la même manière. Un guide de firmware datant de 2022 et un bulletin de correctif publié mardi dernier ont le même poids de récupération tant qu’ils correspondent sémantiquement à la requête. L’agent n’a aucune idée qu’un document a été remplacé ou qu’une version plus récente devrait prévaloir.

Dans un environnement de production, cela entraîne des dommages réels : procédures de conformité obsolètes, références API dépréciées, étapes de dépannage anciennes qui ne s’appliquent plus. L’agent fournit des informations périmées avec une confiance totale, car rien ne lui indique le contraire.

Pour résoudre ce problème, il faut trois éléments, qui s’intègrent tous dans un pipeline Databricks utilisant Vector Search et Delta Sync.

Trois problèmes à trois niveaux

Ils semblent être un seul problème, mais ce sont en réalité des défis distincts à différents stades du pipeline :

1. Ingestion des modifications. Les documents nouveaux et modifiés doivent parvenir à l’index sans nécessiter une reconstruction complète.
2. Dépassement de date. Chaque bloc nécessite un enregistrement de son état actuel, obsolète ou remplacé, afin que le système puisse les distinguer.
3. Classement. Lorsque deux blocs ont un niveau de pertinence à peu près égal, le plus récent doit avoir la priorité.

Problème 1 : intégration de nouveaux contenus

Avec Databricks Vector Search sur Delta Sync, l’infrastructure pour les mises à jour incrémentielles existe déjà. Delta Sync suit le Change Data Feed de la table source et insère, met à jour ou supprime des données dans l’index.

Le choix principal concerne le type de pipeline. En mode déclenché, c’est vous qui décidez quand effectuer la synchronisation :

# 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

En mode continu, l’index se met à jour automatiquement quelques minutes après toute modification de la table :

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

Pour la plupart des configurations de production, le mode déclenché offre un contrôle accru. Vous pouvez importer un lot de documents et effectuer une synchronisation unique, au lieu de déclencher une mise à jour de l’index pour chaque ligne.

Le flux d’importation pour les documents nouveaux ou modifiés se présente comme suit :

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

Examinez de près la colonne source_modified_at. Ce n’est pas current_timestamp(). Il s’agit de l’heure réelle de modification du fichier telle qu’elle est exposée par le lecteur binaryFile. Cette différence est très importante pour l’évaluation de la fraîcheur, car un document modifié il y a trois mois devrait obtenir un score différent de celui d’un document modifié hier, quel que soit le moment où le pipeline d’ingestion s’exécute.

Problème 2 : suivi de la péremption

La recherche de similarité pure ne sait pas qu’un morceau de contenu est obsolète. La table des morceaux nécessite des colonnes de métadonnées qui enregistrent le cycle de vie de chaque contenu.

Le schéma

La table passe de trois colonnes à huit :

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

• ingested_at enregistre le moment où le morceau est entré dans le pipeline, à des fins de traçabilité.
• source_modified_at enregistre le moment où le fichier source a réellement été modifié, et sert à évaluer la fraîcheur des données.
• doc_version est une étiquette de version pour le document.
• is_active est un indicateur de suppression soft ; une valeur fausse signifie que le document est exclu des recherches.
• superseded_by indique le nom du document plus récent qui a remplacé celui-ci.

Ajout des colonnes à une table existante

Une table déjà existante, comme docs_chunked, peut intégrer ces colonnes sans avoir à être reconstruite, et source_modified_at peut être rempli à partir des timestamps réels des fichiers dans la table 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
""")

Opérations liées au cycle de vie

Avec ces colonnes en place, gérer la fraîcheur se fait simplement à l’aide de 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
"""))

Reconstruction de l’index

Après avoir ajouté des colonnes, reconstruisez l’index afin que les nouvelles métadonnées soient synchronisées :

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

Désormais, la recherche vectorielle prend en compte les champs de fraîcheur et peut exclure les lignes inactives lorsqu’une requête arrive.

Problème 3 : privilégier les contenus plus récents

Ici, les choses deviennent intéressantes. Les seules étiquettes laissent le classement parmi les blocs actifs purement sémantique. Un document de conformité de 2023 et un document mis à jour du mois dernier, tous deux marqués comme actifs, reçoivent un traitement identique.

La solution consiste à récupérer davantage de données, à réclasser en fonction de la récente mise à jour, et à ne retourner que les meilleurs résultats.

Le scoring

Un décroissance exponentielle avec une durée de vie configurable suffit à ce faire :

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

La valeur half_life_days est le principal paramètre de réglage et doit correspondre au domaine. Utilisez environ 30 jours pour du contenu à mise à jour fréquente comme les notes de version et les rapports d’incidents, 90 jours pour du matériel modérément mis à jour comme les documents techniques, et 365 jours pour du contenu peu mis à jour comme les spécifications architecturales et les normes de conformité.

Outil de récupération personnalisé

Remplacez le VectorSearchRetrieverTool standard par un outil LangChain personnalisé qui intègre cette logique de fraîcheur :

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)

Puis remettez-le à l’agent :

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

Configuration via YAML

Plutôt que de coder les poids directement, conservez-les dans le fichier de configuration de l’agent :

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

Quelques préférences adaptées à différents domaines :

# 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

Comment les couches s’assemblent

Lorsqu’un nouveau PDF est ajouté dans un volume de Unity Catalog, ingest_new_documents() le lit à l’aide de binaryFile, qui en enregistre l’heure de modification. ai_parse_document() extrait le texte, les tableaux et les figures. Un outil de segmentation récursive divise le texte en blocs. Les blocs plus anciens provenant du même chemin sont supprimés de manière temporaire en définissant is_active = false. Les nouveaux blocs sont ajoutés avec l’heure réelle de modification source_modified_at, et index.sync() transmet ces modifications via le Change Data Feed.

Lorsqu’un utilisateur pose une question, l’outil personnalisé interroge Vector Search avec un filtre sur is_active, récupère 15 candidats, évalue chacun en multipliant la similarité par 0,7 et la fraîcheur par 0,3, puis transmet les cinq meilleurs résultats au modèle accompagnés des étiquettes de source et de version. Le modèle répond ensuite, en privilégiant le contenu le plus récent.

Les administrateurs gèrent le cycle de vie à l’aide de SQL : UPDATE ... SET is_active = false pour retirer les documents obsolètes, superseded_by pour enregistrer les chaînes de versions, ainsi qu’un rapport de péremption pour visualiser la répartition par âge du contenu.

Quelles sont les modifications en pratique ?

Au préalable à l’application du poids de la fraîcheur, une requête telle que « Quelle est la procédure de mise à jour du firmware ? » pouvait retourner des extraits de trois versions du guide du firmware. Tous étaient sémantiquement pertinents, mais seule la version la plus récente contenait les étapes actuelles, et l’agent pouvait combiner les versions v5, v6 et v7 en une seule réponse.

Après l’ajout du filtre is_active et de la notation source_modified_at :

• les extraits de la version v5 sont inactifs et filtrés avant leur récupération.
• les extraits de la version v6 sont inactifs, marqués comme remplacés par la version v7, et également filtrés.
• Les blocs v7 sont actifs, possèdent le score de fraîcheur le plus élevé et sont transmis au modèle.

L’agent répond désormais à partir de la version actuelle et indique précisément quel document il a utilisé.

Leçons à retenir

• Utilisez l’heure réelle de modification du fichier, et non l’heure d’ingestion. Un document modifié il y a trois mois doit paraître avoir trois mois, et non « venant d’être ingéré aujourd’hui » simplement parce que le pipeline a fonctionné ce matin.
• Préférez la suppression temporaire à la suppression définitive. En définissant is_active = false, vous pouvez auditer ce qui a été supprimé et quand, tandis que les suppressions définitives effacent l’historique.
• Adaptez la durée de vie aux besoins du domaine. Trente jours conviennent aux rapports d’incidents, 90 aux documents techniques et 365 aux spécifications architecturales. Il n’existe pas de valeur par défaut universelle, car cela dépend de la vitesse à laquelle le contenu évolue.
• Choisir les poids est simple. Ce qui demande du travail, c’est de bien gérer source_modified_at et d’établir un cycle de vie cohérent : déterminer qui décide quand un document est remplacé, comment les versions sont suivies, et quand le rapport de péremption est généré. Le code est simple, mais c’est le processus qui l’entoure qui permet à ce système de fonctionner en production.

La documentation Databricks sur Vector Search Delta Sync, ai_parse_document, Change Data Feed ainsi que la qualité de la récupération couvre les fonctionnalités de base.