从文档到知识:为AI Agent构建可管理可追踪的RAG知识库
做到阶段 4.1 的时候我发现自己之前对“RAG 知识库”的理解确实太浅了。前几个阶段Agent 的规划、记忆、能力封装都跑通了等到了要给 Agent 接真正能上生产的知识库时才发现“把一堆 Word/PDF 扔进向量库”这个做法在 Demo 阶段完全没问题一旦牵扯到多人协作、内容更新、效果追责就彻底撑不住。所以这篇文章记录的是我在“为 AI Agent 构建可管理、可追踪的 RAG 知识库”这个阶段里的完整思路和最终的落地方案内容包括Agent 场景比问答场景为什么更需要管理追踪、整体架构和技术选型怎么做、元数据该设计哪些字段、Word 和 PDF 怎么解析、切块怎么切以及基于 FastAPI LangChain pgvector 的最小闭环代码。适合正在做 AI Agent、想把 RAG 从玩具做到能用的同学参考。1. 为什么 Agent 的知识库比“问答知识库”难搞1.1 普通 RAG 与 Agentic RAG 的本质区别先放一个结论如果知识库只是给人做问答那召回质量差一点用户最多重新问一次或者换一种问法。但知识库是给 AI Agent 用的时候情况就完全不一样了。Agent 的典型工作方式是“规划—调用工具—观察结果—再规划”如果第一步检索召回的知识就是错的或者过时的Agent 不会停下来问人它会沿着错误上下文继续往下执行把错误放大在后续的工具调用和推理里。我做过一个自动化运维 Agent需要检索历史变更记录来辅助生成变更方案。知识库里混入了一个已经回滚的旧版本变更描述Agent 把这条旧数据当成有效事实基于它生成了新的变更计划。最后排查了半天问题发现根源不是模型幻觉而是知识库检索源本身“带毒”。这就是 Agent 场景下知识库必须可追踪的第一个原因出问题时你得能顺着 Agent 的决策链路一路回溯到“它到底读了哪一篇文档、哪一个段落”。第二个原因是管理上的。企业知识库不是静态的制度会变、产品文档会更新、不同角色会往库里传内容。问答场景里更新慢一点还能忍但 Agent 场景里知识库直接参与业务决策旧版本残留、权限越界、内容归属混乱这些问题都会被成倍放大。1.2 不可管理、不可追踪会踩的四个典型坑这几年我见过太多团队把 RAG 知识库做成“能跑就万岁”结果上线后接二连三出问题。总结下来不可管理、不可追踪的知识库必然踩这四个坑旧版残留导致知识打架文档更新后旧的向量块还留在库里检索的时候新旧内容混在一起返回。Agent 可能前半句引用新制度后半句引用旧制度回答出来逻辑自洽但事实错误。污染源难定位Agent 回答错误时你分不清是模型幻觉、检索召回不对、还是知识源本身有问题。没有检索日志和引用记录就只能靠猜效率极低。权限失控不同密级的文档混在一个库里Agent 一次检索把所有内容都捞回来机密信息被拼进上下文这在企业环境里是合规事故。效果无法评估知识库更新之后召回效果到底是变好了还是变差了没有历史追踪记录全靠主观感觉优化也没有方向。这四个坑我在前几个阶段基本都踩过。阶段 4.1 的设计目标就是把这些坑在架构层面堵死而不是靠人去小心。2. 整体设计从“文档扔进库”到“可管理的知识管道”2.1 分层架构像出版社管稿子而不是往仓库扔纸原先很多“文档扔进库”的做法本质上是把知识库当成一个只进不出的仓库。管理知识库更合理的参照系是出版社管稿子稿子有作者、有来源、有版本、有校对记录、有最终发行版。只有按这个标准来管理AI Agent 才能拿到“经过校验而且版本正确”的内容。我把整套知识库拆成六层数据源层本地文件、Confluence、SharePoint、数据库、外部 API 等。采集解析层拉取文档、识别格式、解析 Word/PDF/HTML。处理切块层清洗文本、结构拆分、切块、嵌入向量。索引存储层pgvector 存向量和元数据配合 SQL 过滤。检索服务层用 REST API 暴露检索能力支持过滤、评分、引用返回。审计追踪层记录入库事件和每次检索调用供回溯和评估。每一层只干一件事层与层之间通过标准的数据结构传递。这样一个直接的好处是底层换解析器、中间换切块策略、上层换嵌入模型都不会牵连其他地方。比如我今天用 PyMuPDF明天发现某个供应商的文档必须用专属解析器只需要改采集解析层检索层完全无感。2.2 技术选型为什么我选了 pgvector LangChain FastAPI向量存储方案很多FAISS、Chroma、Milvus、Qdrant、pgvector。我最终选了 pgvector理由是它在“可管理、可追踪”这个需求上优势太明显了。第一事务支持。文档入库和元数据更新必须是原子的不能出现“向量写进去了、元数据没写进去”这种半截状态。pgvector 在 PostgreSQL 里天然支持事务。第二和业务数据同库。Agent 检索知识的同时可能还要查业务表一个库就能搞定没必要多维护一套向量数据库。第三元数据过滤能力强。可管理知识库必然要按版本、权限、来源过滤pgvector 可以结合普通 SQL where 条件做过滤这一点比很多纯向量库都顺。第四运维成本低。团队本来就有 PostgreSQL不需要额外引入分布式向量集群。当然如果数据量到了几千万级以后再迁移 Milvus 也不迟要预留好接口抽象。编排层用 LangChain 还是自研我的建议是阶段 4.1 用 LangChain 生态先把链路跑通重点是 Document、TextSplitter、Embeddings 这些抽象可以替换不至于被绑死。流程控制这块后续做 Agent 编排时再上 LangGraph跟 FastAPI LangChain 的路线是自洽的。服务层用 FastAPI理由很直接异步性能好、自动生成 OpenAPI 文档、Python 生态兼容好。这样整个项目走的是 FastAPI LangChain LangGraph RAG pgvector 的路线链路统一不会在中间夹一个另一套技术栈的系统。2.3 和 Dify、n8n 这类平台的边界在哪里很多团队一开始会考虑 Dify 或者 n8n这很正常。它们做原型演示非常快Dify 自带知识库管道甚至内置了切块和召回体验。但我的经验是如果你做的是“嵌入到自有 Agent 流程中的知识库”并且要求块级血缘追踪、精细权限控制、对接自己的审计系统低代码平台会很别扭。热词里搜“dify知识库元数据无法过滤”的人很多这确实是真实痛点。平台方案为了通用性元数据体系往往做了简化你很难在块级别自定义几十个字段。而且平台里的知识库和外部 Agent 编排系统之间通常只有标准 API很难做到深度的追踪埋点。所以阶段 4.1 我选择自建管道Dify 那一套可以学习它的交互设计思路但底层一定要掌握在自己手里。3. 元数据设计可管理、可追踪的核心3.1 给每条向量一个“身份证”一个知识从“原始文档”到“可检索的向量”中间经历了很多步骤。可管理的核心在于每一步都记录它从哪来、属于哪个版本、处于什么状态。我常用的 chunks 表字段是下面这样的字段类型说明chunk_iduuid块 ID全局唯一doc_iduuid所属文档 IDdoc_versionint文档版本号source_filetext源文件路径/名称source_typetextpdf/word/markdown/htmlpage_noint页码或章节序号section_pathtext文档内结构路径parent_chunk_iduuid父块 ID用于父子块溯源permissiontext权限标记contenttext原始文本content_hashtext内容哈希用于去重和变更检测embeddingvector向量statustextactive/superseded/deletedcreated_attimestamp入库时间updated_attimestamp更新时间这个“身份证”的设计原则是凡是能回答“这段知识是哪来的、属于哪一版、谁能看”的字段都不嫌多。比如 permission 字段看起来只是个字符串但它在检索时是硬性过滤条件决定了 Agent 有没有资格把这段知识带进上下文。再比如 section_path它记录了块在原文中的层级位置追责时可以定位到“第几章节第几小节”而不是只给你一段孤零零的文本。3.2 数据血缘与版本追踪soft delete 优于物理删除数据血缘简单说就是 document → chunk → embedding 的调用链。一辆车从生产到交付有车辆识别码知识从文档到向量也要有完整的血缘。我的做法是当一个文档的新版本入库时不直接物理删除旧块而是把旧块的 status 置为 superseded并让新版本的 doc_id 指向同一个逻辑文档。举个例子doc_id 是 D001v1 版本产出了块 C1/C2/C3v2 版本产出了块 C4/C5/C6。检索时 SQL 写成这样WHERE doc_id D001 AND doc_version ( SELECT max(doc_version) FROM chunks WHERE doc_id D001 AND status active )这样既保证 Agent 只检索最新版本又保留了历史版本。为什么要保留旧数据因为企业场景里“旧数据”很多时候不能删需要审计。比如新制度发布了但某个决策是依据旧制度做出的追踪系统必须能复现当时的上下文。soft delete 的代价很低只是更新几个字段但它让整个系统的可追溯性上升一个量级。3.3 追踪“谁用了”比追踪“知识本身”更有价值知识库有版本管理还不够Agent 的调用同样要有记录。我建了一张 retrieval_logs 表专门记录每次检索字段说明log_id日志 IDagent_session_idAgent 会话 IDquery本次检索原始 queryrewritten_query改写后的 query如果用了 query rewritechunk_ids命中的块 ID 列表scores对应的相似度得分filter_snapshot检索时的过滤条件快照model_answerAgent 最终回答created_at记录时间这里面最关键的是 filter_snapshot。为什么因为出问题时你要还原现场。比如某个用户本不该看到某段知识却因为权限标签配错了被检索出来有了过滤条件快照就能快速判断是元数据问题还是检索逻辑问题。再往后这些日志还能构成离线评估集定期回放历史 query看知识库升级之后召回效果有没有真正变好。4. 文档解析和切块决定知识库的下限4.1 Word/PDF 解析的实战选择热词里有人搜“ai知识库怎么解析word和pdf”我太理解这个痛点了。我的结论是不同类型的文档必须用不同的解析器不要想一把梭。PDF 文本型文档首选 PyMuPDFfitz速度快对大多数排版文档效果不错。碰到复杂表格可以考虑 pdfplumber它能把表格结构还原得更好但速度会慢不少。扫描版 PDF必须先 OCR 再解析。推荐 PaddleOCR中文效果很好Tesseract 免费但对中文精度稍差。没有 OCR 这一步扫描件解析出来就是一堆乱码。Word 文档python-docx 适合标准 docx如果是老版 .doc先用 LibreOffice 转成 docx再用 python-docx。HTML / Markdown直接用 BeautifulSoup / markdown 库转文本然后按标题结构切分。企业知识库里最常见的是 PDF 和 Word 混排我的建议是解析层写一个 dispatcher按扩展名和 MIME 类型把文档分发到不同解析器解析结果统一成一个 Document 结构包含 source_file、page、content、structure 这些字段。这样后续切块完全不关心来源是什么格式。4.2 切块策略先结构后递归配合父子块溯源切块直接影响召回质量我用了三步组合拳先按结构切。如果文档是 Markdown、HTML 或带标题的 Word先按标题层级把文档切成“顶层级块”保留 section_path。这一步能避免从段落中间生硬切断。再按递归字符切。如果某个顶层块仍然太长用 RecursiveCharacterTextSplitter 按分隔符优先级递归切。常见的 chunk_size 我设在 800chunk_overlap 设在 100。这个值取决于嵌入模型OpenAI text-embedding-3-small 和 bge-m3 这类模型800 到 1000 token 是常见范围。保留父子块关系。顶层块作为 parent_chunk后续切出来的小块作为 child_chunk记录 parent_chunk_id。检索命中子块时可以把父块一起返回避免上下文信息不足。关于 chunk_size 为什么不能太大也不能太小太大检索精度下降而且一次塞给 LLM 的噪音多浪费 token太小语义不完整检索可能召回一段没有上下文的半句话。overlap 一般取 10% 到 20%解决切分边界切断语义的问题。切块这里有个容易被忽略的点每个切出来的块必须带着 provenance也就是 section_path、page_no、parent_chunk_id 这些字段。否则你虽然切好了但“追踪”无从谈起后续返工成本很高。4.3 嵌入模型选择与维度问题嵌入模型直接影响向量质量。我目前常用的是 BAAI/bge-m31024 维中文效果好开源可私有化部署。如果团队已经在用 OpenAI用 text-embedding-3-small 也可以1536 维。这里要特别提醒一个坑向量维度在数据库表 DDL 里是写死的以后换模型如果维度变了就得重建表或者做迁移。所以选嵌入模型时要想清楚不要把维度定在一个很冷门的模型上。维度也不是越大越好。维度越高存储成本越大检索速度越慢。通用经验是中文场景选 1024 维的 bge-m3或者按团队基础设施选 1536 维的 OpenAI embedding都是比较稳妥的选择。5. 实操用 FastAPI LangChain pgvector 落地一套最小闭环5.1 数据库表设计先建 PostgreSQL 扩展和数据表。这里我把核心 DDL 贴出来读者可以直接基于它改造CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE documents ( doc_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), source_file TEXT NOT NULL, source_type TEXT NOT NULL, doc_version INT NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT active, meta JSONB NOT NULL DEFAULT {}, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), UNIQUE(source_file, doc_version) ); CREATE TABLE chunks ( chunk_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), doc_id UUID NOT NULL REFERENCES documents(doc_id), doc_version INT NOT NULL, parent_chunk_id UUID, section_path TEXT, page_no INT, content TEXT NOT NULL, content_hash TEXT NOT NULL, embedding vector(1024), permission TEXT NOT NULL DEFAULT public, status TEXT NOT NULL DEFAULT active, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops); CREATE INDEX ON chunks (doc_id, doc_version, status); CREATE INDEX ON chunks (permission);这里注意几点embedding 的维度必须和嵌入模型匹配bge-m3 是 1024 维HNSW 索引适合中等数据量的相似度检索另外我还建了复合索引专门配合版本和权限过滤。索引不能建太多否则写入会变慢这三个是我测下来性价比最高的组合。retrieval_logs 表的 DDL 我前面列过字段这里不再重复实际建表时加上基础索引即可。5.2 入库管道解析、切块、嵌入、写库下面是一段核心入库流程的简化代码。PDF 解析我用 PyMuPDF切块用 LangChain 的 RecursiveCharacterTextSplitter嵌入用 bge-m3import hashlib import fitz from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceBgeEmbeddings embedding_model HuggingFaceBgeEmbeddings( model_nameBAAI/bge-m3, encode_kwargs{normalize_embeddings: True}, ) text_splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap100, separators[\n## , \n### , \n\n, \n, 。, ., ], ) def process_pdf(file_path): doc fitz.open(file_path) pages [] for page in doc: pages.append(page.get_text()) full_text \n.join(pages) chunks text_splitter.split_text(full_text) return chunks def index_document(file_path, doc_id, doc_version, permissionpublic): chunks process_pdf(file_path) for i, chunk_data in enumerate(chunks): embedding embedding_model.embed_query(chunk_data) content_hash hashlib.sha256(chunk_data.encode(utf-8)).hexdigest() # 这里的 insert 函数需要自己实现 # 把 chunk_data、embedding、doc_id、doc_version、permission、content_hash # 写入 chunks 表并保持 statusactive代码省略了数据库写入部分但核心思路很清楚先解析再切块然后逐条嵌入并写库。我建议入库前先按 content_hash 查一下如果库里已有相同哈希的 active 块就跳过嵌入能省不少模型调用费用。这个对增量更新特别有用。5.3 检索接口过滤、召回、日志、引用返回检索接口是整个知识库对外的窗口我用 FastAPI 暴露核心逻辑是四步组装过滤条件、执行向量检索、写检索日志、返回带引用信息的结果。app.post(/retrieve) async def retrieve(query: str, session_id: str, permission: str public, top_k: int 5): query_embedding embedding_model.embed_query(query) sql SELECT chunk_id, content, section_path, page_no, doc_id, doc_version, 1 - (embedding :emb) AS score FROM chunks WHERE status active AND permission :permission AND doc_version ( SELECT max(doc_version) FROM chunks c2 WHERE c2.doc_id chunks.doc_id AND c2.status active ) ORDER BY embedding :emb LIMIT :top_k # 这里省略数据库驱动执行细节 # 执行查询后把结果写入 retrieval_logs # 返回结构里必须包含 chunk_id / doc_id / doc_version / section_path / score return { session_id: session_id, citations: results, }这段检索逻辑里有几个关键点。第一用运算符计算余弦距离1 减去距离得到相似度范围在 0 到 1 之间。第二在 SQL 里直接过滤 status、permission、doc_version而不是把全部向量捞回内存再过滤这一条决定了元数据过滤不会失效。第三返回结果带完整的引用信息Agent 拿到后可以把它作为可验证的 citation 拼进提示词后续也能用于审计。提醒一句HNSW 索引配合 where 条件做 ANN 检索是可行的但过滤条件命中数据量太小时查询可能会退化成扫描速度变慢。这就是很多人遇到的“加了元数据过滤后检索变慢”的根源属于正常现象需要通过索引设计和数据分布来优化。6. 常见问题与排查技巧实录6.1 元数据过滤失效或变慢热词里搜“dify知识库元数据无法过滤”的人不少自建管道也会遇到类似问题。症状主要有两种一种是加了WHERE permission xxx后返回结果不准另一种是查询突然变得特别慢。排查思路是打开EXPLAIN ANALYZE看执行计划确认是不是走了 HNSW 索引。如果过滤条件稳定可以建复合索引比如(permission, doc_version, status)。还有一个小技巧如果某个权限组的数据量极少而另一个权限组的数据量巨大可以考虑对高频过滤条件建 partial index。6.2 切块边界切断语义问答开始胡编症状是检索出的某段内容读起来像半句话Agent 因此胡编。解决办法按优先级排调大 chunk_overlap从 10% 提到 20%先按结构切标题、段落再递归切不要一上来就按字符硬切命中子块时把父块内容一起返回这就是父子块机制表格类内容单独处理整表作为一个 chunk不要把一个表格的行拆散。我在实际项目中因为表格被切散吃过很大的亏。一个产品参数表被切成好几块Agent 检索到其中一行就乱编整表内容。后来我把表格统一转成 Markdown 格式文本整表作为一个 chunk问题立刻消失。6.3 PDF 表格和扫描件怎么办PDF 表格用 PyMuPDF 提取经常错乱列对不齐、跨页断裂。建议企业文档先用 pdfplumber 的 extract_table 专门解析表格再把表格转成 Markdown 格式的文本块入库例如“| 参数 | 值 |”这种形式这样 LLM 理解起来也容易。扫描件必须 OCR这一步不能省。PaddleOCR 对中文效果好Tesseract 免费但对中文精度一般。建议对扫描件先做 OCR 生成文本层然后再走正常的解析流程。如果扫描件量大可以把 OCR 做成异步任务避免阻塞入库管道。6.4 版本更新后旧块残留的隐患如果没有做 doc_version 过滤v1 和 v2 的块会同时被检索到Agent 可能一半内容用新版、一半用旧版回答出来自己都察觉不到矛盾。解决方案是双管齐下检索 SQL 里强制doc_version (SELECT max(doc_version) ... WHERE statusactive)同时在更新版本时先把新版本写入并 active再异步把旧版本标记为 superseded。这样能避免中间时间窗里检索不到任何版本。千万别直接先删旧再写新中间窗口一查一个空Agent 会直接懵。6.5 多轮对话的检索怎么设计热词里有人问“rag多轮对话怎么设计”在 Agent 场景里这个问题其实分两层。第一层是对话历史里的指代消解。用户说“那它呢”如果不改写上下文直接拿这句话去做检索召回一定不准。做法是先用 LLM 把问题改写成一个包含上下文信息的独立问题再去做检索。第二层是检索上下文管理。每次检索用改写后的独立问题而不是把整段历史拼进去做向量检索否则向量会“淹没”在噪音里。同时 retrieval_logs 要记录 rewritten_query出了问题才能判断是不是改写环节坏了。最后再分享一个我自己的习惯。这套知识库上线之后我会定期抽出 retrieval_logs 里得分在 0.7 以下、但 Agent 最终回答又被用户采纳的案例人工过一遍。你会发现很多问题不是模型不行而是知识库在切块、版本、权限、召回顺序这些地基环节出了问题。RAG 知识库做到“可管理、可追踪”价值不在于仪式感而在于每一个错误你都能定位、能回放、能修复。这个能力才是从 Demo 到生产环境的真正分水岭。