☰
RAG与Agent知识获取管道:索引、检索到生成的全流程解析与工程实践
2026/9/29 18:51:49 网站建设 项目流程

1. 知识获取管道:为什么 Agent 绕不开 RAG

1.1 先聊聊“知识获取管道”这个词

如果你正在从 0 到 1 搭 AI Agent,大概率已经被 RAG 这个词轰炸过。但你有没有想过,为什么 Agent 需要一条“知识获取管道”?我自己的理解是:Agent 本质上是一个“会调用工具的对话系统”,它要回答问题、要执行任务,前提是得有知识。而知识放在哪里、怎么取出来、取完怎么用,这就是管道要解决的事。

RAG 的全称是 Retrieval-Augmented Generation,检索增强生成。核心思路不复杂:先从一个外部知识库里检索出和当前问题相关的片段,再把片段拼进 Prompt 交给大模型,最后生成回答。听起来就像开卷考试——先翻书,再答题。比起让模型凭空发挥,开卷意味着答案有出处、可溯源、能更新,这正是企业场景最需要的东西。

我见过很多团队在 Agent 落地时遇到的尴尬:模型能力选得很强,工具也接了一大堆,但一问到内部制度、产品参数、历史故障记录,就开始一本正经地编。这不是模型不行,而是它缺少一条把“正确答案”送进上下文的管道。RAG 解决的就是这个缺口,它负责让 Agent 在生成前获得“必要的上下文”,而不是靠模型死记硬背。

1.2 为什么 Agent 场景下 RAG 尤其重要

有人会问:大模型上下文窗口已经能做到百万 token 了,是不是不需要 RAG 了?我的回答是:窗口大不等于知识对,更不等于成本对。每轮对话塞进大量无关 token,推理速度变慢,费用成倍上涨,而且关键信息依然可能淹没在噪声里。

RAG 在 Agent 场景里承担的其实是“外置记忆”职能。Agent 的工作流一般包含记忆、规划、工具调用三个核心模块,其中记忆分短期和长期:短期就是当前会话的上下文,长期则必须落到外部存储。RAG 就是把长期知识变成可检索记忆的管道。没有它,Agent 只能靠会话历史里的信息存活,一换会话就“失忆”。

另外,很多团队的私有知识有一个特点:更新频繁且格式杂乱。制度文件隔三差五改一版,产品文档散落在多个 Wiki,数据库里有成百上千条商品信息。如果把知识全部拿去微调模型,一周都跑不完一次,而且知识一变又要重来。RAG 的优势在于知识独立存储、按需注入,更新知识只需要改索引,不需要重新训练模型。

1.3 知识获取管道的四种常见路线

抛开 RAG,其实业界还有几种知识获取方式,我列个对比,方便你理解为什么 RAG 是 Agent 初期的首选:

方式优点缺点适合场景
上下文直接拼接实现简单受限于窗口长度,成本高少量固定知识
微调模型回答原生训练成本高,知识更新慢特定风格、领域语言
RAG更新快、可溯源检索质量决定上限动态知识、私有知识
知识图谱关系推理强构建复杂强关联、多跳问答

我在实际开发中通常是这么组合的:RAG 作为基础管道负责事实类问答,微调只用来调整模型说话风格或者领域术语习惯,知识图谱则留到需要多跳推理时再引入。初学者不要一上来就全都要,先把 RAG 跑通,再逐步叠加。

2. 整体架构拆解:索引、检索、生成三段式

2.1 三段式到底各干什么

RAG 系统不管怎么变,核心骨架永远是三段:索引阶段(Indexing)、检索阶段(Retrieval)、生成阶段(Generation)。我习惯把整个管道画成流水线,每一段都有一个明确输出:

索引阶段输入是原始文档,输出是“向量索引 + 元数据”。过程包括格式解析、清洗、切块、向量化、落库。这个阶段是离用户最远的,却决定了整个系统的上限,文档切得差、embedding 选得差,后面检索和生成怎么补救都有限。

