构建会感知自身过时的知识库：真值失效检测系统设计

如果一个知识库只是“存了很多内容”，那它本质上仍是一个静态快照：你把它构建出来的那天它是真的，三个月后某条政策废止了、某个接口版本下线了、某个结论被新数据推翻了，知识库根本不知道自己已经错了。这篇文章要讨论的方向，正好戳在这个痛点上——让知识库在它“不再为真”的时候主动意识到，并且提醒你。

这个项目标题“A knowledge base that notices when it stops being true”更像是一种设计目标，而不是某个现成开源包的名称。它的核心诉求是：知识条目不应该只有“新增、查询、删除”三个状态，还应该有“已验证、已过期、疑似失效、待复核”这类生命周期状态；一条底层事实失效时，所有依赖它的上层结论都要跟着被牵连标记。

所以本文不会去编造某个不存在的开源仓库细节，而是从工程角度把这个能力拆开，给出一个可运行的最小原型：用 SQLite 存“事实节点”，用验证器去对比外部信号，用级联失效引擎传播“某条知识过时”的影响，最后通过 FastAPI 把能力暴露成接口。如果你正在做 RAG、企业知识库、内部文档问答或数据资产管理，这篇文章可以先收藏。

1. 知识库“会失效”的能力模型速览

基于标题方向拆解，一个具备“真值保鲜”能力的知识库系统，通常包含以下能力。这里按可验证功能而不是虚构硬件参数来列。

这里重点提醒一下：一个知识库要做到“知道自己何时不再为真”，光靠一个字段是不够的，它至少需要三件事。第一，知道自己的每一条知识是从哪来的、什么时候被验证过；第二，知道不同知识之间谁依赖谁；第三，有一个外部信号源能让系统主动发现“原本成立的事实被推翻了”。后面第三节开始就是围绕这三件事做落地设计。

2. 适用场景与使用边界

2.1 这类系统解决什么问题

最适合这类系统的，是“事实经常变化、错了会造成连锁误判”的知识领域。

• RAG 应用：文档切块后存入向量库，但如果源文档已经过期，LLM 仍会用过期内容回答。给知识库加上失效状态后，检索阶段可以直接过滤掉过期片段，或者在引用时提示“该条目已验证/已过期”。
• 企业内部规范库：接口文档、审计要求、财务政策、安全基线经常更新，如果没有失效传播，下游员工可能继续照旧流程操作。
• API 与依赖信息库：某个 SDK 版本被废弃、某个云服务计费规则调整，如果知识库不自知，推荐给用户的方案就是错的。
• 产品定价与活动库：知识库适合存储有明确时间窗口和引用来源的信息，来源一变，知识状态就该跟着变。

2.2 不适合什么场景

不是所有知识都适合做“真值状态管理”。过于主观、缺少明确外部验证信号的内容，比如市场观点、用户反馈、个人经验帖，很难自动化判断对错，强行建模只会增加维护负担。另外，如果是面向闲聊型对话的常识库，也不适合投入太多失效治理逻辑，因为常识的失效频率很低，维护成本不划算。

2.3 边界与合规意识

如果你要抓取外部网页、订阅第三方文档更新、或使用企业内部资料作为事实源，必须注意几个边界：抓取行为要遵守目标站点的服务条款；企业文档要注意权限控制和数据脱敏；引用外部事实时尽量保留来源 URL 和获取时间。知识库并不创造事实，它只负责记住“某个事实在什么时间、基于什么来源是成立的”。

3. 系统设计与核心概念

要做失效感知，首先要把“知识”建模成可以独立验证的最小单元。一条知识可能是一个短结论，也可能是一段长描述，但我建议用“事实节点”作为基础粒度，而不是把整份文档作为一个知识项。

3.1 事实节点 FactNode

一个事实节点至少包含以下信息。

把这些信息落到 SQLite 后，系统不仅能回答“知识库里有什么”，还能回答“知识库里哪些内容已经不值得相信了”。

3.2 真值状态机

建议把状态设计成如下几种：

• pending：刚录入，尚未验证。
• verified：已验证，当前仍然成立。
• expired：已被外部信号或人工判定为不再成立。
• affected：它本身可能没直接过期，但它依赖的底层事实过期了，导致它需要重新评估。
• deprecated：已废弃，不再参与任何检索与推荐。

状态转换有两个关键路径。第一条路径是“pending → verified”，表示验证器检查通过；第二条路径是“verified → expired”，表示验证器发现外部事实与知识库记录不一致。另一个容易被忽略的路径是“verified → affected”，它代表依赖传播，某条知识没有被直接证伪，但它依赖的根事实已经倒下，它也只能被连坐。

