拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

AI Agent知识管道:RAG从文档解析到向量检索的完整实践

AI Agent知识管道:RAG从文档解析到向量检索的完整实践

前几篇把 Agent 的运行循环、工具调用骨架都铺完了,现在该碰一个更现实的问题:Agent 的知识从哪来?在 AI Agent 体系里,RAG(Retrieval-Augmented Generation,检索增强生成)就是核心的知识获取管道——文档进来走解析、切分、向量化、索引,问答时按相关性召回、重组进上下文。这套流程做得好不好,直接决定 Agent 是真助手还是嘴硬的鹦鹉。下面我逐个环节拆开讲,附带一套可以直接跑起来的代码,适合已经玩过基础 Agent 循环、想给 Agent 接入私有知识库的开发者。

1. 先认清 RAG 在 Agent 体系里的位置

1.1 大模型的边界与知识获取的需求

先从 Agent 的角度说起。大模型本身是一个“语言先验”的存储器,它能聊天、能推理,但它的知识边界在训练完成那天就固定了。你的私有文档、内部 FAQ、实时的项目排期,它一概不知道。你硬问它,它只会一本正经地编。这在 Agent 场景里是致命的——Agent 要完成任务,很多时候恰恰需要这些“不在模型里”的信息。

知识获取管道要解决的,就是把这个信息缺口补上。管道起点是一堆乱七八糟的原始文档(PDF、Excel 里的字段说明、企业微信聊天记录导出、Markdown 笔记),终点是 Agent 能直接拿来用的上下文或工具接口。RAG 是现阶段工程上最务实的选择,因为它不用微调模型、不用改变权重,只靠“检索 + 拼装”就能让模型知道它没学过的东西。

我特别喜欢用一个比喻:把模型想象成一个新入职的员工,脑子聪明但没看过公司档案。RAG 就是一个勤奋的档案管理员,每次开会前先把相关资料抽出来摆到桌面上。员工不需要背下档案,只需要在会上按需翻阅。这也是 RAG 和微调最大的区别——微调是逼员工把档案背进脑子,RAG 是直接递纸。

1.2 RAG 解决的不只是“不知道”

RAG 的价值往往被低估。很多人觉得 RAG 只是解决大模型“知识不够”的问题,实际操作下来你会发现,它同时解决了三个问题:知识过期、知识割裂、可追溯性。

知识过期好理解,训练数据有截止日期,RAG 让模型能读到最新内容。知识割裂是企业场景里更痛的——文档散落在 OA、CRM、ERP 各个系统,字段定义、话术口径经常对不上。RAG 管道相当于把这些割裂源统一成一个语义入口,至少在问答层面对用户隐藏了系统的物理边界。可追溯性呢?RAG 生成答案时引用了具体片段,你可以在 Agent 响应里带上来源标记,出问题能追责。纯靠模型内部的隐式知识回答,这是做不到的。

这三个能力放在 AI Agent 体系里有一个共同作用:降低信任的门槛。Agent 做的很多操作是有风险的,用户敢不敢把任务交给它,取决于 Agent 给的答案能不能溯源。RAG 天然提供了“证据链”,这也是为什么我建议 Agent 的核心问答能力从 RAG 开始搭,而不是从微调开始。

这套能力放到现在的 AI Agent 落地语境里,几乎成了标配。我接触到的项目里,从研发文档问答、客服工单辅助、ERP 单据说明,到销售话术库,背后都是同一条 RAG 管道。区别只是文档源不同、切分策略不同。所以说 RAG 影响的不只是聊天机器人,而是所有“需要结合具体知识做决策”的 Agent 场景。

1.3 核心概念:嵌入、向量、相似度

开始写代码之前,三个基础概念必须先过一遍,不然调参的时候全是玄学。

嵌入(Embedding)是把文本变成一串数字向量。向量是模型理解语义的中间表示,每个维度不是一个明确语义,而是整体上让语义相近的文本,在向量空间里距离更近。中文场景我实测下来推荐 BAAI/bge-m3,1024 维,在中文检索上比很多英文模型友好太多。如果你的环境能调用大模型服务,OpenAI 的 text-embedding-3-small 也够用,但本地化部署优先考虑开源嵌入模型。

相似度计算最常用的是余弦相似度,公式是两个向量内积除以模长乘积。数值越接近 1,表示两个文本语义越接近。向量数据库内部就是靠这个数值做暴力扫描或近似搜索(ANN)的。你如果做过推荐系统,会发现这套逻辑和“用户向量召回商品向量”完全同构,只是召回对象从商品变成了文档片段。

