搭建好 RAG 知识库并不意味着工程已经结束。随着制度、产品资料和业务规则不断变化,索引中的旧内容可能继续被召回,进而让模型生成已经失效的答案。要让知识库长期可靠,需要同时解决变更检测、过期数据下线和历史版本管理,并把这些能力纳入稳定的同步流程。
前面几篇聊 RAG,重点都在「怎么把知识库搭起来」:分块怎么切、向量库怎么选、效果怎么评估。
但真正跑在生产环境里,会撞上一个更磨人的问题:知识库会过期。
制度改了、产品下架了、报价调整了,索引里还躺着旧版本。用户问出来的答案,比不知道更糟——错得理直气壮。
这篇聊上线之后的工程问题:增量入库、过期下线、版本管理。
先看三个场景,都来自真实项目:
这三种情况指向同一个事实:知识库的准确性是一条下降曲线,不是一次交付就能锁定的状态。
很多团队的处理方式是「定期全量重建」——每周挑个夜里,把整个知识库重新跑一遍。这套方案能跑,但有两个代价:
更好的做法是按变更量处理。核心问题拆成三个:怎么发现变化、怎么处理消失、怎么区分新旧版本。

最直接的方案:给每篇文档算一个内容哈希,入库时一起存下来,下次同步时比对。
import hashlib
def doc_fingerprint(text: str) -> str:
"""内容指纹:文本内容变了,指纹就变"""
normalized = text.strip().replace("rn", "n")
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
def sync_docs(new_docs, index_store, upsert_fn, delete_fn):
"""增量同步:只处理新增和变更,未变的直接跳过"""
stats = {"added": 0, "updated": 0, "unchanged": 0}
for doc in new_docs:
fp = doc_fingerprint(doc["content"])
old = index_store.get(doc["doc_id"])
if old is None:
upsert_fn(doc, fp)
stats["added"] += 1
elif old["content_hash"] != fp:
# 先删旧块,再写新块:避免同一文档留下两代分块
delete_fn(doc["doc_id"])
upsert_fn(doc, fp)
stats["updated"] += 1
else:
stats["unchanged"] += 1 # 大多数文档会走这里
return stats
这段代码的关键在第 3 个分支:未变更的文档直接跳过,不重新分块、不重新 embedding。 实际运行中,这个分支通常覆盖 95% 以上的文档,这才是增量方案的价值所在。
哈希粒度决定了「变更」的判定标准,有两个极端:
| 粒度 | 优点 | 缺点 |
|---|---|---|
| 整篇文档哈希 | 实现简单,块与文档一一对应 | 改一个字,整篇重新 embedding |
| 分块级哈希 | 只重算变化的那几块,最省 | 需要维护「块→文档」映射,块边界一变就全失效 |
大多数文档更新是局部修改(改一段、加一节),分块级哈希更省。但代价是分块边界要稳定——如果分块器版本升级导致切法变化,全部分块的哈希都会变,等于全量重建。
我的建议:先做整篇级哈希跑通,文档量过万或 embedding 成本成为显性支出后,再升级到分块级。 不要在验证阶段就上最复杂的方案。