3.3 验证器 Verifier

验证器是这套系统的“感觉器官”。每类知识应该有自己的验证器，比如接口文档型知识可以通过请求线上 OpenAPI 元数据来判断；商品价格型知识可以通过价格接口快照来核对；政策规范型知识可以订阅官方公告页面，用页面哈希变化来触发人工复核。

验证器的接口可以抽象得很简单：输入事实节点，返回是否仍然成立、依据、下次验证时间。

class Verifier(Protocol): def verify(self, fact: FactNode) -> VerificationResult: ...

3.4 依赖传播

依赖传播是最容易忽略也最重要的机制。知识库里的结论很少是孤立的，例如“某 API 的认证方式为 OAuth 2.0”是一条底层事实，如果它变成了 expired，那么“使用该 API 的接入教程”“基于该教程生成的运维手册”都应该被标记为 affected，而不是继续被当作准确知识引用。

实现方式可以做成简单的有向无环图遍历：当一个节点被置为 expired，递归查找所有 depends_on 里包含该节点的上层节点，统一标记为 affected。

3.5 失效检测的三种办法

从工程角度看，验证器发现“知识不再是真”有三种常见办法。

• 定期拉取源端快照：比较最新页面或接口内容和存储的 source_hash，如果哈希不一致，说明源端变化了，需要进一步解析是否影响知识点。
• 订阅 Webhook 或变更事件：如果源站提供 webhook，比如文档站点在内容发布时通知订阅方，系统可以收到事件后主动触发验证。
• 人工巡检入口：不可能所有知识都自动验证。给前端或管理端提供一个“标记为过期”的按钮，再由依赖引擎自动传播。

4. 环境准备与前置条件

这个原型系统不需要 GPU，普通 CPU 环境就能跑。建议的版本和工具如下。

先创建一个独立目录，避免依赖冲突。

mkdir selfaware-kb cd selfaware-kb python -m venv venv

激活虚拟环境后再安装依赖。

# Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate

安装必要的 Python 包。

pip install fastapi "uvicorn[standard]" apscheduler requests

如果之后想接向量检索，再额外装向量数据库的客户端。这里的最小系统先用 SQLite 存储事实节点。

5. 最小原型实现

下面这套代码不是某个现成开源项目的搬运，而是“能感知自己失效的知识库”的最小可运行参考实现。读者可以按自己的数据模型调整。

5.1 项目目录结构

selfaware-kb/ ├── main.py # FastAPI 入口 ├── database.py # SQLite 初始化与连接 ├── models.py # 数据模型定义 ├── validator.py # 验证器与模拟外部信号 ├── dependency.py # 依赖传播 └── scheduler.py # 定时批量验证任务

5.2 数据模型与数据库初始化

用 SQLite 存储事实节点，字段保持精简。

# models.py from datetime import datetime, timezone from pydantic import BaseModel, Field from typing import List, Optional class FactNode(BaseModel): content: str source_url: str = "" source_hash: str = "" status: str = "pending" confidence: float = Field(default=0.5, ge=0.0, le=1.0) category: str = "default" depends_on: List[str] = Field(default_factory=list) check_interval: int = Field(default=86400, ge=3600) verified_at: Optional[datetime] = None expire_reason: str = ""

# database.py import sqlite3 import json DB_PATH = "knowledge.db" def get_connection(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_connection() conn.execute(""" CREATE TABLE IF NOT EXISTS facts ( id TEXT PRIMARY KEY, content TEXT NOT NULL, source_url TEXT, source_hash TEXT, status TEXT, confidence REAL, category TEXT, depends_on TEXT, check_interval INTEGER, verified_at TEXT, expire_reason TEXT, created_at TEXT, updated_at TEXT ) """) conn.commit() conn.close()

这个库表结构里面，depends_on 用 JSON 文本存储依赖节点 ID 列表，可视化或关系型查询方便后，再拆成单独的关联表也不迟。

5.3 依赖传播逻辑

写一个函数：当某个事实节点被判定为 expired，它需要把所有依赖它的节点标记为 affected。

# dependency.py from database import get_connection import json def mark_expired(fact_id: str, reason: str): conn = get_connection() conn.execute( "UPDATE facts SET status='expired', expire_reason=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (reason, fact_id), ) cascade_affected(conn, fact_id) conn.commit() conn.close() def cascade_affected(conn, fact_id: str): rows = conn.execute("SELECT id, depends_on FROM facts").fetchall() target_ids = [fact_id] while target_ids: current = target_ids.pop() for row in rows: deps = json.loads(row["depends_on"] or "[]") if current in deps and row["status"] not in ("expired", "deprecated"): conn.execute( "UPDATE facts SET status='affected', updated_at=CURRENT_TIMESTAMP WHERE id=?", (row["id"],), ) target_ids.append(row["id"])

