知识库运维：文档天天改，RAG 怎么做增量更新与版本管理

本文是 AI 应用系统学习系列的 L3 实战篇。前置：RAG 文档处理：PDF、表格、切分策略这些脏活才是效果上限。 学完可以配合面试题食用：02. RAG 流水线、26. RAG 评估忠实度。

先问一句：灌完库就不管了，会发生什么

想象你入职一家公司，行政递给你一本员工手册，同时口头补了一句：「第三章作废了，别看。」这句话就是增量更新的本质——知识有生命周期，光有新知识不够，还得知道哪些旧的失效了。

RAG 知识库的默认状态是没有这句口头补丁的。源文档系统里 3 月版本已经作废，向量库里 3 月、8 月两套 chunk 并排躺着。用户问「试用期多长」，检索召回哪条全凭余弦相似度——如果作废的那条措辞和 query 更接近，旧制度就赢了。更麻烦的是这种错不报异常、不打日志，用户拿到的是措辞通顺、内容过期的答案。

结果是企业问答系统的经典翻车现场：上线第一个月好评，第三个月开始有人吐槽「这系统还在引用去年废掉的差旅标准」。问题不出在检索算法，出在把向量库当成了只进不出的仓库。知识库不是灌一次就完的项目，它是个天天有人改文档、需要运维的服务。

文档生命周期：四种事件，删除最容易漏

对着源文档系统盘点，知识库要处理的事件就四种：

• 新增：新文档入库。最简单，切分、embedding、写入三步走完。
• 修改：文档内容变了。要找出哪些 chunk 受影响，只重建这部分。
• 删除：源文档没了。必须把向量库里对应的所有 chunk 一起删掉。
• 作废：文档还在，但失效了（历史版本、废止制度）。不能物理删——审计可能要查历史——检索时必须过滤掉。

stateDiagram-v2
    [*] --> 生效: 新增入库
    生效 --> 修改中: 源文档变更
    修改中 --> 生效: 增量重建完成
    生效 --> 作废: 新版本生效 / 制度废止
    生效 --> [*]: 源文档删除
    作废 --> [*]: 超过保留期清理

删除是四种事件里最容易漏的。理由很直白：删除是「负面事件」，同步任务通常只扫「有什么新东西」，没人扫「少了什么」。源系统里删掉一份文档，知识库这边毫无感知，向量库里留下几十个僵尸 chunk。处理办法是把「这次扫描里没出现」当成事件：全量列举源文档 id，库里存在、源里消失的进删除队列。作废通常靠元数据驱动：源系统带状态字段，同步写进 chunk 元数据，检索时过滤。

增量更新：稳定主键 + hash 比对

增量更新的基础是两件事：稳定的主键，和便宜的变化检测。

稳定主键指 doc_id 和 chunk_id 在文档两次同步之间保持不变。chunk_id = md5(doc_id + chunk_index) 这种由位置生成的 id 有个坑：文档开头插了一段，后面所有 chunk 的序号整体后移，hash 全变，等于全文档重建。更稳的做法是按内容边界生成：md5(doc_id + section_heading + chunk_text)，没变的段落 hash 不变，插一段只影响插入点附近的块。

hash 比对解决「怎么知道变了」。解析 + 切分 + embedding 一整套跑下来，一份百页 PDF 要几十秒，全量重灌的成本随文档数线性涨。实际变化率很低——企业知识库一天往往只改几份文档——所以同步任务应该只对「解析后文本的 hash」有变化的文档重建：

import hashlib

def file_fingerprint(path: str) -> str:
    """文件级快速比对：mtime 或大小变了才算内容 hash，省解析成本"""
    md5 = hashlib.md5()
    with open(path, "rb") as f:
        for block in iter(lambda: f.read(1 << 20), b""):
            md5.update(block)
    return md5.hexdigest()

def sync_once(conn, embed_fn, qdrant):
    for doc in conn.list_active_docs():
        # 第一层：文件 hash 没变直接跳过，解析都省了
        fp = file_fingerprint(doc.local_path)
        old = conn.get_sync_state(doc.doc_id)
        if old and old.fingerprint == fp:
            continue

        chunks = parse_and_chunk(doc.local_path)          # 解析 + 切分
        new_hashes = {c.hash for c in chunks}

        # 第二层：chunk 级比对，只动变化的块
        old_hashes = conn.list_chunk_hashes(doc.doc_id)
        removed = old_hashes - new_hashes
        added   = new_hashes - old_hashes

        for h in removed:
            qdrant.delete(doc.doc_id, h)                  # 旧块删掉
        for h in added:
            c = chunks.by_hash(h)
            vec = embed_fn(c.text)
            qdrant.upsert(doc.doc_id, c, vec, meta={"version": doc.version})
        conn.save_sync_state(doc.doc_id, fp, new_hashes)