这一块比增量入库更容易被忽略,也更危险。
增量入库处理的是「变了」,过期下线处理的是「没了」。
问题在于:源文档被删掉之后,向量库里没人会主动通知它。 检索照样命中,模型照样生成,用户照样拿到一个已经作废的答案。
而这类错误比「拒答」严重得多。拒答只是没帮上忙,过期答案是帮了倒忙。
直接物理删除有个风险:如果是源目录临时抽风导致的误判,删掉就找不回来了。更稳的做法是两阶段:
from datetime import datetime, timedelta
def mark_deprecated(index_store, source_doc_ids, grace_days=7):
"""源端已消失的文档,先标记失效,不立刻物理删除"""
now = datetime.now()
for doc_id, meta in index_store.items():
if doc_id in source_doc_ids:
continue
if meta.get("status") == "deprecated":
# 宽限期内一直没回来,才真正清理
marked_at = datetime.fromisoformat(meta["deprecated_at"])
if now - marked_at > timedelta(days=grace_days):
meta["status"] = "deletable"
else:
meta["status"] = "deprecated"
meta["deprecated_at"] = now.isoformat()
def search(query, k=5):
"""检索时过滤掉失效文档"""
results = vector_search(query, k=k * 3)
return [
r for r in results
if r.metadata.get("status") == "active"
][:k]
两个设计点值得说明:
deprecated 状态的文档不参与召回,但向量还留在库里,宽限期内可恢复k * 3),因为过滤会损失结果数量,直接取 top-k 可能导致过滤后不够数有些知识天生带时效性:报价、政策、活动规则。对这类文档,可以在元数据里加两个字段:
effective_date:生效时间expire_date:失效时间(可选)检索时按当前时间过滤:
def search_with_time(query, when, k=5):
results = vector_search(query, k=k * 3)
return [
r for r in results
if r.metadata.get("status") == "active"
and r.metadata.get("effective_date", "1900-01-01") <= when
and when <= r.metadata.get("expire_date", "9999-12-31")
][:k]
这样,旧版本继续留在库里,但不会被召回。保留历史数据 + 按时间过滤,比删除更安全——尤其是需要追溯「当时的规定是什么」的场景。
上面几节用到的字段,汇总成一张表。这套字段是我现在每个知识库项目都会配的基础设施:
| 字段 | 类型 | 用途 |
|---|---|---|
doc_id | string | 源文档唯一标识,增量比对的锚点 |
content_hash | string | 内容指纹,变更检测 |
version | int | 版本号,每次更新 +1 |
status | enum | active / deprecated / deletable |
effective_date | date | 生效时间,检索时过滤 |
expire_date | date | 失效时间(可选) |
source_path | string | 源文件路径,便于溯源和排错 |
updated_at | datetime | 最后更新时间,运维排查用 |
前四个是必需的,后四个按场景加。字段不多,但它们决定了知识库能不能被「管起来」。
把上面的机制串起来,就是一个完整的同步任务。用定时任务每天跑一次:
def daily_sync(source_dir, index_store):
# 1. 扫描源目录,读取当前文档集合
new_docs = load_documents(source_dir)
source_doc_ids = {d["doc_id"] for d in new_docs}
# 2. 增量处理:新增 / 更新 / 跳过未变
stats = sync_docs(new_docs, index_store, upsert_fn, delete_fn)
# 3. 标记消失的文档(进入 deprecated 宽限期)
mark_deprecated(index_store, source_doc_ids)
# 4. 输出变更日志,方便排查
log_sync_result(stats, deprecated=count_deprecated(index_store))
几个工程细节:
1. 向量库的删除是弱项
不同向量库对删除的支持差异很大,有的删除后会有残留、有的需要重建索引才彻底生效。上线前先在自己的向量库上验证一遍删除行为,别假设它和数据库一样干净。
2. 别在同步任务里做全量重建
确认「只有少量文档变更」再决定要不要重建。我见过定时任务里写全量重建的,平时没事,文档量涨上去之后每次跑几小时,最后变成没人敢动的定时炸弹。
3. 新增文档要控制质量闸门
自动同步意味着「源目录里有什么,知识库就吃什么」。源目录里的临时文件、草稿、过期备份,都会被同步进来。建议加一层过滤规则(文件名规则 / 目录白名单),不然知识库会越来越脏。
content_hash + status + effective_date 三个字段是基础,保留历史 + 按时间过滤比删除更安全知识库上线只是开始。真正决定它能不能长期好用的,是这套「持续更新」的机制有没有建起来。
你们的知识库是怎么做更新的?全量重建还是增量同步?有没有踩过「过期文档被检索出来」的坑?评论区交流。
Work Agent、Workflow 和智能体三种 AI 应用形态有什么区别?
Work Agent 的智能工作流是否只是由 LLM 增强的低代码工具?
单 Work Agent 与多智能体应用的工作流跳转和全局意图有何差异?
Python 学习者如何获得机器学习实习机会?
Python ML Engineer 应该学习哪些 CS229 机器学习概念?
Python 学习者如何为机器学习实习培养必备技能?