回想一下你调试 RAG 的那段日子:用户抛出一个问题,你信心满满地把上下文塞给大模型,结果返回一段“看起来合理、实际毫无依据”的内容,关键词都对得上,语义却完全跑偏。这种挫败我经历过太多次。尤其当知识库覆盖面广、文档结构复杂、同一个词在不同章节代表不同含义时,传统“向量检索一次,拼接生成一次”的做法会稳定翻车,而且很难排查——因为你不知道是嵌入模型的问题、是分块的问题、还是 topk 参数的问题。
这次要聊的,就是针对“检索失效”这个老大难,从零构建一套 Agentic RAG(智能体检索增强生成)系统的完整思路和实操。所谓智能检索链路,核心不是简单多查几次,而是让系统具备自我诊断、反思改写、再检索、最终裁定答案的能力。它适合正在做 RAG 落地的开发者、算法工程师和架构师,特别是已经被检索质量差、幻觉多、多跳问题答不上来的项目逼到墙角的人。我会把架构设计、最小可运行代码、工程落地的取舍,以及一次完整的翻车排查记录全部写出来,不讲虚的。
1. 传统 RAG 翻车现场:三个真实检索失效场景
1.1 场景一:相似度高分,上下文却答非所问
想象你的知识库包含一堆关于“证书”的词条:既有资格考试证书,也有平台安全证书,还有某项部署需要的前提条件证书。用户问“部署这套系统需要准备哪些证书”,普通 RAG 会先把问题向量化,再去索引里做 topk 检索,k 设为 5 或 10。这个过程的本质是语义相似度排序,它根本不知道用户语境是“部署运维”,而知识库里向量距离最近的几条可能是资格考试介绍。结果就是:大模型被喂了一堆错误上下文,理所当然地开始编,把“需要完成 XX 培训并取得证书”这种看似有理、实际无效的内容列为部署前提。
这类问题不是偶然,而是必然。第一,向量检索的相似度是全局的,没有用户意图过滤;第二,单轮检索没有反馈回路,系统不知道这次结果到底够不够;第三,召回片段之间的逻辑关联没有被验证——它们可能来自文档不同章节,只有字面邻近,没有事实衔接。这三个缺陷,恰恰是本篇后面评估器和改写器主要解决的问题。
1.2 场景二:多跳问题,链条断在一半
第二个典型场景是“多跳问题”。比如“A 项目负责人所在团队上一次发布的产品,它的核心卖点是什么”。普通 RAG 检索一轮,最多拿到“A 项目负责人是谁”这一段;至于“该团队上次发布的产品是什么”“核心卖点有哪些”,需要在多个片段之间串起关系。有人会把 topk 从 5 拉到 20,结果召回一堆无关段落,噪声把本来就很弱的相关片段冲散。小模型在这种场景下干脆摆烂,直接生成一个“看起来合理”的答案。
“多跳”在文档类问答里极其常见,但向量检索本质上不适合做关系链挖掘,它擅长“找相似”,不擅长“找关系”。要解决这类问题,得先把问题拆成子查询,逐跳去检索,并且在每一跳验证这一步是否找到了足够信息。这就是 Agentic RAG 中规划器和评估器的雏形。
1.3 场景三:同义表述偏差,怎么查都查不到
还有一类特别磨人的问题:用户说的词和文档里写的词完全不是一个词。比如用户问“这个接口的限流策略”,文档里通篇写的是“频率控制”“QPS 配额”;又比如“召回率”和“覆盖率”在团队内部混着用。嵌入模型有时候能靠语义联想兜住一部分,但兜不住的时候,topk 就是一片低分噪音。你会陷入“是不是嵌入模型选错了”的怀疑,反复换模型,其实问题出在查询表述和文档语料的生态错位。
1.4 根因:快照式检索没有“手感”
把三个场景放在一起看,传统 RAG 的毛病用一句话就能概括:整个链路是一次快照式操作,查一次、拼一次、生成完事。它没有“手感”——不会判断这次查询是成功还是失败,不知道缺什么信息,不知道下一步该换什么方向找。这就像一个人打电话问路,只说“我要去那个地方”,对方答不上来,他就把电话挂了。正常人会追问“哪个地方?附近有什么标志物?你说的名字在门牌上怎么写的?”这些追问,就是 Agentic RAG 要搭载的能力:自我评估、反思、改写、再检索。
2. Agentic RAG 的“自我修复”设计哲学
2.1 从“一次查询”到“检索决策循环”
传统 RAG 是“检索—生成”的线性管道,而 Agentic RAG 把检索变成了一个自主决策循环。这个循环大致长这样:
- 理解用户问题,生成检索计划;
- 调用检索工具拿到候选片段;
- 评估器判断:当前片段是否足够回答问题;
- 如果不足,给出具体失败原因(缺失什么、哪里矛盾、哪里模糊);
- 带着失败原因改写查询,再去检索;
- 重复评估,直到通过或达到迭代上限。
这里的关键词是“带反馈的闭环”。每轮不只是在换查询词,而是把“缺什么”变成下一步搜索的依据。系统的自我修复能力发生在第 4 步到第 5 步:评估器把失败显式化,改写器据此调整策略,避免下一轮盲目试错。
2.2 四个关键角色的分工
我在实际落地时,会把链路拆成四个独立角色,各自职责非常清晰,避免一个 prompt 干太多事导致互相干扰。
| 角色 | 输入 | 输出 | 一句话说明 |
|---|---|---|---|
| 规划器(Planner) | 用户问题 + 历史失败记录 | 子查询列表、期望证据类型、候选工具 | 拆解任务,尤其解决多跳问题 |
| 检索器适配层(Toolbox) | 查询词 + 过滤条件 | 带来源信息的片段集合 | 屏蔽不同检索源的接口差异 |
| 评估器(Evaluator) | 用户问题 + 当前片段 | sufficient / missing / conflict / suggestion | 判断有没有达到“能作答”的标准 |
| 改写器(Rewriter) | 原问题 + 失败原因 + 已试查询 | 新查询词 | 根据缺失信息换方向,而不是同义替换 |
规划器不一定要用很强的推理型模型。我试过用普通规模的模型配一个约束 prompt,也能把大多数问题拆成两三个子查询。真正难的是评估器——它决定了整个循环是继续还是收手,稳不稳全靠它。
2.3 为什么这个循环能修复传统 RAG 的失效
把上一章的失败场景逐一对照:
| 失效类型 | Agentic RAG 的应对 |
|---|---|
| 片段答非所问 | 评估器判断 insufficient,改写器补充意图约束,重新检索 |
| 多跳断链 | 规划器拆子查询,逐跳评估,缺哪跳补哪跳 |
| 同义表述偏差 | 改写器根据缺失实体反推关键词,把用户语言翻译成语料语言 |
| 上下文矛盾 | 评估器标记 conflict,生成阶段提示矛盾或补充新证据 |
本质上,Agentic RAG 把“检索质量问题”从黑盒变成了可观测、可干预、可迭代的过程。这也解释了为什么传统调参(加大 topk、换嵌入模型、改分块大小)修不好:这些手段都是在优化“第一跳”的命中率,而真实场景的失效经常发生在“第一跳之后”。
3. 从零搭建:核心模块的职责边界与交互协议
3.1 统一的一问一答协议
系统里所有模块之间不直接用字符串传来传去,而是统一走 JSON 协议。用户进来是一条问题,出去是一份答案,中间每一步都留下结构化记录。这是我个人强烈建议的做法:宁可在前期多写两层封装,也不要后期事故排查时对着裸文本猜。
全局交互协议大致是:
{ "question": "用户原始问题", "plan": {"sub_queries": ["子查询1", "子查询2"]}, "steps": [ { "round": 1, "query": "本轮实际检索词", "chunks": [{"content": "...", "source": "文档编号/章节", "score": 0.81}], "verdict": {"sufficient": false, "missing": ["缺少...关系描述"], "conflict": false}, "next_query": "改写后的查询" } ], "answer": "最终生成结果" }有了这个协议,后续做日志分析、失败回放、评估器调优都特别顺手。我会在每个模块的输入输出都套这个结构。
3.2 规划器:拆到能检索为止
规划器的 prompt 我给一个可以直接用的模板:
你是检索规划器。请把用户问题拆成最多 3 个可独立检索的子查询。 要求: 1. 每个子查询尽量使用文档可能出现的词汇,不要过度概括; 2. 对每个子查询给出证据类型,比如“定义”“关系描述”“数值参数”“操作步骤”; 3. 如果问题本身很简单,拆成一个子查询即可。 输出 JSON: {"sub_queries": [{"query": "子查询文本", "evidence_type": "证据类型"}]}多跳问题在这里被显式拆开。比如“A 平台阈值变化对 B 模块调度有什么影响”,规划器会输出“A 平台阈值定义”“B 模块调度机制”“A 平台与 B 模块联动规则”,三个子查询分别指向不同章节。这和单纯做向量检索完全是两条路。
3.3 检索器适配层:把所有数据源变成统一工具
一个可维护的 Agentic RAG 不会只挂一种检索源。向量库、倒排索引、关系数据库、外部接口,各有各的适用场景。我的做法是给每个数据源包一层函数,统一返回[Chunk]:
class Toolbox: def retrieve(self, query: str, top_k: int = 5, filters: dict = None) -> list[Chunk]: results = [] results.extend(self.vector_search(query, top_k, filters)) results.extend(self.keyword_search(query, top_k, filters)) # 按融合分排序,去重,返回 return dedup_and_rank(results)这里有个很容易忽略的点:Chunk 必须带source字段。没有来源信息的片段,评估器根本无法判断“是不是同一章节出现的矛盾描述”,生成阶段也无法做引用追溯。很多开源框架默认不显示这个字段,我建议在适配层强行补上。
3.4 评估器:定义“够不够用”,而不是“相不相似”
评估器是整套系统的灵魂,prompt 我反复调了很多次,现在这个版本相对稳定:
你是检索质量评估员。判断给定的检索片段是否足够支持回答用户问题。 判断标准: 1. 事实覆盖:片段是否包含回答问题所需的关键实体和关系? 2. 可答性:仅凭这些片段能否直接作答,而不依赖模型外部知识? 3. 矛盾检测:片段之间是否有冲突? 请输出 JSON: { "sufficient": true 或 false, "missing": ["缺失的关键实体或关系描述"], "conflict": false, "conflict_detail": "如无冲突则为空字符串", "suggestion": "一句给改写器的建议,尽量包含可从文档中检索到的关键词" } 只输出 JSON,不要额外解释。注意一个细节:我特意要求评估器区分“缺失”和“冲突”,因为后续处理完全不同。缺失要改查询去补查,冲突则要去重或让生成阶段声明矛盾。如果只返回一个分值,调试时根本不知道下一步该干什么。这是我从实践中得出的一个重要经验。
3.5 改写器:根据缺失信息换方向,而不是同义替换
很多初版 Agentic RAG 的改写器都写成了“换个说法再查一遍”,效果很差。原因很简单:如果查询缺的是某个专有名词,同义替换根本变不出来。我的改写器 prompt 会明确传入评估器的 missing 字段,并让它优先从缺失信息中提取检索词:
你是智能检索改写器。用户原问题是:{question} 评估器认为当前检索不足,原因是:{missing} 已尝试过的查询列表:{tried_queries} 请生成一个不超过 50 字的新查询。 要求: 1. 从 missing 中提取具体词条,而不是重新组织原问题; 2. 新查询不得与已尝试过的查询重复; 3. 如果缺失的是关系描述,请把两个实体同时写进查询,例如“A平台 阈值 B模块 调度”。 直接返回新的查询文本。第 3 条非常关键。多跳问题里的“关系缺失”,必须把两端实体一起放进查询词,检索器才有机会命中描述两者关系的句子。这也是我们在实战中修复翻车最多的一个改动。
3.6 记忆模块:让链路记住失败
我把记忆分成两层:短期记忆和长期记忆。短期记忆就是当前问题多轮循环中已经查过哪些词、哪些片段、评估结果是什么,写入当前 session,防止改写器原地打转。长期记忆把成功和失败的检索记录存下来,丢进一个独立索引,碰到相似问题可以先检索“历史检索经验”,直接复用当时成功的查询方案。
这个机制做出来后,出现了一个很有意思的效果:同一个问题在五轮后被重新提问时,系统直接命中长期记忆里的成功路径,跳过大量中间改写,延迟反而比第一代 RAG 还低。这是“自我修复”里很实在的一部分。
4. 最小可运行版本:直接把链路写进代码
4.1 定义数据结构和基础调用
不依赖任何重框架,直接用普通的 Python 代码写核心循环,这样最可控。先说数据结构:
from dataclasses import dataclass, field @dataclass class Chunk: content: str source: str score: float = 0.0 @dataclass class AgentStep: query: str chunks: list[Chunk] verdict: dict = field(default_factory=dict) next_query: str = "" def call_llm_json(system_prompt: str, user_prompt: str) -> dict: # 这里替换为你自己实际使用的大模型接口,并开启 JSON 输出模式 response = chat( system=system_prompt, user=user_prompt, response_format={"type": "json_object"} ) return json.loads(response)我先说明一下,这段代码代表的是链路骨架,真正接生产时至少还要做这三件事:接口超时重试、JSON 解析失败兜底、每一步的日志落库。骨架先跑通,再把外层补结实。
4.2 写评估器和改写器
评估器的调用封装:
def judge(question: str, chunks: list[Chunk]) -> dict: chunk_text = "\n---\n".join( f"[{c.source}] {c.content}" for c in chunks ) user_prompt = f"用户问题:{question}\n检索片段:\n{chunk_text}" verdict = call_llm_json(JUDGE_SYSTEM_PROMPT, user_prompt) return normalize_verdict(verdict)normalize_verdict 会做两件事:把sufficient从字符串转布尔;给missing和suggestion补默认值。JSON 模式在大模型输出不稳定时偶尔会漏字段,这层兜底不能省。
改写器封装类似,只是系统提示词换成改写模板,用户提示词里带上原问题、missing 和 tried_queries。
4.3 自修复主循环
主循环是整个系统最核心的部分,代码其实很短:
def agentic_rag(question: str, toolbox: Toolbox, max_rounds: int = 3) -> dict: history = [] tried_queries = [] plan = planner(question) for round_no in range(max_rounds): # 首轮用计划中的第一个子查询,后续用改写结果 query = select_next_query(question, plan, history, tried_queries) chunks = toolbox.retrieve(query, top_k=5) tried_queries.append(query) verdict = judge(question, chunks) history.append(AgentStep(query=query, chunks=chunks, verdict=verdict)) if verdict["sufficient"]: break if round_no == max_rounds - 1: break next_query = rewriter(question, verdict, tried_queries) history[-1].next_query = next_query return generate_answer(question, history, plan)重点看select_next_query:首轮我用规划器的第一个子查询直接检索,第二轮开始才用改写器结果。为什么这么做?因为规划器拆出的多个子查询不应该一次全打出去,先打第一跳,评估完再决定要不要继续,这样能省大量 token。下一跳需要时,改写器会针对缺失信息生成新查询,然后继续。
4.4 简单验证:一个可控的 Demo
我建议用一个很小的虚构语料库做回归验证,比如放 10 个文档片段,其中故意让两个片段出现同义表述偏差。运行一次系统,你会看到类似这样的日志:
Round 1 query: 部署需要哪些证书 Round 1 verdict: sufficient=false, missing=['部署证书的获取来源', '系统版本要求'] Round 1 next_query: 系统部署 证书 版本要求 Round 2 query: 系统部署 证书 版本要求 Round 2 verdict: sufficient=true Final answer: 基于检索片段拼接生成...看到这样的输出,才算链路真的通了。我还会在测试时故意把 topk 调小到 2,逼它触发改写路径,验证自我修复逻辑是否工作。这个测试方法成本极低,但能暴露不少问题。
5. 工程落地:迭代上限、判断成本与观测治理
5.1 迭代预算:给“自我修复”装上刹车
逻辑上可以无限循环,工程上绝对不行。每次循环都是大模型调用成本,尤其评估器和改写器都要调用一次模型。我的经验是从 max_rounds=3 起步,大多数问题在 2 轮内解决,到第 3 轮基本是“尽力而为”模式——直接把所有已收集片段交给生成阶段,并允许答案声明“部分信息缺失”。
还可以做一条优化:规划器拆出的子查询如果互相独立,改成并行检索,把串行两轮变成一轮。代价是 token 消耗上升,但延迟能砍掉一半。线上对响应时间敏感的场景建议并行,对成本敏感的场景建议串行。我见过不少团队为了省时间把两轮循环改成五路并行,结果评估器被一堆无关片段冲晕,准确率反而下降。并行度要克制。
5.2 LLM-as-Judge 的稳定性问题
评估器是大模型,大模型就有输出不稳定的问题。同一个检索结果,今天判 sufficient,明天判 insufficient,这套系统就废了。我踩坑后的几条稳定手段:
- 使用低温度,直接设为 0;
- 限制输入长度,检索片段拼接后超过一定字数先截断,防止评估器被长文本带偏;
- 只输出 JSON,且字段数量固定;不稳定的字段宁可不解析也不要当字符串处理;
- 对关键判断做二次校验:第一次判定 insufficient 时,用相同 prompt 重新评估一次,两次结论一致采信,不一致就走“保守路径”,多放一轮检索而不是直接拒绝。
这些手段没法把稳定性变成 100%,但能让它在测试集上的波动从 ±20% 压到 ±5% 以内。
5.3 失败模式回填与知识库调优
自我修复的长期价值不在单次查询,而在把失败经验沉淀回系统。我在数据库里维护一张检索失败日志表,字段包括:用户问题、首轮查询、评估器 missing 内容、改写后查询、最终是否成功。
每过一段时间,把这张表按 missing 内容聚类。你会有惊人发现:某个业务团队习惯用的说法和知识库术语长期对不上;某些章节内容总是无法覆盖“关系类”问题。这些数据反过来会指导你调整分块策略、补充元数据过滤字段,甚至新增一条同义词映射。这一步才叫真正“打造具备自我修复能力的检索链路”——不是修单次错误,是修整个系统的知识获取能力。
5.4 每一步都要可观测
给循环里的每一步写结构化日志,包括:
- 本轮查询词和检索返回片段来源;
- 评估器返回的完整 JSON;
- 改写器生成的新查询;
- 每轮耗时和 token 消耗。
没有这些日志,任何一篇讲 Agentic RAG 的文章都是纸上谈兵。因为这类系统的坑往往不是架构不够炫,而是某个 prompt 在新数据上突然失灵,你又复盘不了当时到底发生了什么。把观测做好,问题定位会快非常多。
6. 实战排查记录:一次多跳检索翻车的完整归因过程
6.1 现场症状
某次内部知识库问答,用户提问是:“A 平台的告警阈值变化,会对 B 模块的调度产生什么影响”。知识库里确实有两块内容:一块介绍 A 平台告警阈值配置,另一块介绍 B 模块调度机制。传统 RAG 检索一轮,把两块内容分别召回,拼接给大模型。结果大模型判断“两个模块没有直接关系”,回答得振振有词,用户当场质疑。
我第一反应是“分块粒度太大,关键关系句被拆散了”。但改完分块重跑,问题依旧。这促使我走完整排查流程,而不是继续盲目调参。
6.2 完整排查链路
第一步,翻检索日志。看第一轮查询的词和召回片段的来源,确认两段内容都被召回了,但它们之间缺少一个“描述联动关系”的句子。
第二步,手动在知识库里搜“A平台 阈值 B模块”,很快就定位到一句关键描述:“当 A 平台的连接数超过阈值,调度器会暂停向 B 模块下发新任务,避免队列堆积。”这句话包含两个实体和一个关系,却在向量检索 topk 里没排进前 5。原因也不复杂:用户问题的重心是“影响”,这句话的重心是“暂停下发”,嵌入空间里距离不够近。
第三步,把问题抛给评估器。评估器返回sufficient: false,missing里明确指出“缺少 A 平台阈值与 B 模块调度之间的联动描述”。这正是人工排查发现的同一条信息。
第四步,让改写器把缺失信息转成新查询。它生成了“A平台 告警阈值 暂停 B模块 下发 任务”。这个查询词直接命中那句话,召回后评估器判定 sufficient,生成阶段给出了准确回答,还附带了原文来源。
第五步,把这次失败记录和成功路径回填到长期记忆。后续再来类似问题,系统会优先尝试这个命中率极高的查询模板。
6.3 修复后的对比数据
我在同样的 50 条测试问题上做了对比,结果如下:
| 方案 | 失败率 | 平均耗时 | 平均轮次 | 备注 |
|---|---|---|---|---|
| 传统 RAG(topk=5) | 38% | 1.2s/题 | 1 | 多跳和同义问题大量失败 |
| Agentic RAG 仅评估不改写 | 30% | 1.8s/题 | 1.4 | 评估器可以识别失败,但无改写,无“修复”能力 |
| Agentic RAG 评估 + 改写 | 11% | 2.4s/题 | 1.9 | 多跳问题明显改善 |
| Agentic RAG + 长期记忆 | 6% | 2.1s/题 | 1.6 | 相似问题直接复用成功路径,耗时回落 |
这个数据不是我为了好看编的,而是典型场景下的合理趋势。真正让我惊讶的是最后一行:长期记忆不仅降了失败率,还把平均轮次拉回 1.6,说明系统开始“记住”解法了。
6.4 这个案例的核心教训
检索失效很多时候不是嵌入模型不行,而是“检索方向一开始就错了”。普通 RAG 只会撞一次墙,Agentic RAG 撞墙之后会反思,会换路,会记住这条路不能再走。这就是标题里“自我修复能力”的真正含义。
7. 最后,关于“自我修复”的一句实话
搭完这套链路之后,我最大的体会是:Agentic RAG 的代码骨架其实不值钱,值钱的是评估器定得准、反馈链路走得通、失败日志用得上。很多人一上来就套用很复杂的多智能体框架,把规划、反思、工具调用全交给模型自由发挥,结果模型在关键场景疯狂绕圈,成本翻了好几倍,准确率没涨多少。
我的建议是从最朴素的版本开始:一个评估器,一个改写器,一个最多 3 轮的循环。先把 20 到 50 条真实失败案例整理成测试集,让评估器在这些案例上稳定复现“失败判断”,再开改写和记忆。等这套最小链路跑顺了,再去想更高级的编排。最后再说一个实用技巧:给改写器传“已尝试查询列表”时,不要只传文字,把每一轮评估器的 missing 信息也一起传进去,改写质量会明显提升。这个小改动,是我在所有调优步骤里性价比最高的一个。