Impida que su RAG utilice documentos obsoletos: Seguimiento de la actualidad y el estado de obsolescencia

Supongamos que un sistema RAG ya analiza los PDF, los divide en fragmentos, los incrusta, obtiene el contexto e incluso mantiene la memoria de la conversación a lo largo de las interacciones. Sin embargo, aún puede existir un problema oculto. El índice vectorial trata a cada fragmento de la misma manera. Una guía de firmware de 2022 y un boletín de correcciones de ayer por la tarde tienen el mismo peso en la recuperación de información, siempre y cuando ambas coincidan semánticamente con la consulta. El agente no tiene idea de que un documento haya sido reemplazado o de que una versión más reciente debería tener prioridad.

En la producción, esto causa daños reales: procedimientos de cumplimiento obsoletos, referencias a API desactualizadas y pasos de solución de problemas antiguos que ya no son aplicables. El agente proporciona información desactualizada con total confianza, porque nada le indica lo contrario.

Para solucionar esto se necesitan tres elementos, todos los cuales pueden integrarse en un pipeline de Databricks que utilice Vector Search y Delta Sync.

Tres problemas en tres capas

Parecen ser un mismo problema, pero en realidad son desafíos separados en diferentes etapas del proceso:

1. Ingesta de cambios. Los documentos nuevos y modificados deben llegar al índice sin forzar una reconstrucción completa.
2. Obsolescencia. Cada bloque necesita un registro del estado actual, obsoleto o reemplazado, para que el sistema pueda distinguirlos.
3. Clasificación. Cuando dos bloques son más o menos igualmente relevantes, el más reciente debería tener prioridad.

Problema 1: incorporar contenido nuevo

Con Databricks Vector Search en Delta Sync, ya existe la infraestructura para actualizaciones incrementales. Delta Sync sigue el Change Data Feed de la tabla de origen y envía inserciones, actualizaciones y eliminaciones al índice.

La principal decisión es el tipo de pipeline. En el modo desencadenado, usted decide cuándo sincronizar:

# 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 modo continuo, el índice se actualiza por sí mismo minutos después de los cambios en la tabla:

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

En la mayoría de las configuraciones de producción, el modo desencadenado ofrece más control. Puede ingresar un lote de documentos y sincronizarlos una sola vez, en lugar de provocar una actualización del índice para cada fila.

El flujo de ingreso para documentos nuevos o modificados es el siguiente:

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

Observe atentamente la columna source_modified_at. No se trata de current_timestamp(), sino del verdadero momento de modificación del archivo que revela el lector binaryFile. Esa diferencia es de gran importancia para la evaluación de la recencia, ya que un documento editado hace tres meses debería obtener una puntuación diferente al de uno editado ayer, independientemente del momento en que se ejecute el pipeline de procesamiento.

Problema 2: seguimiento de la antigüedad

La búsqueda pura de similitud no tiene forma de saber si un fragmento está desactualizado. La tabla de fragmentos necesita columnas de metadatos que registren el ciclo de vida de cada contenido.

El esquema

La tabla pasa de tener tres columnas a ocho:

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

• ingested_at registra cuándo el fragmento ingresó al pipeline, como registro de auditoría.
• source_modified_at registra cuándo se modificó realmente el archivo de origen, y sirve para calcular la recencia.
• doc_version es una etiqueta de versión para el documento.
• is_active es una marca de eliminación suave; un valor falso significa que está excluido de la recuperación.
• superseded_by indica el nombre del documento más reciente que reemplazó a este.

Agregar las columnas a una tabla existente