为什么用向量而不是关键词?因为业务问答里,用户的问法和文档原文经常不是一个字面。比如文档写“年假按在职时长折算”,用户问“我工作八年半能休几天假”,字面上几乎没有重合的词,但语义是连着的。关键词检索对这个完全失效,向量检索能跨过这层字面差异。

2. 管道拆解:从文档到上下文的五个环节

2.1 文档解析与清洗

管道第一个环节是解析。你要有心理准备:真实世界的文档远比 PDF 干净样例复杂。PDF 里有扫描件、有分栏、有页眉页脚;Word 里有批注和文本框;Excel 里一格塞一整段。解析的目标是把这些物理格式转成纯文本,同时尽量保留段落结构。

我常用 Python 库组合方案:PDF 优先用 PyMuPDF 或 PyPDFLoader,扫描件必须接 OCR(PaddleOCR 或 Tesseract),Word 用 python-docx,Excel 用 pandas + openpyxl。这里有个经验:宁可解析得慢一点,也要把页眉页脚和页码去掉,不然切分时会制造大量噪声文本,向量库里全是“第 1 页共 23 页”这种垃圾片段。

清洗阶段也别省。全角半角不统一、乱码字符、空行过多的,最好在切分前统一处理。我在项目里写了一个小的预处理函数:去空白字符、合并断行、归一化标点。看似不起眼,但对后面的检索命中率影响极其显著。

2.2 切分策略:chunk 大小和 overlap

切分是整个管道里最考验经验的地方。切得太碎,上下文语义不完整,检索时单片段信息量不足;切得太长,向量语义被稀释,还容易把不相关的内容搅进同一个向量,检索噪声大。

经验区间是 256 到 1024 个 token。token 不是一个汉字一个 token,中文通常一个汉字约等于 0.6 到 1 个 token,取决于分词器。所以如果你用 500 字符作为一个 chunk,实际上是 300-500 个 token,这个量级是合适的。具体项目里我没有上来就定死,而是先看文档结构:按章节标题切、按段落切、按句号切,都比纯固定长度切好。

切分器我用得最多的是 LangChain 的 RecursiveCharacterTextSplitter,它按分隔符优先级递归切,优先保住段落,逼不得已才按字符切。参数里有 chunk_overlap,我建议设成 chunk 大小的 10% 到 20%,作用是让跨 chunk 的语义有缓冲,避免一句话被拦腰截断。注意,overlap 太大等于变相增加存量和 token 消耗,太小又起不到缓冲作用。

这里有一个容易被忽略的点:很多技术类文档的列表项、代码块、表格,切分时要考虑保留整体结构。表格如果被切开,检索回来的片段就是断头的表格,模型根本读不懂。遇到这种文档,我有时候会先做结构感知切分,再对特殊类型做额外标记。

2.3 嵌入与向量索引

切分完的每块文本,通过嵌入模型转成向量。向量维度取决于模型,bge-m3 是 1024 维,OpenAI 的小模型是 1536 维。维度越高,理论上语义表达能力越强,但存储和检索开销也越大。

向量数据库我按项目规模选型。数据量小于 10 万条、单机开发调试,Chroma 和 FAISS 完全够用;生产环境、数据量大、需要分布式和实时更新,我首推 Milvus,其次 Qdrant。对了,很多人会忽略一个细节:向量数据库建索引时有一个 metric_type 参数,默认可能是 L2 距离,但你的嵌入模型在训练时用的可能是余弦相似度。如果两侧不一致,检索结果就会偏差。用 bge 系列模型,通常要确保向量库的 metric 配置为 COSINE。

索引类型也值得说两句。数据量小的时候暴力扫描(FLAT)反而最准;数据量大了才考虑 HNSW 这种近似索引。我见过有人用 HNSW 后召回率掉了一截,一查发现参数 M 和 efConstruction 没调,纯默认值。经验是:先 FLAT 验证效果,再换 HNSW 做性能优化,别一步到位。

2.4 检索:召回、过滤、重排