检索阶段输入是用户 query,输出是“一组相关文档片段”。过程包括 query 向量化、相似度搜索、重排序、后处理。这里的核心矛盾是召回率和精度的平衡:召得太多,上下文被无关内容污染;召得太少,正确答案根本没进来。

生成阶段输入是“用户原始问题 + 检索到的片段 + 系统指令”,输出是最终答案。这个阶段看似最简单,其实藏着很多细节:Prompt 怎么写、引用来源怎么标注、模型拒绝回答时怎么办、片段的顺序怎么排,都会影响回答质量。

2.2 索引阶段:文档处理才是真正的重头戏

很多教程把 RAG 讲得高大上,但落到代码,第一步其实是写文本处理器。因为企业里的文档往往不是干净的 markdown,它有 PDF、Word、表格、扫描件,有页眉页脚,有目录,有截图。你如果把整本《员工手册》当作一段文本直接 embedding,那检索效果会惨不忍睹。

我做索引时的通用步骤是:抽文本 -> 清理页眉页脚和不相关元素 -> 按结构标记进行语义切块 -> 对每个块做 embedding -> 写入带元数据的向量库。注意,切块不能只按固定字符数硬切,要优先保留语义完整性。比如 markdown 文档可以按标题层级切,HTML 可以按标签切,PDF 可以先还原阅读顺序再切。

关于 chunk 大小,我给一个经验范围:512 token 到 800 token 之间通常表现比较稳。太小了上下文碎片化,太大了噪声增加。但这不是绝对的,我会在后面专门写一个调优实验。

2.3 检索阶段:向量检索不是唯一答案

基础检索方案是向量检索,也就是把用户 query 过一遍 embedding,然后在向量库里做最近邻搜索。它的优势在于处理语义相似,比如用户问“报销流程”,文档里写“差旅费用如何申报”,字面不同但语义一致,向量检索能兜住。

但向量检索有个老毛病:对专有名词和精确 ID 不友好。用户问“我的订单号是 2025A001”,向量检索可能会因为拼写细微差别而翻车。所以我现在几乎不会只用向量,而是混着来:向量召回一批 + BM25 关键词召回一批,再用重排序模型把两边合并后的结果重新打分。混合检索的效果在知识库内容重复度高、术语密集的场景里,提升非常明显。

重排序(Rerank)是很容易被忽略但性价比极高的一步。向量召回 top 50,Rerank 压缩到 top 5,准确率往往比直接向量 top 5 高出一截。代价就是多一次模型推理,好在有各种轻量 reranker 可选。

2.4 生成阶段:Prompt 编排决定答案质感

生成阶段的输入通常叫“增强提示”(Augmented Prompt)。我这边生产环境的模板大概是:系统角色设定 + 检索片段列表 + 用户问题。其中片段列表按相关度降序排列,每条前加来源标识,比如 [1]、[2],方便模型引用。

这段模板设计有几个坑要提:第一,不要把所有片段不加筛选地塞进去;第二,要在 Prompt 里明确告诉模型“如果片段里没有答案,就说不知道,不要编”;第三,如果场景要求回复格式统一,最好在 Prompt 里给一个样例。这些细节看起来简单,但直接影响问答体验。

3. 实操:用 Python 搭一条最小可运行的 RAG 管道

3.1 技术选型:轻量起步,不搞重型框架

这一节我直接给出一个可以本地跑起来的最小实现。选型方面不追求大而全,够演示核心流程就行:Embedding 用兼容 OpenAI API 的接口,向量库用 Chroma,模型调用也走兼容 OpenAI 的接口。这样一套组合的好处是代码干净,便于初学者看清楚每一环到底发生了什么。

先准备环境:

pip install chromadb openai

然后准备一个 JSON 格式的知识库,模拟企业内部产品文档。实际中你可能要写十几个解析器来兼容不同文档格式,但最小版本先用最方便的结构:

[ {"id": 1, "title": "产品A介绍", "content": "产品A是企业级AI助手平台,支持知识库问答、工作流编排……"}, {"id": 2, "title": "产品B介绍", "content": "产品B是文档协同工具,支持多人实时编辑……"} ]

3.2 文本切块:不让语义断在中间

切块是第一个关键细节。我的建议是:先按章节标题、段落、句子逐级切割,然后再补长度控制。下面这段代码用了一个很朴素的递归切分思路,适合快速理解:

def split_text(text: str, max_chars: int = 500, overlap: int = 50): # 先按段落拆 paragraphs = [p.strip() for p in text.split("\n") if p.strip()] chunks = [] current = "" for para in paragraphs: if len(current) + len(para) <= max_chars: current += para + "\n" else: if current: chunks.append(current.strip()) # 处理超长段落:按句子继续切 if len(para) > max_chars: sentences = para.replace("。", "。\n").split("\n") buf = "" for sent in sentences: if len(buf) + len(sent) <= max_chars: buf += sent else: if buf: chunks.append(buf.strip()) buf = sent if buf: chunks.append(buf.strip()) current = "" else: current = para + "\n" if current: chunks.append(current.strip()) # 合并过小的尾部 chunk merged = [] buffer = "" for c in chunks: if len(buffer) + len(c) <= max_chars: buffer += c + "\n" else: if buffer: merged.append(buffer.strip()) buffer = c if buffer: merged.append(buffer.strip()) return merged

这段代码不追求极致性能,但它体现了一个重要原则:切块要尊重文本的自然边界。你把代码里的 max_chars 换成 500,overlap 在这里没真正用上,但你可以在生产环境实现时加上滑窗重叠,让相邻 chunk 之间保留少量重复内容,避免信息恰好被切断。

实际生产环境我建议直接看 LangChain 的 RecursiveCharacterTextSplitter 或者 LlamaIndex 的 MarkdownNodeParser,它们处理多种分隔符和结构化文档更成熟。自己写一遍主要是为了理解切块背后的取舍。

3.3 向量化与入库:batch 是个好习惯

向量化这步很简单,但有一个容易踩的点:不要逐条请求 Embedding 接口,要走批量请求,速度和费用都有差别。下面用 OpenAI 兼容接口写入 Chroma:

from openai import OpenAI import chromadb client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY") chroma_client = chromadb.PersistentClient(path="./kb_store") collection = chroma_client.get_or_create_collection( name="product_kb", metadata={"hnsw:space": "cosine"} ) def embed_texts(texts: list[str]) -> list[list[float]]: resp = client.embeddings.create( model="text-embedding-model", input=texts ) # 按输入顺序取向量 ordered = sorted(resp.data, key=lambda x: x.index) return [item.embedding for item in ordered] chunks = [] metadatas = [] ids = [] for doc in knowledge_base: doc_chunks = split_text(doc["content"]) for idx, chunk in enumerate(doc_chunks): chunks.append(chunk) metadatas.append({"doc_id": doc["id"], "title": doc["title"], "chunk_index": idx}) ids.append(f"{doc['id']}_{idx}") # 分批入库,每批 32 条 for start in range(0, len(chunks), 32): batch = chunks[start:start + 32] batch_meta = metadatas[start:start + 32] batch_ids = ids[start:start + 32] batch_vecs = embed_texts(batch) collection.add(ids=batch_ids, embeddings=batch_vecs, documents=batch, metadatas=batch_meta)

有几个细节要注意。向量库的相似度度量我用了 cosine,因为它对 embedding 向量长度不敏感,适合大多数开源中文 embedding 模型。collection 的 metadata 里设了hnsw:space: cosine,这会直接影响底层的最近邻检索算法。如果你后面发现检索结果明显不对,先检查是不是空间类型设置错了。