这段代码是一个广度优先的影响面扩散。一次执行后，不只是第一层引用会受影响，更上层引用也会被连坐。

5.4 验证器实现

为了演示，验证器用一个模拟源端接口。真实场景中，这里会替换成请求具体的第三方 API、网页抓取，或者数据库对比。

# validator.py import hashlib from datetime import datetime, timezone class VerificationResult: def __init__(self, ok: bool, reason: str = ""): self.ok = ok self.reason = reason def fetch_source_hash(fact_content: str) -> str: # 模拟一个外部源端：正常情况下这个函数会访问真实 URL # 这里用 content + 当前时间戳来模拟“源端信号变化” return hashlib.sha256( (fact_content + datetime.now(timezone.utc).strftime("%Y-%m-%d")).encode() ).hexdigest() def verify_fact(fact) -> VerificationResult: # 简化演示：如果当天日期是偶数日，认为源端内容变化 today = datetime.now(timezone.utc).day if today % 2 == 0: return VerificationResult( ok=False, reason="source content changed: simulated external signal mismatch" ) return VerificationResult(ok=True) def verify_and_update(fact_id: str): from database import get_connection conn = get_connection() row = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone() if row is None: conn.close() return None fact = { "id": row["id"], "content": row["content"], "source_url": row["source_url"], "source_hash": row["source_hash"], "status": row["status"], } result = verify_fact(fact) if result.ok: conn.execute( "UPDATE facts SET status='verified', verified_at=?, source_hash=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (datetime.now(timezone.utc).isoformat(), fetch_source_hash(fact["content"]), fact_id), ) else: conn.execute( "UPDATE facts SET status='expired', expire_reason=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (result.reason, fact_id), ) cascade_affected(conn, fact_id) conn.commit() # 重新查回状态 updated = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone() conn.close() return dict(updated)

5.5 FastAPI 入口

将数据库与验证逻辑暴露为 HTTP 接口。

# main.py import uuid import json from datetime import datetime, timezone from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional from database import init_db, get_connection from validator import verify_and_update from dependency import mark_expired app = FastAPI(title="Self-Aware Knowledge Base API") class FactCreate(BaseModel): content: str source_url: str = "" source_hash: str = "" category: str = "default" depends_on: List[str] = [] check_interval: int = 86400 class FactUpdate(BaseModel): status: Optional[str] = None expire_reason: Optional[str] = "" @app.on_event("startup") def on_startup(): init_db() @app.get("/health") def health(): return {"status": "ok"} @app.post("/api/facts") def create_fact(body: FactCreate): conn = get_connection() fact_id = uuid.uuid4().hex[:12] now = datetime.now(timezone.utc).isoformat() conn.execute( """ INSERT INTO facts (id, content, source_url, source_hash, status, confidence, category, depends_on, check_interval, verified_at, expire_reason, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( fact_id, body.content, body.source_url, body.source_hash, "pending", 0.5, body.category, json.dumps(body.depends_on), body.check_interval, None, "", now, now, ), ) conn.commit() row = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone() conn.close() return dict(row) @app.get("/api/facts") def list_facts(category: Optional[str] = None): conn = get_connection() if category: rows = conn.execute("SELECT * FROM facts WHERE category=?", (category,)).fetchall() else: rows = conn.execute("SELECT * FROM facts").fetchall() conn.close() return [dict(r) for r in rows] @app.get("/api/facts/{fact_id}") def get_fact(fact_id: str): conn = get_connection() row = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone() conn.close() if row is None: raise HTTPException(status_code=404, detail="fact not found") return dict(row) @app.post("/api/facts/{fact_id}/verify") def verify_fact_endpoint(fact_id: str): result = verify_and_update(fact_id) if result is None: raise HTTPException(status_code=404, detail="fact not found") return result @app.post("/api/facts/{fact_id}/expire") def expire_fact_endpoint(fact_id: str): from database import get_connection conn = get_connection() row = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone() conn.close() if row is None: raise HTTPException(status_code=404, detail="fact not found") mark_expired(fact_id, "manually marked as expired") return {"result": "ok", "fact_id": fact_id, "action": "expire"}

5.6 启动服务

uvicorn main:app --host 127.0.0.1 --port 8000

启动后访问 Swagger 文档：

http://127.0.0.1:8000/docs

在这个界面上可以直接新增事实、触发验证、查看状态。一个最小可用系统到这里就通了。