检索阶段通俗叫“召回”,就是把最相关的几个 chunk 从向量库里捞出来。这一步直接决定模型能看到什么,所以我一直强调,检索不是简单的相似度取 top,而是一个需要精心调参的环节。下面几个参数我逐一说明,每一项都在实际项目里踩过坑。特别提醒,这些参数是耦合的,调整 top_k 时常常需要同步调整分数阈值,只动一个参数往往看不出效果。

  • top_k:返回多少条候选。太小漏信息,太大灌进模型的杂讯多。我默认设 4 到 6,文档复杂时会临时提高到 8。不要因为上下文窗口大就无脑增加,召回太多反而干扰。
  • 相似度阈值:设置分数下限,低于阈值的 chunk 直接丢弃。这个参数能避免模型拿完全不相关的内容硬编答案。具体阈值要实测,不同嵌入模型和文档集分数分布差异很大,别照抄网上的数字。
  • 过滤条件:向量库支持的 metadata 过滤功能非常关键。比如按文档来源、按部门、按时间范围过滤,能大幅缩小检索范围。这相当于给检索加了数据库的 WHERE 条件,在知识库里掺了多个业务域时常备。
  • 重排序:第一轮向量召回可以宽进,比如召回 20 条,再用 cross-encoder 重排序模型逐条打分,取前 4 条。重排序模型对语义对匹配判断更精准,代价是计算量高,但只对几十条候选打分,开销可以接受。bge-reranker-base 就是专门干这个的,效果对比非常明显。

2.5 生成:上下文组装与提示词约束

检索到的片段,最后要组装进大模型提示词。组装有几个细节:

一是上下文的组织顺序。把置信度最高的片段放在靠近问题的位置,或者按检索得分排序,模型对更靠前的信息关注度更高。

二是明确标注来源。我在每条资料前加“[资料1] [资料2]”,提示词要求模型引用时用编号标注,答案可追溯。用户在企业聊天工具上问完问题,点开引用能看到原文,信任感完全不同。

三是指令约束。提示词必须反复强调:只依据提供资料回答,资料不包含答案就明确说不知道,禁止编造。这一步能显著降低幻觉。

还有个容易被忽视的问题:上下文总量与模型窗口的匹配。即便模型有 128K 上下文,塞入全部 chunk 也不明智。我以 token 数动态截断,控制检索内容总量在一个合理范围,比如不超过 3000 token,剩下的空间留给 Agent 的后续规划和工具输出。

3. 实操:搭一个能直接跑的本地 RAG 管道

3.1 环境准备与依赖安装

先列一下环境。Python 3.10+,pip 安装依赖。下面是我实际项目里的 requirements 清单,你复制过去基本能直接用:

langchain>=0.2 langchain-community>=0.2 langchain-chroma>=0.1 chromadb>=0.4 sentence-transformers>=2.2 openai>=1.0 pypdf python-docx unstructured pandas openpyxl

注意 langchain 版本更迭非常快,很多老教程里的接口在新版本里换名字了。比如早期流行的 Chroma.from_documents 现在可能要在 langchain_chroma 包里导入,HuggingFaceEmbeddings 在新版本里也可能被抽到独立的 langchain_huggingface 包里。如果导入时报 ModuleNotFoundError,不要硬刚,查一下当前版本的迁移文档,把 import 路径换掉就行。这个问题我处理过不下十次,解决思路很统一:先看环境里的实际包版本,再对依赖版本做固定,最后再跑导入测试。

3.2 文档加载、切分与嵌入的代码实现

假设我手里有三种类型的企业文档:一份 PDF 版员工手册、一份 Markdown 格式的 API 文档、一份 Excel 格式的客户 FAQ。代码分四步走。

第一步加载。每种格式用各自的 Loader:

from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain_community.document_loaders import UnstructuredExcelLoader pdf_docs = PyPDFLoader("employee_handbook.pdf").load() md_docs = TextLoader("api_docs.md", encoding="utf-8").load() xlsx_docs = UnstructuredExcelLoader("customer_faq.xlsx", mode="elements").load() all_docs = pdf_docs + md_docs + xlsx_docs

UnstructuredExcelLoader 的 mode="elements" 会尽量按单元格拆分,比整表塞进一个 chunk 友好得多。

第二步切分:

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "!", "?", ";", " ", ""], length_function=len, ) chunks = splitter.split_documents(all_docs) print(f"切分后共 {len(chunks)} 个块")

这里我把中文句号、感叹号、问号、分号都放进 separators 列表,目的是让切分优先在中文句子边界断开。默认的 separators 是按英文标点设计的,对中文文本不友好,这个是必须改的坑。

第三步嵌入模型初始化,我用本地模型避免外部 API 依赖:

from langchain_community.embeddings import HuggingFaceEmbeddings embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-m3", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, )

encode_kwargs 里 normalize_embeddings 设为 True,是把向量归一化。之后用余弦相似度时,内积就等于余弦相似度,这个细节能减少一些向量库配置上的歧义。

第四步入库:

from langchain_chroma import Chroma vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./kb_store", collection_name="enterprise_kb", collection_metadata={"hnsw:space": "cosine"}, )

collection_metadata 指定 hnsw space 为 cosine,和上面说的 metric_type 对应上。persist_directory 是本地持久化目录,下次启动直接加载,不用重新嵌入。

3.3 检索 + ReRank 的完整链路

入库完成后,是检索侧的代码设计。我先做向量召回 20 条,再做重排序取前 5 条:

def retrieve_with_rerank(query: str, k_first: int = 20, k_final: int = 5): # 第一轮:向量召回 raw_results = vectorstore.similarity_search_with_score(query, k=k_first) docs = [doc for doc, _ in raw_results] # 第二轮:交叉编码器重排序 from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-base") pairs = [[query, doc.page_content] for doc in docs] scores = reranker.predict(pairs) ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True) return [doc for doc, _ in ranked[:k_final]]

这里有一个取舍。sentence_transformers 的 CrossEncoder 和 HuggingFaceEmbeddings 是不同的库,但都是加载本地模型,磁盘占用稍微多一点,可以接受。如果部署环境太紧张,也可以只用向量检索的 top_k,但要接受语义排序没那么准。

我实测过,把 20 条候选交个重排序模型打分,时间开销大约 0.3 到 0.5 秒,在可接受范围内。相比换来的是更干净的上下文,这笔投资很划算。如果你的接口延迟要求极高,才考虑省掉这步。

3.4 提示词组装与问答调用

最后一步组装上下文并调用生成模型。我以 OpenAI 兼容接口为例,因为很多大模型服务都兼容这个接口,换 base_url 就行:

from openai import OpenAI client = OpenAI( base_url="http://your-llm-endpoint/v1", api_key="local-key", ) def answer(query: str): hits = retrieve_with_rerank(query) context_blocks = [] for i, doc in enumerate(hits, 1): source = doc.metadata.get("source", "未知来源") context_blocks.append(f"[资料{i}] {doc.page_content}\n来源: {source}") context = "\n\n".join(context_blocks) system_prompt = ( "你是企业知识助手。请严格基于给定的资料回答问题。\n" "规则:1. 只能使用资料中的信息;2. 资料不足时直接说明缺少哪些信息;" "3. 每个结论后标注资料编号,方便溯源;4. 禁止编造。" ) resp = client.chat.completions.create( model="qwen-plus", temperature=0.3, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"资料如下:\n{context}\n\n问题:{query}"}, ], ) return resp.choices[0].message.content

temperature 设 0.3,是 RAG 场景里我的习惯值。回答需要忠实于资料,创造性参数调太高容易放飞,调太低又显得机械。0.3 到 0.4 之间是一个平衡点。

我这边实测过一个完整流程:把一份 40 页的员工手册和一份 350 行的 API 文档灌进管道,问“数据库连接串放在哪个配置项?”按 top_k=5 能拿到准确答案并标注资料编号。没做重排序前,同样的问题偶发会召回一份不相干的段落,加上重排序后基本稳定。

4. 把 RAG 接进 Agent:三种集成姿势

4.1 方案一:RAG 作为可调用工具

最简单的方式,是把上面的 answer 函数包成一个工具注册给 Agent。Agent 在规划时判断“这个问题需要查知识库”,就调用这个工具,把返回文本当作工具结果继续推理。

代码如下:

# 伪代码,以你用的 Agent 框架为准 from agents.tools import Tool rag_tool = Tool( name="enterprise_kb_query", description="查询企业内部知识库(员工手册、API文档、客户FAQ等)。当问题涉及公司内部制度、技术资料时使用。", func=answer, )

关键在 description。Agent 是靠工具描述来判断何时调用的,描述写得越精确,工具调用的准确率越高。我见过只写“知识检索”三个字的,Agent 就会在不需要的时候乱调。description 最好包含触发场景、覆盖领域、使用限制。

4.2 方案二:RAG 作为记忆模块

Agent 的多轮对话里,历史信息是分散在不同轮次的。传统做法是把最近几轮历史直接塞进上下文,但窗口有限,早期的关键信息会丢失。RAG 可以充当“长期记忆”角色:每轮对话结束后,把关键信息写入向量库,下一轮开始时检索相关记忆注入上下文。

有一个工程细节大家要注意:对话摘要写入向量库时,要附带时间戳和会话 ID 元数据,检索时按会话过滤,否则多用户的记忆会互相串味。我在一个客服 Agent 项目里就用这种方式实现“老客户复访时自动回忆起上次沟通的诉求”,体验完全不一样。