3.4 检索与生成:最小闭环

检索环节,我用同一个 Embedding 接口把用户 query 向量化,然后在向量库里搜 top_k:

def search(query: str, top_k: int = 5): q_vec = embed_texts([query])[0] res = collection.query( query_embeddings=[q_vec], n_results=top_k, include=["documents", "metadatas", "distances"] ) docs = res["documents"][0] metas = res["metadatas"][0] scores = res["distances"][0] return list(zip(docs, metas, scores)) results = search("如何创建智能体?") for doc, meta, score in results: print(f"[{score:.4f}] [{meta['title']}] {doc[:80]}...")

拿到片段后,组装 Prompt 再请求大模型。这一段是整条管道真正产生价值的地方:

def answer_question(question: str): hits = search(question, top_k=5) if not hits: return "没有检索到相关信息。" context = "\n\n".join( f"[{idx + 1}](来源:{meta['title']})\n{doc}" for idx, (doc, meta, _) in enumerate(hits) ) prompt = f"""你是企业知识助手。请根据下面提供的资料回答用户问题。 要求: 1. 回答内容必须基于资料,不要编造; 2. 如果资料中没有答案,请明确说“根据现有资料无法回答”; 3. 在回复末尾列出引用来源编号。 资料: {context} 用户问题:{question} """ resp = client.chat.completions.create( model="chat-model", messages=[ {"role": "system", "content": "你是严谨的企业知识助手。"}, {"role": "user", "content": prompt} ], temperature=0.2 ) return resp.choices[0].message.content

为什么 temperature 要设 0.2?因为知识问答场景要求稳定和忠实,过高的 temperature 会让模型发挥过头甚至编造。如果模型支持,我还建议开启 json_mode 或者设置 logprobs,方便后续做更精细的控制。

3.5 完整效果验证

我自己跑这个最小管道时,往知识库里放了四条产品文档,然后连续问了几个问题。其中“如何创建智能体”能从嵌入文档里正确召回片段,回答结构完整;而“今年春节放假安排”这种知识库外的问题,模型会直接说“根据现有资料无法回答”,没有瞎编。

这套最小闭环虽然简陋,但它把 RAG 的完整链路跑通了。你在自己项目里升级时,无非就是在每个环节替换成更鲁棒的组件:文档解析换成 Unstructured,切块换成 LangChain 的生产级 Splitter,向量库换成 Milvus,检索加上 Rerank,Prompt 加上更多指令约束。

4. 检索质量调优:从“能跑”到“好用”

4.1 chunk 大小的影响:一个对比实验

我拿一份约 8000 字的内部 FAQ 文档做过实验,用同一套向量库和相同问题集,对比了三种 chunk 策略。结果如下:

策略召回命中率回答主观质量平均上下文 token
128 token 硬切差碎片化严重约 300
512 token 语义切块较好完整、准确约 800
1024 token 语义切块中等噪声增多约 1300

这个实验说明三件事:第一,chunk 太小,一条知识被拦腰切断,检索到了但信息不全;第二,chunk 太大,无关内容混进来,模型容易被带偏;第三,没有绝对最优的 chunk 大小,它和你的文档结构、embedding 模型、问题粒度都相关。所以我在项目里总会留一个配置项,方便针对不同知识库单独调。

4.2 不只是 top_k:评估召回质量的两个指标

很多人上线 RAG 后只凭“感觉”判断好不好,这是我最不建议的做法。你在调优前先建立一套评测集,哪怕只有三四十条问题,每条标注好期望命中的文档片段,然后用两个指标衡量:

  • Hit Rate(命中率):正确片段是否出现在召回结果中。这个指标反映检索链路的基础能力,等于“有没有找到”。
  • MRR(平均倒数排名):正确片段排在结果第几位。比如排第一得 1.0,排第二得 0.5。它反映排序质量,等于“找得到且排得好”。