6. 真值失效检测机制详解

6.1 定时批量轮询

知识库中的条目不能等用户发起验证才检查，后台需要有一个定时任务扫所有“超过 check_interval 时间未验证”的条目。

# scheduler.py import time from datetime import datetime, timezone from database import get_connection from validator import verify_and_update def run_batch_verification(max_items: int = 20): conn = get_connection() now = datetime.now(timezone.utc).isoformat() rows = conn.execute( """ SELECT id, verified_at, check_interval, status FROM facts WHERE status IN ('verified', 'pending', 'affected') ORDER BY verified_at ASC LIMIT ? """, (max_items,), ).fetchall() conn.close() verified_count = 0 expired_count = 0 for row in rows: # 如果最近验证时间距今超过 check_interval，则触发验证 if row["verified_at"] is None: should_verify = True else: last = datetime.fromisoformat(row["verified_at"]) elapsed = (datetime.now(timezone.utc) - last).total_seconds() should_verify = elapsed >= row["check_interval"] if should_verify: result = verify_and_update(row["id"]) if result["status"] == "verified": verified_count += 1 elif result["status"] == "expired": expired_count += 1 return {"checked": len(rows), "verified": verified_count, "expired": expired_count} def scheduler_loop(): while True: run_batch_verification(max_items=10) time.sleep(3600) if __name__ == "__main__": scheduler_loop()

批量验证应该限制每次扫描数量，避免短时间里把外部源端请求打爆。上面的代码加入了 max_items 参数。

6.2 Webhook/事件订阅

定时轮询有一个缺点，就是发现延迟取决于轮询周期。更主动的方式是提供 Webhook 接收端点，由源端在内容发生变化时通知系统把对应分类的条目标记为 pending，再触发验证。

@app.post("/api/webhooks/source-update") def source_update_webhook(payload: dict): category = payload.get("category", "default") conn = get_connection() rows = conn.execute( "SELECT id FROM facts WHERE category=?", (category,) ).fetchall() conn.close() for row in rows: mark_pending(row["id"]) return {"message": "queued for re-verification", "count": len(rows)}

注意，不是每个源端都提供 Webhook。现实项目里更常见的是“源端页面哈希变化 + 人工确认”，哈希变化只是给系统一个触发验证的信号，最终可以结合页面 diff 给管理员生成待确认工单。

6.3 RAG 检索侧如何过滤过期知识

如果这个知识库要服务于 RAG，最关键的一点是让“失效状态”参与检索前的过滤。向量数据库中每个 chunk 都带上事实节点 ID 和状态，检索时直接过滤掉 status 为 expired/deprecated/affected 的 chunk，或者在 prompt 中额外提示 LLM：“以下知识来自一条已验证但已过期的事实，引用时需谨慎”。

这一步看起来简单，但很多 RAG 应用根本没做。结果就是知识库其实存过正确答案，也标记过错误答案，查询时仍然把这些片段送给模型。知识库能不能“知道自己已经错了”没有意义，除非下游检索流程愿意听它的。

7. 接口 API 与批量任务接入

在最小原型里，已经出现了几个核心接口。下面是常用 API 汇总。

用 curl 做一次完整验证流程：

curl -X POST http://127.0.0.1:8000/api/facts \ -H "Content-Type: application/json" \ -d '{ "content": "API v2 requires OAuth 2.0 client credentials", "source_url": "https://docs.example.com/api-v2", "category": "api-docs", "depends_on": [], "check_interval": 7200 }'

新增后会返回事实 ID，假设返回abc123。触发验证：

curl -X POST http://127.0.0.1:8000/api/facts/abc123/verify

触发人工过期：

curl -X POST http://127.0.0.1:8000/api/facts/abc123/expire

Python 侧的调用示例：

import requests base_url = "http://127.0.0.1:8000" # 新增 r = requests.post(f"{base_url}/api/facts", json={ "content": "RDS MySQL 8.0 将于 2026 年停止小版本更新", "source_url": "https://example.com/rds-mysql-lifecycle", "category": "lifecycle", }) fact_id = r.json()["id"] # 验证 verify_resp = requests.post(f"{base_url}/api/facts/{fact_id}/verify") print(verify_resp.json())

批量任务的思路有点类似队列系统：每天凌晨跑一次批量验证，扫描所有超过验证周期的条目；验证失败的进入一个“失效清单”；失效清单再通过企业微信、钉钉、飞书或邮件机器人推送给维护者。推荐给运行脚本加上日志记录，把每次验证的结果和时间落到另一张验证日志表，方便事后复盘。