Una tabla que ya existe, como docs_chunked, puede incluir estas columnas sin necesidad de reconstruirla, y source_modified_at puede rellenarse con las marcas de tiempo reales de los archivos en la tabla 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
""")

Operaciones del ciclo de vida

Con esas columnas en su lugar, gestionar la frescura se hace con SQL sencillo:

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

Reconstruyendo el índice

Después de agregar columnas, reconstruya el índice para que los nuevos metadatos se sincronicen:

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

A partir de aquí, la Búsqueda Vectorial incluye los campos de actualidad y puede excluir las filas inactivas cuando llega una consulta.

Problema 3: dar prioridad al contenido más reciente

Aquí es donde las cosas se vuelven interesantes. Las marcas por sí solas hacen que la clasificación entre los fragmentos activos sea puramente semántica. Un documento de cumplimiento de 2023 y uno actualizado del mes pasado, ambos marcados como activos, reciben el mismo trato.

La solución es obtener más datos de los necesarios, volver a clasificarlos según su reciente actualización y devolver solo los mejores.

La puntuación

La decaimiento exponencial con una vida media configurable cumple con esta función:

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

El valor half_life_days es el principal parámetro de ajuste y debe adaptarse al dominio. Utilice unos 30 días para contenido de actualización rápida como notas de lanzamiento e informes de incidentes, 90 días para material que cambia de forma moderada como documentación técnica, y 365 días para contenido de cambio lento como especificaciones de arquitectura y estándares de cumplimiento.

Una herramienta de recuperación personalizada

Sustituya la herramienta predeterminada VectorSearchRetrieverTool por una herramienta personalizada de LangChain que aplique esta lógica de recencia:

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)

Luego entrégueselo al agente:

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

Configurándolo mediante YAML

En lugar de codificar los pesos de forma fija, guárdelos en el archivo de configuración del agente:

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

Algunos ajustes predefinidos adecuados para diferentes dominios:

# 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

Cómo se integran las capas

Cuando un nuevo PDF llega a un volumen de Unity Catalog, ingest_new_documents() lo lee mediante binaryFile, el cual registra la hora de modificación. ai_parse_document() extrae el texto, las tablas y las figuras. Un divisor recursivo divide el texto en fragmentos. Los fragmentos antiguos del mismo camino se eliminan de forma temporal al establecer is_active = false. Los nuevos fragmentos se añaden con la verdadera hora de modificación source_modified_at, y index.sync() transmite el cambio a través del Change Data Feed.

Cuando un usuario hace una consulta, la herramienta personalizada consulta Vector Search con un filtro en is_active, obtiene 15 candidatos, califica cada uno como similitud multiplicada por 0.7 más recencia multiplicada por 0.3, y pasa los cinco mejores al modelo junto con las etiquetas de origen y versión. El modelo luego responde, dando prioridad al contenido más reciente.

Los administradores gestionan el ciclo de vida con SQL: UPDATE ... SET is_active = false para retirar documentos obsoletos, superseded_by para registrar cadenas de versiones, y un informe de antigüedad para ver la distribución por edad del contenido.

Qué cambia en la práctica

Antes de la ponderación por recencia, una consulta como “¿Cuál es el procedimiento de actualización del firmware?” podría devolver fragmentos de tres versiones de la guía del firmware. Todos son semánticamente relevantes, pero solo la más reciente contiene los pasos actuales, y el agente podría combinar v5, v6 y v7 en una sola respuesta.

Después de añadir el filtro is_active y la puntuación basada en source_modified_at:

• Los fragmentos de v5 están inactivos y se filtran antes de su recuperación.
• Los fragmentos de v6 están inactivos, se marcan como reemplazados por v7 y también se filtran.
• Los fragmentos v7 están activos, tienen la puntuación de actualidad más alta y llegan al modelo.

El agente ahora responde a partir de la versión actual y muestra con exactitud qué documento utilizó.

Lecciones importantes

• Use la hora real de modificación del archivo, no la hora de ingestión. Un documento modificado hace tres meses debe parecer tener tres meses de antigüedad, y no “acabado de ser ingresado hoy” solo porque el proceso se ejecutó esta mañana.
• Utilice la eliminación temporal en lugar de la eliminación permanente. Al establecer is_active = false puede auditar qué se eliminó y cuándo, mientras que las eliminaciones permanentes pierden el historial.
• Ajuste el tiempo de vida según el dominio. Treinta días son adecuados para informes de incidentes, 90 para documentos técnicos y 365 para especificaciones de arquitectura. No existe un valor predeterminado universal, ya que depende de la rapidez con la que cambia el contenido.
• Elegir los pesos es sencillo. Lo que requiere esfuerzo es configurar correctamente source_modified_at y establecer un ciclo de vida coherente: quién decide cuándo un documento se reemplaza, cómo se rastrean las versiones y cuándo se ejecuta el informe de obsolescencia. El código es sencillo, pero es el proceso que lo rodea el que hace que funcione en producción.

La documentación de Databricks sobre Vector Search Delta Sync, ai_parse_document, Change Data Feed y calidad de recuperación aborda las funcionalidades subyacentes.