我自己调优的顺序是:先关注 Hit Rate,确保正确内容能进来;再优化 Rerank 和排序,把正确内容顶到最前面。如果 hit rate 就低的离谱,别调 prompt 了,回去看切块和 embedding。

4.3 混合检索和 Rerank:性价比最高的两板斧

向量检索擅长语义匹配,BM25 擅长关键词和专有名词精确匹配。把两者结合再重排,是我在生产项目中收益最明显的一次升级。具体做法不复杂:向量检索取 top 30,BM25 取 top 30,一起去重后交给 Rerank 模型,取 top 5 进入 Prompt。

Rerank 模型国内现在有不少好选择,比如 bge-reranker。它的推理方式是把 query 和每个候选片段拼在一起,输出相关性分数,所以对语义理解比单纯 embedding 距离更准。代价是要跑几十次推理,不过候选量控制在 50 以内,耗时几十毫秒到几百毫秒,完全可接受。

有个经验想分享:Rerank 并不是所有场景的必备。如果你的知识库只有几百个 chunk,向量检索 top 10 基本就已经很准了,加 Rerank 增益有限。但当 chunk 规模上万甚至百万,Rerank 的收益会越来越大。

4.4 Query 改写:让“找”的方向更准

还有一个容易忽略的问题:用户问得模糊,检索自然不准。原样拿用户 query 去做 embedding,效果往往一般。现在的主流做法是让大模型先做 query 改写再用改写后的结果去检索。比如用户问“员工社保怎么交”,改写后变成“员工社会保险缴纳流程、基数计算、企业和个人承担比例”。

Agentic RAG 把这一点做到了极致:Agent 不再只是做一次检索,而是根据初步结果决定要不要换个关键词再搜、要不要拆分成多个子问题分别搜、要不要沿着某条线索深入追问。我在落地时至少会让 Agent 具备“二次检索”能力,也就是第一次结果不满意时,自动改写问题再检索一轮。这一步对复杂问题帮助很大。

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

5.1 检索不到答案:先别急着调模型

如果用户问题明明在知识库里有答案,但系统就是答不上来,我一般按这个顺序排查:

  1. 确认问题有没有命中的 chunk:直接打印检索结果,看看前 5 条跟问题是否相关。
  2. 检查命中但内容残缺:是不是 chunk 切得太碎,答案被断了。
  3. 检查命中但语义偏了:是不是 embedding 模型对领域术语理解不够。
  4. 检查知识到底有没有成功入库:查看库里的 chunk 数量、内容是否完整。

一个真实案例:某次排查发现所有问题都召回了一条不相关文档,查到最后是入库时把某批文档的元数据 title 写反了,导致排序逻辑受影响。这类问题光靠看回答看不出来,必须把检索中间结果打出来对一遍。

5.2 版本更新后,旧知识一直在“捣乱”

RAG 上线后最常遇到的问题之一:文档已经更新了,但回答还是老版本内容。原因通常是向量库里新旧版本 chunk 并存,或者删除旧文档时只删了部分 chunk。

我的经验是:每个 chunk 的 metadata 里必须带上文档版本号和更新时间,删除时按 doc_id 批量操作,而不是按文本内容去匹配。如果向量库自带过滤功能,还可以在检索时直接加 doc_id 白名单,从源头挡住废弃版本。增量更新建议做成独立任务,尽量避免边查边写。

5.3 上下文窗口溢出和内容错位

很多初学者直接把 top_k 设为 10,结果每轮请求都超长。这里有两个实用技巧:一是限制每个 chunk 的最大字符数,二是动态计算 prompt 长度,如果超了就缩减 top_k,而不是直接报错。这个逻辑有点像式内存管理:给 prompt 设置一个预算,内容按重要程度竞争进入上下文。

内容错位也值得注意。有时候检索结果相关性没问题,但模型还是答非所问,这时要重点排查 prompt 里的片段顺序。我踩过一次坑:代码里把片段按数据库默认顺序拼进 context,没有按相似度降序,结果模型被排在前面的一段低相关文本带偏了。

