1. 为什么先聊知识获取管道做 AI Agent 做到第四篇我发现有个坎是绕不过去的你的 Agent 再聪明、工具调度再丝滑如果肚子里的知识是空的它就只能靠大模型的“死记硬背”来撑场面。RAG检索增强生成解决的就是这个问题——把外部知识真正灌进你的 Agent 体系里让它能“先查再答”而不是“瞎编乱答”。网上关于 RAG 的教程确实不少但很多要么停在概念层面要么直接甩一堆 LangChain 代码让你跑通 demo 就完事。这篇我想换个角度从“知识获取管道”这个视角把 RAG 拆开来讲它不是一个单一功能而是一条完整的流水线从文档进来、切分、向量化、存储再到查询时召回、重排、喂给大模型。每一步都有坑也有对应的解法。这篇适合谁看适合已经对 AI Agent 有基本概念、想给自己的 Agent 加入“私有知识”能力的开发者也适合那些听说过 RAG、但一直没搞清楚它内部到底在做什么的读者。我会尽量把原理讲清楚也会给出可复现的实操代码让你看完之后能直接搭出一条最小可用的知识管道。2. 整体思路RAG 在 Agent 体系里的定位2.1 从“记忆”这个角度理解 Agent我们先把 Agent 想象成一个刚入职的员工。这个大模型本身就相当于一个“名校毕业的聪明人”——逻辑能力强、常识丰富但对你公司的具体业务一无所知。你不可能让它靠“常识”去回答你公司内部的问题它不知道你们的文档格式、历史数据、项目细节。那怎么办两种思路第一种把所有知识都塞进它的“脑子里”——微调。但这个方案成本高、周期长而且知识一更新又要重新训并不适合动态变化的业务知识。第二种给它配一个“资料库”让它每次回答之前先去翻资料——这就是RAG的思路。这两种不是对立关系我见过不少生产项目是两者结合的。但对大多数从 0 到 1 搭 Agent 的场景来说RAG 是性价比最高的起步方案——你不需要昂贵的训练资源只要把文档管好、检索做好就能让 Agent 具备“知道你们公司内部知识”的能力。2.2 为什么说 RAG 是一条“管道”你如果只是跟着 QuickStart 跑一遍很容易觉得 RAG 就是“把文档向量化→存起来→查一下”。但一旦你要把 RAG 接入到真实的 Agent 系统里就会发现它不是单个节点而是一条流水线文档加载 → 切分 → 嵌入向量化 → 存入向量库 ↓ 用户提问 → 查询改写/扩展 → 检索召回 → 重排序 → 组装上下文 → 大模型生成前一段叫索引管道Indexing Pipeline后一段叫检索管道Retrieval Pipeline。两个阶段加起来才是完整的知识获取管道。我在设计 Agent 时习惯把这条管道抽象为一个独立的服务Agent 的逻辑层不直接操作向量库而是调用一个“检索接口”。这样做的原因很实际知识管道的调优频率远高于 Agent 的逻辑调整分开治理改检索策略时不需要动 Agent 主流程反之亦然。2.3 一个生活化的类比图书馆管理员为了把 RAG 的行为模式讲清楚我常用图书馆来打比方文档加载与切分把装箱的书拆开整理成一个个可以单独借阅的章节。嵌入向量化给每段内容做“索引标签”标签描述的是这段话的语义而不只是关键词。向量库一排排带标签的书架存的是这些“语义标签”。召回读者提问管理员先根据问题语义去找最像的几个“标签”。重排序管理员把找来的一堆候选段落再精挑细选一遍选出真正对题的几段。大模型生成管理员把这几段内容摆在桌上让一个擅长写作的助手据此写答案并要求它注明出处。这个类比虽然不是百分之百精确但能帮你抓住 RAG 的核心思想不直接问大模型而是让它“带着参考资料”来回答。3. 核心细节解析索引管道的每个环节3.1 文档加载先把格式问题处理好你实际会碰到的文档远远不止“纯文本”这一种。可能是 PDF、Word、Markdown、HTML甚至可能是一堆扫描件。每个格式都有不同的解析方案。我在实际项目中处理 PDF 的经验是文本型 PDF用PyMuPDF即fitz或pypdf提取文本速度快效果稳定。扫描型 PDF需要 OCR。Tesseract是开源方案但中文识别效果一般实际我更多用 PaddleOCR 或商业 OCR 服务的接口。这一步容易被低估很多“检索不到内容”的坑其实源头在 OCR 质量差。Word 文档python-docx可以读.docx但.doc老格式需要先转成.docx或直接用 LibreOffice 命令行批量转换。一个容易被忽略的环节元数据保留。我在切分与向量化之前会先把每个文档的基本信息文件名、章节号、页码、来源链接、更新时间提取出来后面检索时可以按来源过滤也能在 Agent 回答时带上引用来源。没有这一步Agent 回答时想“注明出处”就很被动。3.2 切分RAG 里最值得花时间调的参数文档加载完接下来要切分。很多初学者在这里直接套默认参数然后发现检索结果一塌糊涂。切分本身没有绝对正确答案但有一些规律可以参考。切分粒度切得太小每段缺乏上下文检索到的片段可能语义不完整切得太大向量化后语义被稀释且超出模型输入限制。我用下来比较稳妥的起始参数是chunk_size500~800个 token、chunk_overlap50~100。切分策略无脑按字符数是最后的办法。更好的方式是“按结构切”比如 Markdown 按标题切、Python 代码按类/函数切、PDF 按段落切。LangChain 的RecursiveCharacterTextSplitter支持传一组分隔符列表会尽量在更合理的位置断开。父子切分Parent-Child Chunking这是我自己做知识库时比较推荐的一种策略——把文档切两层父层级保留较大块比如一段文章子层级切成小片段比如一句或一小段。检索时用小子片段去命中但喂给大模型时把命中的父层级一起带上。这样既保留了语义检索的“精确性”又保证了上下文完整。3.3 嵌入向量化选对模型比调参更重要嵌入模型的选择直接影响检索质量而且这个影响比多数人以为的更大。我用过几类模型简单对比一下模型语言能力维度场景建议OpenAI text-embedding-3-small中英皆可1536英文为主、调用 API 方便OpenAI text-embedding-3-large更好但更贵3072对精度要求高、预算充足BGE-M3 / BGE-embedding中文效果好1024中文知识库首选M3E / GTE 系列中文场景友好768中文文档多、离线部署首选我的经验是如果业务以中文为主优先考虑国产开源嵌入模型不要一上来就推荐团队接 OpenAI。原因不只是成本而是中文语义在通用英文模型下表达得不够细腻尤其涉及专业术语时效果差异肉眼可见。嵌入模型选定后有个小细节把它固定成一个版本不要频繁换。因为一旦换了嵌入模型向量的坐标分布整体变化之前向量化的数据全部需要重新索引。这个坑我踩过换模型一时爽全库重跑一下午。3.4 向量库选型不是越重越好向量库负责存储和检索向量。市面上选择很多我的选型原则是“按阶段选”练手/原型阶段Chroma或FAISS都够用。Chroma 可以本地跑、支持持久化代码量小FAISS 本质是纯向量索引库不提供完整数据库能力但检索速度极快、内存友好。单体应用阶段直接用 PostgreSQL pgvector插件。你的业务数据本来就在库里加一个扩展就能当向量库用省去维护两套系统。我做过不少企业项目这个方案最务实。大规模独立服务阶段Milvus或Qdrant支持高并发、水平扩展、混合检索。到这个阶段时你的 RAG 已经是核心链路了值得上专业基础设施。不要一上来就上重型方案。我见过团队连 demo 都没跑通就引入分布式向量数据库最后运维成本远超收益。先让管道跑起来再按瓶颈演进。4. 实操过程从零搭一条最小可用的索引管道4.1 环境准备我下面用 Python LangChain 组合来做示例配合开源的 BGE 嵌入模型和 Chroma 向量库实现一套完全离线可用的 RAG 基础管道。这样你不需要申请 API key、不需要外网调用也能把整条链路跑通。需要安装的依赖pip install langchain langchain-community langchain-text-splitters pip install chromadb sentence-transformers pip install pymupdflangchain-community是 LangChain 的社区集成包里面包含大量文档加载器和向量库适配器。sentence-transformers用来加载本地嵌入模型。4.2 加载文档并保留元数据以下载一份 Markdown 文档为例假设叫knowledge_base.md加载到 LangChain 的Document结构中from langchain_community.document_loaders import TextLoader from langchain_core.documents import Document raw_docs TextLoader(knowledge_base.md).load() # 为文档补充元数据 for i, doc in enumerate(raw_docs): doc.metadata[source] knowledge_base.md doc.metadata[doc_id] fkb-{i} doc.metadata[updated_at] 2024-06-01 print(len(raw_docs), raw_docs[0].metadata)这一步虽然简单但元数据是后续过滤、溯源的基础。如果你处理的是 PDF 或 Word建议在加载时同步记录页码和章节号后面聊到“按来源过滤”和“引用标注”时会非常有用。4.3 结构化切分与父子块策略这里我用 LangChain 的RecursiveCharacterTextSplitter实现父子切分from langchain_text_splitters import RecursiveCharacterTextSplitter # 父块较大粒度保留上下文 parent_splitter RecursiveCharacterTextSplitter( chunk_size1200, chunk_overlap150, separators[\n\n, \n, 。, , , ., , ], ) # 子块较小粒度用于精确检索 child_splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n, 。, , , ., , ], ) parent_docs parent_splitter.split_documents(raw_docs) # 为每个子块打上所属父块的引用信息 child_docs [] for p_idx, parent in enumerate(parent_docs): children child_splitter.split_documents([parent]) for c in children: c.metadata[parent_id] p_idx c.metadata[parent_content] parent.page_content c.metadata.update(parent.metadata) child_docs.append(c) print(f父块数量: {len(parent_docs)}, 子块数量: {len(child_docs)})说明一下为什么分隔符列表里要放中文标点RecursiveCharacterTextSplitter会按顺序尝试这些分隔符尽量在第一个可能的位置断开。对中文文档来说按句子边界句号、问号、感叹号切分的效果要远好于按字符硬切。为什么子块要存parent_content这又是实操时的一个关键选择。检索命中子块后我们需要把完整上下文一并取回来。如果到时候再从父块 ID 去向量库二次查询不仅慢还可能因为数据删改导致查不到。直接把父块内容冗余存进子块元数据用空间换效率对中小型知识库来说非常划算。4.4 本地嵌入模型与向量化入库嵌入部分我用 BGE-M3 的轻量版模型加载from langchain_community.embeddings import HuggingFaceBgeEmbeddings model_name BAAI/bge-small-zh-v1.5 encode_kwargs {normalize_embeddings: True} embeddings HuggingFaceBgeEmbeddings( model_namemodel_name, encode_kwargsencode_kwargs, )这里有两个细节值得说明normalize_embeddingsTrue表示对向量做归一化。归一化之后内积等价于余弦相似度不同查询之间拿到的分数有更好的可比性后续你如果要设阈值过滤低相关结果归一化会帮你省掉很多麻烦。BGE 系列模型支持中文查询指令前缀query instruction中文场景对查询侧和文档侧使用不同的前缀模板效果会更好。在 LangChain 里可以通过query_instruction参数传入如果你在非 LangChain 环境下用sentence-transformers手动编码记得查询和文档两个方向分别用为这个句子生成表示以用于检索相关文章和空串。向量库我选 Chroma代码最为直接from langchain_community.vectorstores import Chroma vectorstore Chroma.from_documents( documentschild_docs, embeddingembeddings, persist_directory./kb_chroma_db, collection_namekb_collection, )注意我这里存的是child_docs但每个子块元数据里已经包含了父块内容。检索时命中的是子块取回来时直接用其parent_content字段即可拿到完整上下文。如果你想把 Chroma 换成其他向量库LangChain 的VectorStore接口基本是统一的替换成本不高。这也是我现阶段建议“先用 LangChain 这类框架把管道跑通”的原因——后面演进时你的业务代码不需要推倒重写。4.5 检索侧召回、重排序与上下文组装索引管道搭好后我们写一个检索函数。它内部做三件事向量召回、重排序、组装上下文。向量召回部分直接基于相似度查询retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 8}, )重排序这一步基础版本可以先用一个轻量级的“互信息重排序”思路对召回的 8 段结果再计算一次它们与查询的语义相关性保留最相关的 3~4 段。生产环境我建议用bge-reranker或Cohere Rerank效果会更好。最小可用版本里也可以直接用向量得分排序后截断hits vectorstore.similarity_search_with_score(query, k8) # 过滤低分结果得分是距离越小越相关 filtered [(doc, score) for doc, score in hits if score 0.5] filtered.sort(keylambda x: x[1]) top_docs [doc for doc, _ in filtered[:4]]这里0.5是经验阈值具体值要看你嵌入模型的分布建议先用少量实际查询样本观察一下分数区间再定。阈值不是越严越好太严会召回不到太松会把无关内容交给大模型白白增加幻觉概率。上下文组装则是把命中的子块对应的父块内容拼起来context_parts [] for doc in top_docs: context_parts.append(doc.metadata[parent_content]) context \n\n---\n\n.join(context_parts)到这里你已经有一个可用的知识管道了。把它封装为一个函数def retrieve_context(query: str) - str: hits vectorstore.similarity_search_with_score(query, k8) filtered [(doc, score) for doc, score in hits if score 0.5] filtered.sort(keylambda x: x[1]) top_docs [doc for doc, _ in filtered[:4]] if not top_docs: return return \n\n---\n\n.join( doc.metadata[parent_content] for doc in top_docs )这个函数就是 Agent 的“知识接口”Agent 收到用户问题先调用retrieve_context(query)再把context与用户问题一起组装进 Prompt提交给大模型。5. 实操过程把知识管道接入 Agent 的完整链路5.1 组装 Prompt告诉大模型如何使用参考资料检索到内容只是第一步更关键的是要让大模型“正确地使用”这些内容。我在刚做 RAG 时犯过一个典型错误直接把检索内容拼进 Prompt没有给引导结果大模型要么忽略参考资料自己编要么一字不改地背资料。现在我在 Prompt 里会明确做三件事告诉模型只能基于提供的知识内容回答。如果知识内容不足必须说“未找到相关信息”而不是编造。回答时标注引用来源来自哪份文档、哪个章节。示例 Prompt 模板PROMPT_TEMPLATE 你是一个企业内部知识助手。请基于以下“参考资料”回答问题。 参考资料 {context} 要求 1. 如果参考资料中没有足够的信息请直接回答“未找到相关信息”不要推测。 2. 使用参考资料中的信息回答时请用[1][2]等标注引用来源编号。 3. 参考资料可能包含多条内容请综合归纳不要只抄其中一段。 用户问题{question} 组装完成后的完整调用链路可以用大模型提供商提供的 SDK 直接调用。以 OpenAI 兼容接口为例from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1) # 本地部署的兼容服务 def ask_agent(query: str) - str: context retrieve_context(query) if not context: return 未找到相关信息 prompt PROMPT_TEMPLATE.format(contextcontext, questionquery) resp client.chat.completions.create( modelqwen2.5-7b-instruct, messages[{role: user, content: prompt}], temperature0.2, ) return resp.choices[0].message.content这里把temperature调低到 0.2是因为知识问答场景要求“稳定忠实”而不是“天马行空”。如果是闲聊型 Agent可以调高一些。5.2 何时需要“查询改写”与“多跳检索”如果你的 Agent 遇到的问题是“用户提问很模糊”或者“用户的问题需要多个事实组合才能回答”单次检索就很难满足。这时代入一个简单的查询改写Query Rewriting会很有用先让大模型把用户问题改写成更利于检索的若干个子问题再分别去检索。举一个实际例子用户问“上季度华东区销售为什么下降”这个 query 太抽象直接检索可能什么都召不回。改写后可以拆成几个子问题“上季度华东区销售额数据”“华东区销售下降原因分析”“上季度销售相关政策”。每个子问题分别检索合并结果再回答。这个能力在 LangChain 社区里有现成的MultiQueryRetriever模块但理解背后的思路更重要前期的检索并不完美那就用“多次检索结果合并”来兜底。这个思路也是后续 Agentic RAG 里“Agent 自主决定查什么”的雏形。5.3 离线部署与 API 调用的取舍我上面在嵌入和检索部分都选了开源离线方案大模型部分则用了本地 OpenAI 兼容服务。这个选型逻辑是整套系统不依赖外网数据不出内网。如果你的团队允许调用外部 API嵌入模型和大模型都可以换成云服务代码结构不用改。但我要提醒的是上生产前务必确认数据的隐私合规要求。很多企业做知识库时的潜在风险不在于技术难而在于“谁在用、数据流到了哪里”没有提前设计好。5.4 从“单段召回”到“引用溯源”的工程化封装当我们把 RAG 从“跑通 demo”推向“Agent 正式功能”我强烈建议把检索结果连同引用信息一起返回。前面我们已经在子块元数据里存了source字段这里正好派上用场def retrieve_context_with_citations(query: str): hits vectorstore.similarity_search_with_score(query, k8) filtered [(doc, score) for doc, score in hits if score 0.5] filtered.sort(keylambda x: x[1]) top_docs [doc for doc, _ in filtered[:4]] contexts [] citations [] for i, doc in enumerate(top_docs, start1): contexts.append(f[{i}] {doc.metadata[parent_content]}) citations.append({ id: i, source: doc.metadata.get(source, unknown), doc_id: doc.metadata.get(doc_id, ), }) return \n\n---\n\n.join(contexts), citations在 Agent 的返回结果里附上citations前端可以展示来源链接后端可以追踪知识使用情况。这一步的价值不仅是用户体验更是知识库后续更新优化时的追踪依据——你能知道哪份文档被频繁引用、哪份文档从未被命中。6. 常见问题与排查技巧实录6.1 检索结果质量差先定位是“哪一段”出了问题RAG 出错很多人第一反应是调向量库参数但真正的源头往往更靠前。我的排查顺序是看文档解析质量加载后的文本有没有乱码、缺字、多余空格这一步如果错了后面所有优化都白搭。看切分质量随机打印 20 个子块看看有没有一句话被切成两半上下文是不是断在不该断的位置看嵌入质量拿几个典型 query 直接跑检索看召回的 Top5 是不是和问题语义相关。这一步能判断是嵌入模型的问题还是后续过滤阈值的问题。看重排效果如果 Top5 里混着无关文档但 Top10 里有正确内容那问题在重排/截断策略。看 Prompt 组装检索到的内容如果是对的但最终答案不对那大概率是 Prompt 引导不到位比如没有强调“只能基于参考资料回答”。我整理成一个速查表方便你快速对号入座现象可能原因优先检查项检索召回完全无关嵌入模型与业务语言不匹配换中文专用嵌入模型相关内容在 Top10 有但 Top3 里没出现召回数量不足或重排逻辑缺失调大k值、加重新排序步骤文档内容乱了、缺页解析环节问题换解析库检查 OCR 质量语义被切成碎片、上下文断裂切分策略问题调大 chunk_size或采用父子切分检索没问题答案还是凭空捏造Prompt 引导弱强化 Prompt 约束加入引用标注检索耗时过高向量库数据量过大、无索引升级索引类型如 HNSW或迁移到专业向量库6.2 知识库更新后检索不到新内容这个问题的根源多数是增量索引没做。Chroma 这类库在from_documents时是整体写入如果知识库文件变化了需要重新做增量更新。我的做法是维护一个“文档变更清单”每次有文档新增或修改时按 doc_id 查重后只更新变化的部分def upsert_documents(vectorstore, docs, embeddings): existing_ids set() # 从向量库中读取已有的 doc_id all_data vectorstore.get() if doc_id in all_data: existing_ids set(all_data[doc_id]) new_docs [] for doc in docs: if doc.metadata.get(doc_id) not in existing_ids: new_docs.append(doc) if new_docs: vectorstore.add_documents(new_docs)增量更新还有个隐藏好处避免全量重算向量大幅降低更新成本。如果你的知识库每天都有新文档这个机制迟早要上。6.3 “模型会编造出处”这件事很多初学者以为有了 RAG 就不会有幻觉但实际加入引用功能后发现模型可能给你造一个根本不存在的出处。这个问题的根源是模型在“自由发挥”引用编号。一个有效但常被忽略的解法是不要在 Prompt 里让模型自己编编号而是在组装上下文时就强制给每个片段编号并且把编号与来源绑定contexts_with_idx [ f[{i}] {content} for i, content in enumerate(context_parts, 1) ] # Prompt 中明确写只能使用参考资料中给出的 [n] 标注同时做一层程序化校验模型输出的引用编号必须存在于我们返回的 citations 列表中否则就认为该条引用无效重新生成或直接提示用户。6.4 向量库数据量增长后的性能劣化Chroma 在数据量几千条时飞快但到了几十万条时不加索引会明显变慢。Chroma 默认会创建 HNSW 索引但参数可能不是最优。如果你开始感觉到检索变慢检查chroma_db目录下的索引配置。在 Chroma 客户端中可以通过collection.modify调整hnsw:space内积/余弦/欧式和hnsw:ef_construction、hnsw:M等参数。余弦场景建议spacecosine内积归一化后效果也接近。对个人知识库规模的场景其实不太会走到这一步先不用过度优化。等数据量明显变大时再考虑迁移到pgvector或Milvus那个阶段才需要专门关注索引参数。6.5 多源知识之间的“打架”实践中我遇到过知识库里存了两份文档一份写“A 产品的保修期为 1 年”另一份写“A 产品的保修期为 2 年”检索时两段都召回了大模型容易“选择性失明”只挑一条答或者两头摇摆。这种情况没有银弹但有三个缓解策略检索阶段做来源过滤如果 Agent 知道用户问的具体产品型号就在检索时加元数据过滤条件只查该产品相关文档。生成阶段做冲突提示在 Prompt 中加入“如果参考资料中存在相互矛盾的内容请明确指出”让模型把冲突暴露出来而不是强行掩盖。知识库治理层面建立文档优先级/有效期机制低优先级或过期的文档在索引时就被过滤掉。这个需要业务方配合但长期来看是最根本的解。我之前做企业知识库时第一条策略使用最多效果也最直接。元数据从文档加载环节就保留好的价值在这里就体现出来了。7. 我的一点经验总结如果让我用一句话概括 RAG 的学习曲线概念半天能懂效果要磨一个月。很多人以为 RAG 是“搭好就能用”实际上它更像一个需要持续调优的管道工程。文档怎么切、模型怎么选、向量库怎么存、查询怎么召每一个环节的小改动都可能在最终答案质量上放大。我自己走过一轮弯路之后现在做 RAG 项目的固定动作是先拿 50 条真实业务问题当“评测集”把管道各环节跑一遍记录每条问题的召回情况和最终答案再逐步调优。不要凭感觉调参用数据说话。知识获取管道只是 Agent 能力拼图的一块但这一块直接影响用户对整个 Agent 的信任度——一个回答带正确引用的 Agent和一个“一本正经胡说八道”的 Agent体验完全是两个世界。下一篇我会继续沿着 Agent 的能力拼图往下走聊聊工作流编排相关的经验。基础打得越扎实后面做复杂点的东西才不慌。
