外观
一文搞懂 Agentic RAG:让检索 Agent 自主决定何时检索、如何改写的架构设计
传统 RAG 系统是一条固定流水线:问题进来,切块入库的向量库被检索一次,取回的文档拼进提示词,模型生成答案。这套流程在「文档里明明白白写着」的问题上表现尚可,但真实用户的问题往往不配合:问句口语化、指代不清,直接拿去检索命中质量很差;答案分散在多份文档里,一次检索根本凑不齐;还有些问题压根不需要检索外部知识,流水线也照检索不误,白白增加延迟和成本。
Agentic RAG 的解法是把「检索」从流水线的固定环节,变成 Agent 手里可以随时调用、随时放弃的工具:模型自己判断要不要检索、检索词怎么写、拿回的结果够不够好、不够好是换个说法再检索还是直接回答。决策权从流程图转移到了模型手里,这正是它比固定流水线强的原因,也是它更容易失控的原因——本文围绕这套架构的设计与踩坑展开。
方案对比:从固定流水线到 Agentic RAG
| 维度 | 固定流水线 RAG | 自适应 RAG(路由+自评) | Agentic RAG |
|---|---|---|---|
| 检索时机 | 每问必检索 | 入口路由判断 | 模型按需自主决定 |
| 查询改写 | 无或固定规则 | 常内置一轮 | 按检索反馈多轮改写 |
| 失败恢复 | 无,一次定输赢 | 相关性差时重检 | 多轮迭代直至达标或止损 |
| 延迟与成本 | 低且确定 | 中 | 高且波动,需上限控制 |
| 适用场景 | 高频简单问答 | 大部分生产场景 | 复杂分析、多跳问题 |
三种形态不是替代关系,而是演进关系:流量大头是简单问题,就用固定流水线兜底;加一层入口路由过滤掉无需检索的问题,是性价比最高的一步升级;只有当业务确实存在大量多跳、模糊问题时,才值得为那一部分流量上完整的 Agentic 架构。用一套复杂的 Agentic 流程去接所有问题,是把延迟和成本花在大多数用不上的地方。
实操:用 LangGraph 搭一个带自纠错的状态图
Agentic RAG 的本质是一个带条件边和循环的状态图。下面用 LangGraph 实现核心链路:入口路由 → 查询改写 → 检索 → 质量评估(不达标则回环改写)→ 生成。
python
from typing import TypedDict
from langgraph.graph import StateGraph, END
class RagState(TypedDict):
question: str # 用户原始问题,全程保留不改写
rewritten: str # 当前轮的检索查询
documents: list # 本轮检索到的文档
loop_count: int # 已执行的检索轮数
def route_question(state: RagState) -> str:
"""入口路由:闲聊与常识问题直接生成,知识型问题才进检索链路"""
decision = llm_judge_route(state["question"]) # 返回 "direct" 或 "retrieve"
return "generate" if decision == "direct" else "rewrite"
def rewrite_query(state: RagState) -> dict:
"""查询改写:把口语化、有指代的问句改写成适合向量检索的独立查询"""
rewritten = llm_rewrite(state["question"])
return {"rewritten": rewritten}
def retrieve(state: RagState) -> dict:
query = state.get("rewritten") or state["question"]
docs = vector_store.similarity_search(query, k=6)
return {"documents": docs, "loop_count": state.get("loop_count", 0) + 1}
def grade_documents(state: RagState) -> str:
"""质量评估:命中足够多相关文档就生成,否则改写重检,超限强制止损"""
relevant = [d for d in state["documents"] if llm_grade_relevance(d, state["question"])]
if len(relevant) >= 2:
state["documents"] = relevant
return "generate"
if state.get("loop_count", 0) >= 2:
return "generate" # 两轮仍不达标:带上现有结果生成并标注不确定性
return "rewrite"
def generate(state: RagState) -> dict:
answer = llm_answer(state["question"], state.get("documents", []))
return {"answer": answer}
builder = StateGraph(RagState)
builder.add_node("rewrite", rewrite_query)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.set_entry_point("route")
builder.add_node("route", lambda s: s)
builder.add_conditional_edges("route", route_question,
{"rewrite": "rewrite", "generate": "generate"})
builder.add_edge("rewrite", "retrieve")
builder.add_conditional_edges("retrieve", grade_documents,
{"rewrite": "rewrite", "generate": "generate"})
builder.add_edge("generate", END)
rag_app = builder.compile()实现要点有三处。其一,原始问题全程保留在 state 里,改写只产生检索查询——改写是给检索器看的,不是替换用户的原话,生成答案必须回到原问题上。其二,质量评估节点是整个架构的发动机,它让「检索质量差」从静默失败变成可感知、可恢复的信号。其三,loop_count 是安全阀,任何带循环的 Agent 图都必须有硬性止损,否则遇到检索质量差的语料库,系统会陷入「改写—检索—评估」的无限回环。
关键设计点:评估与改写怎么写才不跑偏
质量评估的判分标准要具体:「这份文档是否包含回答该问题所需的信息」远比「这份文档好不好」有效。判分可以用轻量级 LLM 逐文档判断,也可以先用向量相似度阈值做粗筛、只对边界样本调 LLM,把每轮的评估成本压下来。
查询改写的提示词要给模型明确的改写目标:补全指代、剥离口语、拆分复合问题。对多跳问题(如「A 公司最新财报里 B 业务的占比是多少」),单一改写不够,更有效的做法是把问题拆成两个子查询分别检索,这可以作为改写节点的一种扩展策略:让模型输出一个查询列表而不是单个查询。
最终进入生成环节的上下文拼接同样值得花心思。多轮迭代后拿到的文档要按相关度排序而不是按召回顺序堆放,每份文档标注来源与章节标题,方便模型在答案里引用出处;超长文档只截取与问题最相关的片段,避免单份文档独占半个上下文窗口。这些拼接细节不影响架构图上的节点个数,却直接影响答案的可信度——同一个图结构,拼接策略粗糙的实现往往评测得分低出一截。
避坑总结
- 迭代轮数必须有硬上限。实测中不设上限的回环在语料质量差时会产生三到五轮的无效检索,延迟和成本双双失控;两到三轮封顶,超限后降级生成并明确告知用户信息不足。
- 相关性评估别全信单次 LLM 判分。判分模型自己也会出错,用「相似度阈值粗筛 + LLM 精判边界样本」的混合策略,比逐文档全量判分更稳也更便宜。
- 改写查询不能替代原始问题。生成环节必须基于用户原话,改写仅服务于检索器;把改写结果当问题生成答案,是 Agentic RAG 最常见的正确性事故。
- 多轮检索的文档要去重合并。相邻两轮改写往往召回大量重叠文档,拼接上下文前按内容指纹去重,否则重复内容既浪费 token 又放大噪声。
- 上线前先建检索评测集。准备一批带标准出处的问题,覆盖简单、模糊、多跳三类,每次调整路由、改写或评估逻辑后回归跑一遍,避免「感觉变好了」式的迭代。
写在最后
Agentic RAG 的核心不是让流程更复杂,而是让系统获得「发现自己没找到答案」的能力,并把恢复动作交给模型。架构上它只多了路由、评估、改写三个节点,工程上却要求把每一次失败都变成下一轮的输入。先用路由过滤简单流量、再用受控的迭代兜住复杂问题,是这套架构在生产里站稳的务实路径。
免责声明
本工具与文章内容仅用于学习测试场景,优化建议仅供参考,请勿直接用于生产环境,由此产生的风险自行承担。