RAG 知识库更新失败怎么办？Python + SQLite 实现版本切换、回滚与删除

网页内容更新后重新切块再写入知识库如果中途失败查询可能看到一半旧版本、一半新版本。先删除旧块再导入又会在导入失败时让原本可用的文档消失。本文用 Python 标准库和 SQLite 演示一个最小版本生命周期新块写入、切换可见版本和删除旧块在同一事务内完成。代码已在本地 Python 3.9.6 运行验证了初始导入、异常回滚、成功切换、过期写入拒绝和级联删除。一、先规定查询能看见什么文档实体使用稳定 ID例如 manual版本号表示一次完整内容快照分块身份由文档、版本和块序号共同组成。docs.active 指向当前可见版本查询只返回与它匹配的分块。分块序号不能单独充当长期 ID。正文前面插入一段后后续块的序号可能全部变化。来源 URL、采集时间和内容指纹在真实系统中也应保留本文为了聚焦事务省略了这些字段。二、完整离线示例保存为 rag_version_demo.py使用 python3 rag_version_demo.py 执行。不要使用 -O示例依赖 assert 检查结果。无需安装向量库也不调用模型。importsqlite3 consqlite3.connect(:memory:)con.execute(PRAGMA foreign_keys ON)con.executescript( CREATE TABLE docs(id TEXT PRIMARY KEY, active INTEGER NOT NULL); CREATE TABLE chunks( doc TEXT REFERENCES docs(id) ON DELETE CASCADE, version INTEGER NOT NULL, pos INTEGER NOT NULL, body TEXT NOT NULL, PRIMARY KEY(doc, version, pos)); )defreplace_version(doc,expected,new_version,texts,failFalse):ifnew_versionexpectedornottextsorany(nott.strip()fortintexts):raiseValueError(invalid_version_or_content)withcon:rowcon.execute(SELECT active FROM docs WHERE id?,(doc,)).fetchone()currentrow[0]ifrowelse0ifcurrent!expected:raiseValueError(stale_update)ifrowisNone:con.execute(INSERT INTO docs VALUES (?,0),(doc,))con.executemany(INSERT INTO chunks VALUES (?,?,?,?),[(doc,new_version,i,text)fori,textinenumerate(texts)])iffail:raiseRuntimeError(simulated_failure)con.execute(UPDATE docs SET active? WHERE id?,(new_version,doc))con.execute(DELETE FROM chunks WHERE doc? AND version?,(doc,new_version))defvisible(doc):return[r[0]forrincon.execute( SELECT c.body FROM chunks c JOIN docs d ON c.docd.id WHERE d.id? AND c.versiond.active ORDER BY c.pos,(doc,))]replace_version(manual,0,1,[old A,old B])assertvisible(manual)[old A,old B]try:replace_version(manual,1,2,[new A],failTrue)exceptRuntimeError:passassertvisible(manual)[old A,old B]assertcon.execute(SELECT COUNT(*) FROM chunks WHERE version2).fetchone()[0]0replace_version(manual,1,2,[new A])assertvisible(manual)[new A]try:replace_version(manual,1,3,[stale])exceptValueErroraserror:assertstr(error)stale_updateelse:raiseAssertionError(stale update accepted)withcon:con.execute(DELETE FROM docs WHERE id?,(manual,))assertvisible(manual)[]assertcon.execute(SELECT COUNT(*) FROM chunks).fetchone()[0]0con.close()print(PASS: initial version, rollback, switch, stale rejection, cascading delete)运行输出PASS: initial version, rollback, switch, stale rejection, cascading delete三、失败时为什么还能看到旧内容replace_version 先核对调用方知道的旧版本再写新分块。模拟异常发生在新分块插入后、active 更新前。异常离开连接上下文时触发回滚所以新分块不会残留active 仍指向旧版本。成功路径中active 更新和旧块清理随事务一起提交。这里强调的是同一个 SQLite 数据库内的事务边界不是对两个不同存储服务的分布式原子提交。SQLite 事务说明可查官方文档https://www.sqlite.org/lang_transaction.html 。连接显式开启了 foreign_keys删除 docs 行时ON DELETE CASCADE 删除其分块。外键开关和级联行为参见https://www.sqlite.org/foreignkeys.html 。四、这份测试的边界示例使用单连接、串行操作和内存数据库。expected 检查证明的是顺序执行时能拒绝过期任务不能据此声称跨进程并发更新安全。默认延迟事务下读取版本与开始写事务之间的竞争需要另外设计和验证例如明确事务起点、使用带版本条件的更新并核对影响行数。failTrue 是受控 Python 异常不是进程被杀死或断电实验内存数据库也没有验证重启后的持久化恢复。生产部署还需要真实存储、并发读写和故障注入测试。代码没有向量检索、embedding 或模型回答评估。visible 返回正文分块只用来展示可见版本规则。事务正确不代表召回正确后者仍要用问题、证据和预期答案组成的评测集验证。五、换成外部向量库后怎么落地当元数据在关系库、向量在另一个服务时不要把这段事务代码理解成跨库保证。可以先把新版本作为暂存数据写入验证块数、版本、embedding 配置与检索结果准备完成后再切换可见版本。每次查询都必须落实版本过滤如果先检索再过滤旧版本可能占满候选集让有效结果不足。这需要根据所用向量库能力选择服务端过滤、扩大候选或隔离索引并用回归问题检查。切换后清理旧版本应可重试且幂等。利用持久化任务或 outbox 记录待执行操作定期对账“元数据声明存在的版本”和“向量库实际存在的块”。旧任务再次执行时必须核对版本避免覆盖新内容。删除也是完整链路先确保被撤回文档不可检索再清理分块、向量与相关缓存最后核验引用页面和历史回答的处理策略。短暂抓取失败不能直接等价于原文永久删除应设置可解释的确认状态。从爬虫到 RAG稳定系统需要回答三个问题当前对外提供哪个版本、半途失败如何恢复、旧内容何时不再出现。把这三件事变成可验证的规则再扩展模型和检索配置更容易定位线上问题。本文由 AI 辅助编写代码为本地可复现的机制示例。