import logging logging.basicConfig( filename="verify.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) logger = logging.getLogger(__name__) def run_batch_verification_with_logging(max_items=20): result = run_batch_verification(max_items=max_items) logger.info("batch verification result: %s", result) return result

8. 运行资源与性能观察

这套原型系统不涉及 GPU，普通个人电脑、2 核 4G 的云服务器都能跑。运行时主要占用在三个方向：SQLite 查询、验证器发出的外部请求、以及依赖级联传播的计算量。

单次级联传播需要遍历全表找到依赖关系，如果事实节点数量只有几千条，性能压力很小。但如果到百万级节点，建议把依赖关系改成邻接表，并给 status、category、verified_at 建索引。从工程角度，先避免每次更新都写 status 历史表，验证日志字段保留最后 5 次即可，否则存储会膨胀。

对外部源端的请求一定要做限流。批量验证 1000 条知识，如果每条都请求接口，可能触发源端限流，导致验证结果失真。建议的做法是给同一 category 的条目做聚合验证，或者先比较页面哈希，哈希不变就不逐条验证。

日志观察重点看三个指标：

• 每次批量验证平均耗时。
• 被验证为 expired 的条数比例。
• 从“源端变化”到“知识库标记过期”的延迟。

前两个可以在日志里直接统计。第三个最能体现系统的真实价值：如果源端发布变更后 1 小时内知识库就自动把相关条目标记为过期，说明这套失效感知机制是有效的。

9. 常见问题与排查方法

一个常见误区是只把知识条目标记为“已过期”，却不写过期原因。建议把 expire_reason 做成必填字段，这样可以汇总高频失效原因，反过来优化知识来源的选择。比如你发现某类知识每次都是因为“官方 API 版本更新”而失效，那就应该降低这类知识的 check_interval，或者直接换更稳定的源。

10. 最佳实践与使用建议

10.1 原子化沉淀事实

知识节点不要做得太大。一条包含大量推论的长文很难判断整体是否失效。更好的粒度是“一句话断言 + 来源 + 时间”。如果你需要存储整篇文档，建议在文档内抽取需要失效检查的关键断言节点，再让文档与这些节点关联。

10.2 来源优先于人工记忆

知识库必须记录 source_url 和记录时的快照。当系统产生争议时，能否回溯到原始来源是关键。真正的“知识库会注意到自己何时不再为真”，依赖的是可追溯的外部事实源，而不是一个硬编码的 JSON 文件。

10.3 验证器必须幂等

验证器本质上是一个判断函数：给定一种外部状态，返回是否与该知识匹配。设计成幂等后，定时任务和手动触发可以随意重复调用，不会造成状态错乱。如果验证器本身带副作用，比如请求接口时修改了远端数据，一定要拆到独立模块里。

10.4 从“自动标记过期”到“待人工复核”

机器判断错误会造成误伤。最稳妥的机制是：系统发现源端变化后，不是直接把事实标记为 expired，而是将状态置为 affected 或 pending，并给维护人发一个待确认消息；人工确认后，再执行失效传播。这个设计能避免因为抓取页面模板变动造成的批量误判。

10.5 权限与隐私保护

如果是企业知识库，常见的分类有内部安全策略、客户数据、供应商信息，这些数据的访问必须走权限模型。批量验证日志中也不要记录来源正文，只记录事实 ID、状态、验证耗时等元数据，避免敏感信息泄漏。

10.6 与现有 RAG 管线的集成方式

不要把失效感知做成一个孤岛。实际接入时，通常是在旧知识入库的 Embedding 切片阶段增加 status 字段，在检索阶段用 metadata filter 过滤。你可以先用一个简单的 FastAPI 中间层代理原有的向量检索请求，在其中注入状态过滤逻辑，等验证稳定后再改造原有管线。

11. 扩展方向与收尾建议

从标题“A knowledge base that notices when it stops being true”出发，最核心的可执行结论是：知识库应该是一种带生命周期的系统，而不是一堆静态文件的集合。如果你正打算做企业知识库、RAG 文档库或高质量问答系统，我建议你把“失效感知”设计放进第一版需求，不要等上线后再补。

最容易踩的坑有三个：知识粒度太粗导致单个失效影响面过大；没有给知识记录来验证时间；检索时不读取状态字段。先避开这三个坑，再逐步加入定时批量验证、Webhook 触发、级联失效传播和源端快照比对即可。

第一次落地时，建议选择一个“更新频率高、业务影响大、来源权威明确”的细分场景练手，比如技术团队内部的 SDK 兼容性列表。验证跑通以后，你就会直观看到知识库在源端变化后数分钟内主动标记自己“不再为真”的效果，那时再往更多业务分类推广就容易得多。