两个要点：第一，比对的粒度是 chunk 不是文档——文档里改了一段话，只有受影响的块需要重算 embedding，其他块原样保留（所以主键必须稳定）；第二，删除和新增必须放在同一个事务里执行，先删后加的顺序错了，中间窗口内检索会返回空结果。

版本管理：入库带版本，检索带过滤

处理「作废」靠给每个 chunk 打上版本元数据：version（版本号）、effective_at（生效时间）、status（生效/作废），检索时按条件过滤：

def search_current(qdrant, query_vec: list[float], top_k: int = 5):
    return qdrant.search(
        collection="kb",
        vector=query_vec,
        query_filter={
            "must": [{"key": "status", "match": {"value": "active"}}]
        },
        limit=top_k,
    )

这条 status=active 的过滤，就是行政那句「第三章作废了，别看」的程序化表达。用户的问题在语义上和作废版高度相似也没关系，过滤发生在召回之前，旧版本根本没有参赛资格。

实施时有三个坑：

• 版本字段必须在第一天就设计进 schema。后补意味着全量重建一遍打标签，几百万 chunk 的库重建一次是大工程。
• 过滤别只做一层。检索带 status=active，生成端也要校验：模型引用的引用块必须来自当前版本，防的是过滤条件写错或被人绕过。
• 历史版本不能硬删。财务制度类文档，审计要求能回答「2025 年 6 月当时的标准是什么」。作废版本保留在库里但默认过滤，需要时可以显式查历史。

顺带说去重：同一份文档经常从多个入口进来（人工上传、爬虫抓取、OA 同步），重复灌入后同一内容出现 N 份，检索时 top-k 被同一段话霸屏。轻量做法是对 doc_id + chunk hash 做唯一约束，重复灌入直接幂等跳过；跨文档重复（转发、抄袭）用 embedding 相似度检测，相似度超过 0.98 的进人工审核队列。

Embedding 模型升级：双索引蓝绿切换

embedding 模型迟早要升级——效果更好的开源模型隔几个月出一个。这里有个硬约束：两套模型的向量空间完全不通用，维度可能不同（384 维、768 维、1024 维都常见），就算维度凑巧一样，同一个文本在两套模型下的向量也毫无对应关系。新老向量混在一个 collection 里做相似度检索，结果等于随机数。

所以模型升级必然意味着全量重灌，工程问题是怎么在不停机的前提下换。做法是把 collection 命名带版本号，新旧两套并行：

1. 建新 collection：kb_v2（新模型、新维度、全量灌入）；
2. 灌完后用同一个评测集（攒下来的真实 query + 标注）跑离线评测，召回质量达标才放行；
3. 配置里把 collection 名从 kb_v1 切到 kb_v2，相当于一次发布；
4. 观察几天，没问题后下线 kb_v1。

flowchart LR
    A[源文档] --> B[新模型 embedding]
    B --> C[(kb_v2 新索引)]
    D[(kb_v1 旧索引)] --> E{离线评测}
    C --> E
    E -->|达标| F[配置切换指向 v2]
    F --> G[观察期后下线 v1]
    E -->|不达标| H[排查 / 换模型 重灌]

这正是部署领域的蓝绿发布：kb_v1 是绿，kb_v2 是蓝，流量整体切换而不是混着跑。全量灌入期间线上检索一直走旧索引，用户零感知；评测不达标随时切回去，回滚成本是一次配置变更。代价是双倍存储，切换窗口期两个索引并存，磁盘要按 2 倍预留；全量重灌中断后的重复写入，靠灌入逻辑的幂等性兜底。

反馈闭环：让用得不好的人帮你找病灶

前几篇讲过 RAG 评测集，但线上问题光靠离线评测集发现不了，因为用户真实 query 的分布和评测集的分布不一致——评测集是你想出来的问题，用户问的是他们工作中卡住的问题。两个低成本的线上信号源：

• 负反馈：答案卡片上放点踩按钮，点了必须带原因（答案错误 / 引用过期 / 没找到 / 其他）。带原因的负反馈才能直接映射到修复动作：「引用过期」指向版本过滤漏洞，「没找到」指向覆盖率缺口。
• 检索为空的 query：用户问了，库里根本没有相关 chunk。这类 query 是最好的需求清单——每一条都意味着一块知识缺口，按频次排序，高频的优先补文档。

把这两个队列放进周会流程：每周盘点负反馈 top 问题（它们往往指向知识库的系统性缺陷，而不是个案），挖空 query 转成补文档任务。知识库越用越准靠的不是模型升级，是这条反馈链路真的在转。

动手实操

把前文的骨架拼成一个可跑的同步任务，加上蓝绿切换函数。依赖：qdrant-client。

