防止您的 RAG 服务过时文档：最新性与过时性追踪

假设一个RAG系统已经能够解析PDF文件、将其分割成块、嵌入这些块、检索相关上下文，甚至还能在多轮对话中保留记忆。但其中仍可能存在一个较为隐蔽的问题：向量索引对所有数据块一视同仁。只要在语义上与查询匹配，2022年发布的固件指南和上周二发布的补丁公告就会被赋予相同的检索权重。该系统根本无法意识到某个文档已被替代，也不知道应该优先使用更新版本。

在实际应用中，这会带来严重后果：过时的合规流程、已废弃的API引用，以及不再适用的旧故障排查步骤。由于没有其他提示，该系统会满腔信心地提供过时信息。

解决这个问题需要三个要素，而这些要素都可以通过结合向量搜索和Delta Sync的Databricks流程来实现。

三层结构中的三个问题

它们看似是同一个问题，但实际上是处理流程中不同阶段的独立挑战：

1. 变更的导入。新文档和已编辑的文档必须能够被纳入索引，而无需进行完整的重建。
2. 数据过时问题。每个数据块都需要记录其当前状态、是否已过时或已被替代，以便系统能够区分它们。
3. 排序问题。当两个数据块的相关性大致相同时，较新的那个应当排在更靠前的位置。

问题1：如何引入新内容

借助 Databricks Vector Search 与 Delta Sync，实现增量更新的基础设施已经存在。Delta Sync 会跟随源表的变更数据流，将插入、更新和删除操作推送到索引中。

主要需要决定的是处理流程的类型。在触发模式下，由你决定何时进行同步：

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

从现在开始，向量搜索会携带新鲜度字段，因此在接收到查询时可以排除那些不活跃的行。

问题3：偏好最新内容

这里的情况就变得有趣了。仅依靠标记无法让活跃数据块之间的排序完全基于语义。一份2023年的合规文件和一份上个月更新的文件，只要都被标记为活跃，就会得到相同的处理方式。

解决办法是先大量获取数据，再根据最新性进行重新排序，最后仅返回最优秀的几项。

评分机制

通过可配置半衰期的指数衰减函数即可实现这一目标：

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天。

自定义检索工具

可以用一个封装了此最新性逻辑的定制LangChain工具来替换现有的VectorSearchRetrieverTool：

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() 会将这些变更通过变更数据流传递出去。

当用户提出查询时，自定义工具会基于 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设为true可以记录被删除的内容及其时间，而硬删除则会丢失相关历史记录。
• 需根据领域调整半衰期长度：事故报告适合30天，技术文档适合90天，架构规范则适合365天。由于内容更新速度不同，不存在通用的默认值。
• 选择权重很简单，真正需要费力的是正确设置source_modified_at以及建立统一的生命周期管理机制：由谁来决定何时更新文档、如何追踪版本，以及何时运行过时性报告。代码本身很简单，但围绕它的流程才是确保其在生产环境中正常运行的关键。

Databricks关于向量搜索Delta同步、ai_parse_document、变更数据流以及检索质量的相关文档介绍了这些功能背后的原理。