4.3 方案三:Agentic RAG

最近常聊的 Agentic RAG,是把检索从“一次查询一次召回”升级为“Agent 自主决定检索策略”。比如:先检索一次,发现信息不够,再换关键词检索;或者信息里涉及多条业务线,自动按业务线逐条检索再合并。这本质上让 RAG 有了自主推理能力。

工程实现上,可以用 LangGraph 或 LlamaIndex 等框架,把“检索”“判断”“再检索”编排成 Agent 的步骤节点。我个人的建议是:不要一上来就上 Agentic RAG。只有单轮检索满足不了需求时再升级,比如企业场景里经常需要多跳检索、多文档对比时,值得做。简单 FAQ 场景硬上 Agentic RAG,反而增加延迟和失败率。

再往后走,还有 GraphRAG、Ontology RAG 这些变体,适合更复杂的知识推理场景,但基础管道跑不稳之前,先别碰这些花活。

5. 常见问题与排查技巧实录

5.1 检索召回不准:命中不精准

症状是模型答非所问或强行凑答案。排查顺序:先看召回片段本身,你是想问“数据库密码”却召回了“数据库连接失败案例”,那说明检索链路有问题。我先打印 doc 内容和相似度分数,人眼比对分数和业务相关度。

经验上最常见原因是:1)切分太大导致语义混杂;2)嵌入模型不适合中文;3)没有重排序。逐步排查是大思路。如果分数整体偏低,比如都低于 0.6,可能是文档里术语过多、领域特殊,可以考虑领域适配的嵌入模型或先对文档做术语归一化。

5.2 向量库配置不一致:维度或度量方式对不上

报错常见的有 “query vector dimension mismatch”,就是查询向量维度和库里不一致。多半是你换嵌入模型时没重建索引,或者先后用了不同维度的模型。这个好解决,但容易忽视:如果改了嵌入模型,一定要清库重建,旧的索引文件不能复用。

度量方式不一致更隐蔽。索引建的时候用的是 L2,查询的时候代码里用了余弦,分数表现会说不清楚地怪。这个我在前面提过,collection_metadata 和检索参数的 space 要统一。

还有一个容易踩的:多个人协作时,A 用 bge-m3 建了库,B 用 m3e-base 跑查询,两个模型向量空间完全不可比。团队里最好约定嵌入模型版本,并在向量库名字里带上模型标识。

5.3 知识更新了但 Agent 还是答旧内容

增量更新是个大坑。Chroma 默认是本地持久化目录,你往库里添加新文档就 append,但修改或删除旧文档很麻烦。我建议把 metadata 里的版本号或更新时间作为过滤条件,检索时只召回最新版本。

再进阶一点,我维护了两层结构:热数据(近 30 天变更)和冷数据(历史全量)。查询时优先检索热数据,不存在再查冷数据。引入双库虽复杂,但能显著提升新鲜度体验。

5.4 幻觉排查与评估指标

很多“幻觉”其实来自 RAG 检索环节:模型拿了无关资料开始编。不要只盯着提示词,先确认检索片段是否真的覆盖答案。我的习惯是拿 20 到 50 条典型问题做成测试集,人工标注每条问题的标准答案和期望片段,再用 hit rate(能命中相关片段的百分比)和 MRR(排序质量)两个指标度量,改参数前后对比指标,而不是靠感觉。

这里我分享一个速查表供参考:

现象可能原因排查动作
答案编造但语气笃定检索片段不相关或不足打印召回片段人工检查覆盖度
分数高但内容无关嵌入模型不适配换模型对比测试
多轮对话后答案漂移上下文累积噪声限制注入 token 数,RAG 当记忆
新文档查询不到索引未更新或切分异常检查入库日志和 chunk 内容
同一问题两次答案差异大temperature 过高降到 0.2-0.3

最后说点个人体会。我搭了四五个 RAG 管道之后最大的感受是:RAG 的瓶颈往往不在模型,而在数据治理和链路细节。文档格式乱、切分颗粒度不匹配、指标度量缺失,这些问题每一个都会在 Agent 上线后以用户投诉的形式回来找你。不要一上来追求复杂的 Agentic RAG 或炫酷的重排序,把文档解析、切分、检索评估这些基础环节做扎实,效果往往更明显。切分和召回调参时,每一处改动都要记录前后对比,否则就是靠运气调参。如果你正准备给 Agent 接知识库,我的建议很简单:先把一个管道从文档到答案完整跑通,再谈优化。这条知识获取管道跑通之后,后面的 Agent 能力扩展才有了地基。

返回列表