5.4 冷启动:知识库从 0 到 1 的准备工作

如果你是从零开始给团队搭知识库,不要一上来就埋头写代码。先做一次知识盘点:哪些文档是高频被问到的,格式统一吗,有没有敏感信息。敏感信息一定要在入库前做脱敏处理,否则后续每一轮检索都会带着它进入模型上下文,风险很大。

我经历过一次教训:把客户名单直接塞进知识库做 demo,结果在检索日志里发现了完整的客户联系方式。从那以后,我在入库流程里强制增加了敏感词过滤和字段脱敏步骤,并且移除后重建索引。这条建议对 To B 项目特别重要。

6. 从基础管道到工程落地:我的一些心得体会

6.1 先有评测,再优化

如果只让我给一条 RAG 落地的建议,那就是:先建立评测集。哪怕一开始只有五十条问题,也要先把评测跑出来,再开始调参。因为 RAG 链条上的每个环节都有优化空间,没有评测指标你就不知道改的是哪一环,很容易陷入“改了三天,感觉变好了一点”的状态。

我自己经常用十来条高难度问题当“拦路虎”,比如包含专有名词的、需要多步推理的、故意带有歧义的。每隔几天跑一轮,看命中率和回答质量有没有回退。这个习惯帮我避免了多次上线后才发现效果差的情况。

6.2 小团队怎么选择技术栈

小团队或个人的技术选型,我建议按数据规模分三档:

  • 千级 chunk:直接用 Chroma 或 FAISS,一个 Python 进程就能搞定。
  • 十万级 chunk:建议上 Milvus 或 Qdrant,用容器部署,同时引入混合检索。
  • 百万级以上:必须引入分布式向量库,并考虑索引分区和 GPU 资源调度。

模型选择方面,中文场景下 bge-m3 是不错的起点,兼顾语义和稀疏检索能力。如果想省成本,可以先试用 OpenAI 兼容的在线 embedding;如果数据敏感,则优先部署开源 embedding 模型。

还要提醒一点:技术包装永远只是手段。我在很多项目里发现,真正决定 RAG 效果的是数据清理做得干不干净、切块策略合不合理、评测集贴不贴近真实用户问题。你在炫酷的框架上花的时间,远不如在知识治理上花的时间值得。

6.3 下一步:走向 Agentic RAG

基础 RAG 跑通之后,你自然会遇到一个问题:用户的问题往往不是一个检索能搞定的。比如“对比一下我们三款产品的定价策略”,它需要多次检索、跨文档整合、甚至需要查一下最新价格表。这就是 Agentic RAG 的场景。

我目前在生产环境的做法是:把 RAG 封装成 Agent 的一个工具,由 Agent 决定何时调用、调用几次、要不要改写查询。Agent 内部再维护一个记忆池,把历史检索结果存下来,避免同一个问题反复检索。这个架构听起来复杂,但实现上就是在基础管道外面套了一层“规划和调度”的逻辑。

热词里有人提到 skill 怎么和 RAG 结合,我的建议是:先让 skill 负责纯逻辑流程,比如读取上传文件、调用内部 API,RAG 则稳定承担知识查询,两者之间通过 Agent 的规划模块衔接,不要强行耦合。等基础管道稳定后,再逐步增加 graphrag、知识图谱这些高级能力。这个系列后续我会单独展开讲。

最后分享一个我反复踩过的坑:上线前一定要检查模型的“不知道”行为。很多模型在 knowledge base 没有答案时,还是会尽力圆一个答案,这对企业场景是致命的。你必须在 prompt 和模型两个层面都约束好,必要时加一个低分拒答机制——检索相似度低于某个阈值时直接拒绝回答,让用户补充信息或转人工。这一步,比任何调参都更能守住最后一条防线。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询