"""文档同步任务骨架：扫描 → hash 比对 → 增量 upsert/delete → 版本过滤
依赖: pip install qdrant-client
"""
import hashlib
from dataclasses import dataclass

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, FieldParams, Filter, MatchValue,
    PointStruct, VectorParams,
)

DIM = 1024                       # 新模型维度，按实际改
COLL_NEW, COLL_OLD = "kb_v2", "kb_v1"

@dataclass
class Chunk:
    doc_id: str
    hash: str                    # md5(doc_id + heading + text)，见正文
    text: str
    version: str

def chunk_hash(doc_id: str, heading: str, text: str) -> str:
    """按内容边界生成稳定 id：没变的段落 hash 不变"""
    return hashlib.md5(f"{doc_id}|{heading}|{text}".encode()).hexdigest()

class VectorStore:
    """Qdrant 上的幂等读写 + 蓝绿切换"""

    def __init__(self, client: QdrantClient, coll: str):
        self.client, self.coll = client, coll
        if not client.collection_exists(coll):
            client.create_collection(
                coll, vectors_config=VectorParams(DIM, Distance.COSINE))

    def upsert(self, chunks: list[Chunk], vecs: list[list[float]]):
        pts = [PointStruct(id=int(c.hash[:16], 16), vector=v,
                           payload={"doc_id": c.doc_id, "text": c.text,
                                    "version": c.version, "status": "active"})
               for c, v in zip(chunks, vecs)]
        self.client.upsert(self.coll, pts)      # 幂等：重复灌入结果不变

    def delete_by_doc(self, doc_id: str):
        # 删除靠 filter：僵尸 chunk 就是漏了这一步
        self.client.delete(self.coll,
            points_selector=Filter(must=[doc_filter(doc_id)]))

    def search(self, vec: list[float], top_k: int = 5):
        # 只召回当前生效版本
        return self.client.search(self.coll, query_vector=vec, limit=top_k,
            query_filter=Filter(must=[FieldCondition(key="status",
                                      match=MatchValue(value="active"))]))

def doc_filter(doc_id: str):
    from qdrant_client.models import FieldCondition
    return FieldCondition(key="doc_id", match=MatchValue(value=doc_id))

def blue_green_switch(client: QdrantClient, eval_fn) -> str:
    """新索引评测达标才切换，返回要启用的 collection 名"""
    new, old = VectorStore(client, COLL_NEW), VectorStore(client, COLL_OLD)
    score = eval_fn(new)                 # 用固定评测集测召回质量
    if score < 0.85:                     # 阈值按业务定
        raise RuntimeError(f"新索引评测不达标: {score:.3f}，保持 {COLL_OLD}")
    return COLL_NEW                      # 达标 → 改配置指向新库，观察后删旧库

关键行说明：chunk_hash 用「内容边界 + 文本」生成稳定 id，文档插入新段落时未受影响的块 hash 不变，天然跳过重算；delete_by_doc 是防僵尸 chunk 的那一步，同步任务里「源里消失的 doc_id」全要走它；search 的 status=active 过滤把作废版本挡在召回之前；blue_green_switch 把「评测达标」变成切换的前置条件，不达标抛异常、旧索引继续服务。

常见误区与小结

• 同步任务只扫新增不扫删除。源文档删了，向量库里全是僵尸 chunk，还会被召回引用——把「这次没出现」也当事件处理。
• chunk_id 由位置序号生成。文档开头插一段，后面全部 id 位移，hash 全变等于全量重灌；主键按内容边界生成。
• 版本元数据后补。上线半年才想到要区分新旧制度，只能全量重建打标签；version/status 第一天就进 schema。
• embedding 升级直接改在线 collection。新旧向量空间不通用，混用等于随机检索；双索引蓝绿切换，评测达标再切。
• 点踩按钮收集不到原因。不带原因的负反馈没法变成修复动作，收集时强制选原因分类。

小结：知识库的日常运维围绕四类生命周期事件展开，核心设施是三件套——稳定主键加 hash 比对支撑增量更新，版本元数据加检索过滤挡住过期知识，双索引蓝绿切换承接 embedding 升级。再到点踩和挖空 query 两条反馈链路，知识库才算从「一次性灌库项目」变成「持续服务」。至此第二批 45-55 的缺口补齐篇全部完成：记忆、选型、检索、反思、运维、成本，这条线补上了 Agent 工程化落地的主要短板。

参考

• Qdrant 官方文档 — Filtering 与 Update 语义项：https://qdrant.tech/documentation/
• LangChain Indexing API — 文档去重与增量清理（index 函数的 RecordManager 机制）：python.langchain.com
• OpenAI Cookbook — Embedded vector re-indexing 示例：https://cookbook.openai.com