让RAG学会自己纠错：Agentic RAG实战指南
05篇讲Agentic RAG时给了基本概念和查询路由的代码。但那篇是"架构概览"级别——你知道了Agentic RAG是什么，但真要落地还有一堆问题：Agent怎么判断检索结果好不好？检索到垃圾内容怎么自动修正？多步推理怎么避免死循环？延迟太高怎么优化？
这篇把Agentic RAG从概念推到工程落地。重点讲两个2024-2025年最有影响力的方案：Corrective RAG（CRAG）和Self-RAG。这两个方案的核心思想一样——让RAG学会自我纠错——但实现路径完全不同。
基础RAG的问题在于它太"老实"：检索到什么就用什么，不管检索结果好不好。如果检索到了不相关的内容，LLM要么硬编答案，要么说"我不知道"。Agentic RAG的思路是加一层"裁判"：先评估检索质量，不好的话就换方式重新检索。
基础RAG vs Agentic RAG：差在哪
| 维度
|
基础RAG
|
Agentic RAG
检索策略
|
固定：1次检索，取TopK
|
自适应：根据问题复杂度决定检索几次
| |
结果评估
|
无：检索到什么用什么
|
有：评估检索结果质量，低质量触发重检索
| |
错误修正
|
无：答错就答错了
|
有：发现答案有问题自动重试
| |
数据源
|
单一：向量库
|
多源：向量库+数据库+API+搜索引擎
| |
决策
|
无：固定流水线
|
有：Agent自主选择检索策略
| |
延迟
|
200-500ms
|
1-10秒（多了决策和重试）
| |
成本
|
1次LLM调用
|
3-8次LLM调用
| |
准确率
|
基准
|
提升15-30%
|
核心权衡：Agentic RAG用更多的时间和成本换取更高的准确率。适合对准确率要求高、对延迟不敏感的场景（如企业知识库问答、法律咨询）。不适合实时对话场景（延迟要求<2秒）。
Corrective RAG（CRAG）：检索结果打分 + 自动纠正
CRAG是2024年提出的方案，核心思路：给检索结果打分，分三档处理。
<span leaf="">检索结果 → 质量评估 → ① Correct（相关）：直接用</span><span leaf=""><br></span><span leaf=""> → ② Incorrect（不相关）：丢掉，去网络搜索</span><span leaf=""><br></span><span leaf=""> → ③ Ambiguous（不确定）：部分用 + 网络搜索补充</span><span leaf=""><br></span>
质量评估：怎么判断检索结果好不好
<span><span leaf="">from</span></span><span leaf=""> langchain_core.prompts </span><span><span leaf="">import</span></span><span leaf=""> ChatPromptTemplate</span><span leaf=""><br></span><span><span leaf="">from</span></span><span leaf=""> pydantic </span><span><span leaf="">import</span></span><span leaf=""> BaseModel, Field</span><span leaf=""><br></span><span><span leaf="">from</span></span><span leaf=""> enum </span><span><span leaf="">import</span></span><span leaf=""> Enum</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">class</span></span><span leaf=""> </span><span><span leaf="">RetrievalGrade</span></span><span><span leaf="">(str, Enum)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""检索结果质量等级"""</span></span><span leaf=""><br></span><span leaf=""> CORRECT = </span><span><span leaf="">"correct"</span></span><span leaf=""> </span><span><span leaf=""># 相关：直接用</span></span><span leaf=""><br></span><span leaf=""> INCORRECT = </span><span><span leaf="">"incorrect"</span></span><span leaf=""> </span><span><span leaf=""># 不相关：丢弃</span></span><span leaf=""><br></span><span leaf=""> AMBIGUOUS = </span><span><span leaf="">"ambiguous"</span></span><span leaf=""> </span><span><span leaf=""># 不确定：部分使用</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">class</span></span><span leaf=""> </span><span><span leaf="">GradeResult</span></span><span><span leaf="">(BaseModel)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""检索结果评估"""</span></span><span leaf=""><br></span><span leaf=""> grade: RetrievalGrade = Field(description=</span><span><span leaf="">"检索结果质量等级"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> reason: str = Field(description=</span><span><span leaf="">"判断理由"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> relevant_parts: str = Field(description=</span><span><span leaf="">"如果ambiguous，哪些部分是相关的"</span></span><span leaf="">, default=</span><span><span leaf="">""</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">grade_retrieval</span></span><span><span leaf="">(question, retrieved_docs, llm)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""评估检索结果质量</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> CRAG的核心：不是所有检索结果都值得用。</span><span leaf=""><br></span><span leaf=""> 先让LLM判断检索结果跟问题的相关度，再决定怎么处理。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> grade_prompt = ChatPromptTemplate.from_template(</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""你是一个检索质量评估专家。请判断以下检索结果是否能回答用户问题。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">用户问题：{question}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">检索结果：</span><span leaf=""><br></span><span leaf="">{documents}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">判断标准：</span><span leaf=""><br></span><span leaf="">- correct：检索结果直接包含回答问题所需的信息</span><span leaf=""><br></span><span leaf="">- incorrect：检索结果跟问题完全无关</span><span leaf=""><br></span><span leaf="">- ambiguous：检索结果部分相关，但不完整或不确定</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">{format_instructions}"""</span></span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 拼接检索结果</span></span><span leaf=""><br></span><span leaf=""> doc_text = </span><span><span leaf="">"\n---\n"</span></span><span leaf="">.join([</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">f"[文档</span><span><span leaf="">{i+</span><span><span leaf="">1</span></span><span leaf="">}</span></span><span leaf="">] </span><span><span leaf="">{doc.page_content}</span></span><span leaf="">"</span></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">for</span></span><span leaf=""> i, doc </span><span><span leaf="">in</span></span><span leaf=""> enumerate(retrieved_docs)</span><span leaf=""><br></span><span leaf=""> ])</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">from</span></span><span leaf=""> langchain_core.output_parsers </span><span><span leaf="">import</span></span><span leaf=""> PydanticOutputParser</span><span leaf=""><br></span><span leaf=""> parser = PydanticOutputParser(pydantic_object=GradeResult)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> chain = grade_prompt | llm | parser</span><span leaf=""><br></span><span leaf=""> result = chain.invoke({</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"question"</span></span><span leaf="">: question,</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"documents"</span></span><span leaf="">: doc_text,</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"format_instructions"</span></span><span leaf="">: parser.get_format_instructions(),</span><span leaf=""><br></span><span leaf=""> })</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> result</span><span leaf=""><br></span>
CRAG的纠正策略
<span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">crag_search</span></span><span><span leaf="">(question, retriever, llm, web_search_fn=None)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""Corrective RAG：检索 → 评估 → 纠正</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 三档处理：</span><span leaf=""><br></span><span leaf=""> - correct → 知识转换（提取关键信息，去噪）</span><span leaf=""><br></span><span leaf=""> - incorrect → 丢弃，用网络搜索补充</span><span leaf=""><br></span><span leaf=""> - ambiguous → 提取相关部分 + 网络搜索补充</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># Step 1: 初始检索</span></span><span leaf=""><br></span><span leaf=""> retrieved_docs = retriever.invoke(question)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># Step 2: 质量评估</span></span><span leaf=""><br></span><span leaf=""> grade = grade_retrieval(question, retrieved_docs, llm)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> print(</span><span><span leaf="">f"检索评估: </span><span><span leaf="">{grade.grade}</span></span><span leaf=""> - </span><span><span leaf="">{grade.reason}</span></span><span leaf="">"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> grade.grade == RetrievalGrade.CORRECT:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 相关：知识转换（提取关键信息，去掉噪声）</span></span><span leaf=""><br></span><span leaf=""> refined_docs = knowledge_transformation(retrieved_docs, question, llm)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> refined_docs, </span><span><span leaf="">"knowledge_base"</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">elif</span></span><span leaf=""> grade.grade == RetrievalGrade.INCORRECT:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 不相关：丢弃，去网络搜索</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> web_search_fn:</span><span leaf=""><br></span><span leaf=""> print(</span><span><span leaf="">"检索结果不相关，启用网络搜索..."</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> web_results = web_search_fn(question)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> web_results, </span><span><span leaf="">"web_search"</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">else</span></span><span leaf="">:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> [], </span><span><span leaf="">"no_result"</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">else</span></span><span leaf="">: </span><span><span leaf=""># AMBIGUOUS</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 部分相关：提取相关部分 + 网络搜索补充</span></span><span leaf=""><br></span><span leaf=""> relevant_docs = [doc </span><span><span leaf="">for</span></span><span leaf=""> doc </span><span><span leaf="">in</span></span><span leaf=""> retrieved_docs </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> any(keyword </span><span><span leaf="">in</span></span><span leaf=""> doc.page_content </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">for</span></span><span leaf=""> keyword </span><span><span leaf="">in</span></span><span leaf=""> grade.relevant_parts.split())]</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> web_search_fn:</span><span leaf=""><br></span><span leaf=""> web_results = web_search_fn(question)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> relevant_docs + web_results, </span><span><span leaf="">"hybrid"</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">else</span></span><span leaf="">:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> relevant_docs, </span><span><span leaf="">"partial"</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">knowledge_transformation</span></span><span><span leaf="">(docs, question, llm)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""知识转换：提取跟问题最相关的信息，去掉噪声</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> CRAG的一个细节：即使是"correct"的检索结果，</span><span leaf=""><br></span><span leaf=""> 也不是整篇用，而是提取最相关的部分。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> transform_prompt = ChatPromptTemplate.from_template(</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""从以下文档中提取与问题最相关的信息，去掉无关内容。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">问题：{question}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">文档：</span><span leaf=""><br></span><span leaf="">{documents}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">提取关键信息（保留原文表述，不要改写）："""</span></span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> doc_text = </span><span><span leaf="">"\n"</span></span><span leaf="">.join([doc.page_content </span><span><span leaf="">for</span></span><span leaf=""> doc </span><span><span leaf="">in</span></span><span leaf=""> docs])</span><span leaf=""><br></span><span leaf=""> chain = transform_prompt | llm</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> refined = chain.invoke({</span><span><span leaf="">"question"</span></span><span leaf="">: question, </span><span><span leaf="">"documents"</span></span><span leaf="">: doc_text})</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 返回精炼后的文档</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">from</span></span><span leaf=""> langchain_core.documents </span><span><span leaf="">import</span></span><span leaf=""> Document</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> [Document(page_content=refined.content, metadata={</span><span><span leaf="">"source"</span></span><span leaf="">: </span><span><span leaf="">"crag_refined"</span></span><span leaf="">})]</span><span leaf=""><br></span>
Java类比：CRAG就像你写了一个带重试机制的HTTP客户端——先发请求，检查响应状态码，200就用，404就换URL重试，206就提取可用部分。基础RAG是发完请求不管返回什么都直接用。所以CRAG在Java开发者看来非常直觉——这不就是带fallback的REST模板嘛。
Self-RAG：让模型自己决定要不要检索
Self-RAG和CRAG的思路不同。CRAG是"先检索再纠错"，Self-RAG是"让模型自己决定要不要检索、检索结果好不好、要不要用"。
Self-RAG训练了一个特殊的LLM，它能输出"反思token"（reflection tokens）：
-
[Retrieve]：这个问题需要检索吗？ -
[IsRel]：检索结果相关吗？ -
[IsSup]：生成的答案有检索结果支持吗？ -
[IsUse]：这个答案对用户有用吗？
Self-RAG的推理流程
<span leaf="">用户问题 → [Retrieve]判断 → 需要检索 → 检索 → [IsRel]判断 → 相关 → 生成 → [IsSup]判断 → 有支撑 → [IsUse]判断 → 有用 → 输出</span><span leaf=""><br></span><span leaf=""> → 不需要 → 直接生成 → 输出</span><span leaf=""><br></span><span leaf=""> → 不相关 → 重新检索或直接生成</span><span leaf=""><br></span><span leaf=""> → 无支撑 → 重新生成</span><span leaf=""><br></span>
工程实现：用LangGraph模拟Self-RAG
真正的Self-RAG需要微调一个能输出反思token的模型。但工程上我们可以用LangGraph模拟这个流程——用普通LLM做判断节点：
<span><span leaf="">from</span></span><span leaf=""> langgraph.graph </span><span><span leaf="">import</span></span><span leaf=""> StateGraph, END</span><span leaf=""><br></span><span><span leaf="">from</span></span><span leaf=""> typing </span><span><span leaf="">import</span></span><span leaf=""> TypedDict, Annotated</span><span leaf=""><br></span><span><span leaf="">import</span></span><span leaf=""> operator</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">class</span></span><span leaf=""> </span><span><span leaf="">SelfRAGState</span></span><span><span leaf="">(TypedDict)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""Self-RAG的状态"""</span></span><span leaf=""><br></span><span leaf=""> question: str </span><span><span leaf=""># 用户问题</span></span><span leaf=""><br></span><span leaf=""> need_retrieval: bool </span><span><span leaf=""># 是否需要检索</span></span><span leaf=""><br></span><span leaf=""> documents: Annotated[list, operator.add] </span><span><span leaf=""># 检索到的文档</span></span><span leaf=""><br></span><span leaf=""> is_relevant: bool </span><span><span leaf=""># 检索结果是否相关</span></span><span leaf=""><br></span><span leaf=""> answer: str </span><span><span leaf=""># 生成的答案</span></span><span leaf=""><br></span><span leaf=""> is_supported: bool </span><span><span leaf=""># 答案是否有文档支撑</span></span><span leaf=""><br></span><span leaf=""> is_useful: bool </span><span><span leaf=""># 答案是否有用</span></span><span leaf=""><br></span><span leaf=""> retry_count: int </span><span><span leaf=""># 重试次数</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">judge_need_retrieval</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""[Retrieve] 判断是否需要检索</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 简单问题（闲聊、通用知识）不需要检索，</span><span leaf=""><br></span><span leaf=""> 复杂问题（公司政策、产品细节）需要检索。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> question = state[</span><span><span leaf="">"question"</span></span><span leaf="">]</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> judge_prompt = </span><span><span leaf="">f"""判断以下问题是否需要从知识库检索信息。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">问题：</span><span><span leaf="">{question}</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">判断标准：</span><span leaf=""><br></span><span leaf="">- 需要：涉及具体事实、公司政策、产品细节、技术文档</span><span leaf=""><br></span><span leaf="">- 不需要：闲聊、通用知识、观点表达</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">回答（需要/不需要）："""</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> response = llm.invoke(judge_prompt).content</span><span leaf=""><br></span><span leaf=""> need = </span><span><span leaf="">"需要"</span></span><span leaf=""> </span><span><span leaf="">in</span></span><span leaf=""> response</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"need_retrieval"</span></span><span leaf="">: need, </span><span><span leaf="">"retry_count"</span></span><span leaf="">: state.get(</span><span><span leaf="">"retry_count"</span></span><span leaf="">, </span><span><span leaf="">0</span></span><span leaf="">)}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">retrieve_documents</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""执行检索"""</span></span><span leaf=""><br></span><span leaf=""> docs = retriever.invoke(state[</span><span><span leaf="">"question"</span></span><span leaf="">])</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"documents"</span></span><span leaf="">: docs}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">judge_relevance</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""[IsRel] 判断检索结果是否相关"""</span></span><span leaf=""><br></span><span leaf=""> grade = grade_retrieval(state[</span><span><span leaf="">"question"</span></span><span leaf="">], state[</span><span><span leaf="">"documents"</span></span><span leaf="">], llm)</span><span leaf=""><br></span><span leaf=""> is_relevant = grade.grade != RetrievalGrade.INCORRECT</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"is_relevant"</span></span><span leaf="">: is_relevant}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">generate_answer</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""生成答案"""</span></span><span leaf=""><br></span><span leaf=""> context = </span><span><span leaf="">"\n"</span></span><span leaf="">.join([doc.page_content </span><span><span leaf="">for</span></span><span leaf=""> doc </span><span><span leaf="">in</span></span><span leaf=""> state[</span><span><span leaf="">"documents"</span></span><span leaf="">]])</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> prompt = </span><span><span leaf="">f"""基于以下上下文回答问题。如果上下文不足以回答，请说明。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">上下文：</span><span leaf=""><br></span><span><span leaf="">{context}</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">问题：</span><span><span leaf="">{state[</span><span><span leaf="">"question"</span></span><span leaf="">]}</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">回答："""</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> answer = llm.invoke(prompt).content</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"answer"</span></span><span leaf="">: answer}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">judge_support</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""[IsSup] 判断答案是否有文档支撑</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 关键：防止LLM编造答案。</span><span leaf=""><br></span><span leaf=""> 检查答案中的每个论断是否都能在检索到的文档中找到依据。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> answer = state[</span><span><span leaf="">"answer"</span></span><span leaf="">]</span><span leaf=""><br></span><span leaf=""> docs = state[</span><span><span leaf="">"documents"</span></span><span leaf="">]</span><span leaf=""><br></span><span leaf=""> doc_text = </span><span><span leaf="">"\n"</span></span><span leaf="">.join([doc.page_content </span><span><span leaf="">for</span></span><span leaf=""> doc </span><span><span leaf="">in</span></span><span leaf=""> docs])</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> support_prompt = </span><span><span leaf="">f"""判断以下答案是否被文档支撑。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">文档：</span><span leaf=""><br></span><span><span leaf="">{doc_text}</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">答案：</span><span leaf=""><br></span><span><span leaf="">{answer}</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">判断标准：</span><span leaf=""><br></span><span leaf="">- supported：答案中的信息都能在文档中找到依据</span><span leaf=""><br></span><span leaf="">- unsupported：答案包含文档中没有的信息（可能是编造的）</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">回答（supported/unsupported）："""</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> response = llm.invoke(support_prompt).content</span><span leaf=""><br></span><span leaf=""> is_supported = </span><span><span leaf="">"supported"</span></span><span leaf=""> </span><span><span leaf="">in</span></span><span leaf=""> response.lower()</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"is_supported"</span></span><span leaf="">: is_supported}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">judge_useful</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""[IsUse] 判断答案是否有用"""</span></span><span leaf=""><br></span><span leaf=""> answer = state[</span><span><span leaf="">"answer"</span></span><span leaf="">]</span><span leaf=""><br></span><span leaf=""> question = state[</span><span><span leaf="">"question"</span></span><span leaf="">]</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> useful_prompt = </span><span><span leaf="">f"""判断以下答案对用户问题是否有用。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">问题：</span><span><span leaf="">{question}</span></span><span leaf=""><br></span><span leaf="">答案：</span><span><span leaf="">{answer}</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">判断标准：</span><span leaf=""><br></span><span leaf="">- useful：直接回答了问题，信息完整</span><span leaf=""><br></span><span leaf="">- not_useful：没有回答问题，或答非所问</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">回答（useful/not_useful）："""</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> response = llm.invoke(useful_prompt).content</span><span leaf=""><br></span><span leaf=""> is_useful = </span><span><span leaf="">"useful"</span></span><span leaf=""> </span><span><span leaf="">in</span></span><span leaf=""> response.lower()</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"is_useful"</span></span><span leaf="">: is_useful}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">direct_generate</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""不需要检索，直接生成"""</span></span><span leaf=""><br></span><span leaf=""> prompt = </span><span><span leaf="">f"回答以下问题：</span><span><span leaf="">{state[</span><span><span leaf="">'question'</span></span><span leaf="">]}</span></span><span leaf="">"</span></span><span leaf=""><br></span><span leaf=""> answer = llm.invoke(prompt).content</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> {</span><span><span leaf="">"answer"</span></span><span leaf="">: answer, </span><span><span leaf="">"is_supported"</span></span><span leaf="">: </span><span><span leaf="">True</span></span><span leaf="">, </span><span><span leaf="">"is_useful"</span></span><span leaf="">: </span><span><span leaf="">True</span></span><span leaf="">}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">should_retrieve</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""路由：是否检索"""</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> state.get(</span><span><span leaf="">"need_retrieval"</span></span><span leaf="">, </span><span><span leaf="">True</span></span><span leaf="">):</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">"retrieve"</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">else</span></span><span leaf="">:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">"direct_generate"</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">should_regenerate</span></span><span><span leaf="">(state)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""路由：答案无支撑时是否重新生成"""</span></span><span leaf=""><br></span><span leaf=""> retry = state.get(</span><span><span leaf="">"retry_count"</span></span><span leaf="">, </span><span><span leaf="">0</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> </span><span><span leaf="">not</span></span><span leaf=""> state.get(</span><span><span leaf="">"is_supported"</span></span><span leaf="">, </span><span><span leaf="">True</span></span><span leaf="">) </span><span><span leaf="">and</span></span><span leaf=""> retry < </span><span><span leaf="">2</span></span><span leaf="">:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">"regenerate"</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">"end"</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf=""># 构建LangGraph流程图</span></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">build_self_rag_graph</span></span><span><span leaf="">()</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""构建Self-RAG的LangGraph流程"""</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> workflow = StateGraph(SelfRAGState)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 添加节点</span></span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"judge_retrieval"</span></span><span leaf="">, judge_need_retrieval)</span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"retrieve"</span></span><span leaf="">, retrieve_documents)</span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"judge_relevance"</span></span><span leaf="">, judge_relevance)</span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"generate"</span></span><span leaf="">, generate_answer)</span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"judge_support"</span></span><span leaf="">, judge_support)</span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"judge_useful"</span></span><span leaf="">, judge_useful)</span><span leaf=""><br></span><span leaf=""> workflow.add_node(</span><span><span leaf="">"direct_generate"</span></span><span leaf="">, direct_generate)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 设置入口</span></span><span leaf=""><br></span><span leaf=""> workflow.set_entry_point(</span><span><span leaf="">"judge_retrieval"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 条件路由</span></span><span leaf=""><br></span><span leaf=""> workflow.add_conditional_edges(</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"judge_retrieval"</span></span><span leaf="">,</span><span leaf=""><br></span><span leaf=""> should_retrieve,</span><span leaf=""><br></span><span leaf=""> {</span><span><span leaf="">"retrieve"</span></span><span leaf="">: </span><span><span leaf="">"retrieve"</span></span><span leaf="">, </span><span><span leaf="">"direct_generate"</span></span><span leaf="">: </span><span><span leaf="">"direct_generate"</span></span><span leaf="">},</span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> workflow.add_edge(</span><span><span leaf="">"retrieve"</span></span><span leaf="">, </span><span><span leaf="">"judge_relevance"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> workflow.add_conditional_edges(</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"judge_relevance"</span></span><span leaf="">,</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">lambda</span></span><span leaf=""> state: </span><span><span leaf="">"generate"</span></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> state[</span><span><span leaf="">"is_relevant"</span></span><span leaf="">] </span><span><span leaf="">else</span></span><span leaf=""> </span><span><span leaf="">"retrieve"</span></span><span leaf="">,</span><span leaf=""><br></span><span leaf=""> {</span><span><span leaf="">"generate"</span></span><span leaf="">: </span><span><span leaf="">"generate"</span></span><span leaf="">, </span><span><span leaf="">"retrieve"</span></span><span leaf="">: </span><span><span leaf="">"retrieve"</span></span><span leaf="">}, </span><span><span leaf=""># 不相关就重新检索</span></span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> workflow.add_edge(</span><span><span leaf="">"generate"</span></span><span leaf="">, </span><span><span leaf="">"judge_support"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> workflow.add_conditional_edges(</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"judge_support"</span></span><span leaf="">,</span><span leaf=""><br></span><span leaf=""> should_regenerate,</span><span leaf=""><br></span><span leaf=""> {</span><span><span leaf="">"regenerate"</span></span><span leaf="">: </span><span><span leaf="">"generate"</span></span><span leaf="">, </span><span><span leaf="">"end"</span></span><span leaf="">: </span><span><span leaf="">"judge_useful"</span></span><span leaf="">},</span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> workflow.add_edge(</span><span><span leaf="">"judge_useful"</span></span><span leaf="">, END)</span><span leaf=""><br></span><span leaf=""> workflow.add_edge(</span><span><span leaf="">"direct_generate"</span></span><span leaf="">, END)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> workflow.compile()</span><span leaf=""><br></span>
坑1：重试死循环。Self-RAG的判断节点可能形成循环——检索不相关→重新检索→还是不相关→重新检索……必须在路由函数里加重试次数限制。上面代码的 should_regenerate 里有 retry < 2 限制，最多重试2次。
Agentic RAG的查询路由：多数据源调度
Agentic RAG和基础RAG的另一个区别：多数据源。Agent根据问题类型选择不同的检索源。
<span><span leaf="">from</span></span><span leaf=""> enum </span><span><span leaf="">import</span></span><span leaf=""> Enum</span><span leaf=""><br></span><span><span leaf="">from</span></span><span leaf=""> pydantic </span><span><span leaf="">import</span></span><span leaf=""> BaseModel, Field</span><span leaf=""><br></span><span><span leaf="">from</span></span><span leaf=""> typing </span><span><span leaf="">import</span></span><span leaf=""> Optional</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">class</span></span><span leaf=""> </span><span><span leaf="">DataSource</span></span><span><span leaf="">(str, Enum)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""数据源类型"""</span></span><span leaf=""><br></span><span leaf=""> KNOWLEDGE_BASE = </span><span><span leaf="">"knowledge_base"</span></span><span leaf=""> </span><span><span leaf=""># 向量知识库</span></span><span leaf=""><br></span><span leaf=""> DATABASE = </span><span><span leaf="">"database"</span></span><span leaf=""> </span><span><span leaf=""># SQL数据库</span></span><span leaf=""><br></span><span leaf=""> WEB_SEARCH = </span><span><span leaf="">"web_search"</span></span><span leaf=""> </span><span><span leaf=""># 网络搜索</span></span><span leaf=""><br></span><span leaf=""> CALCULATOR = </span><span><span leaf="">"calculator"</span></span><span leaf=""> </span><span><span leaf=""># 计算器</span></span><span leaf=""><br></span><span leaf=""> DIRECT_ANSWER = </span><span><span leaf="">"direct_answer"</span></span><span leaf=""> </span><span><span leaf=""># 直接回答（不需要检索）</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">class</span></span><span leaf=""> </span><span><span leaf="">QueryRoute</span></span><span><span leaf="">(BaseModel)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""查询路由结果"""</span></span><span leaf=""><br></span><span leaf=""> source: DataSource = Field(description=</span><span><span leaf="">"数据源"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> rewritten_query: str = Field(description=</span><span><span leaf="">"针对该数据源优化后的查询"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> reason: str = Field(description=</span><span><span leaf="">"选择该数据源的理由"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">route_query</span></span><span><span leaf="">(question, llm)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""查询路由：判断问题该走哪个数据源</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 这是Agentic RAG的"大脑"——决定了信息从哪来。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> route_prompt = ChatPromptTemplate.from_template(</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""你是一个查询路由专家。请分析用户问题，选择最合适的数据源。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">可用数据源：</span><span leaf=""><br></span><span leaf="">- knowledge_base：公司内部知识库（产品文档、政策流程、技术文档）</span><span leaf=""><br></span><span leaf="">- database：业务数据库（销售数据、用户数据、财务数据，支持SQL查询）</span><span leaf=""><br></span><span leaf="">- web_search：网络搜索（实时信息、新闻、外部知识）</span><span leaf=""><br></span><span leaf="">- calculator：计算器（数值计算、统计）</span><span leaf=""><br></span><span leaf="">- direct_answer：直接回答（闲聊、通用知识）</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">用户问题：{question}</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">请选择数据源并给出优化后的查询。"""</span></span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> chain = route_prompt | llm.with_structured_output(QueryRoute)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> chain.invoke({</span><span><span leaf="">"question"</span></span><span leaf="">: question})</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf=""># 路由示例</span></span><span leaf=""><br></span><span><span leaf=""># 问题："公司退货流程是什么" → knowledge_base, "退货流程 退换货政策"</span></span><span leaf=""><br></span><span><span leaf=""># 问题："上季度华东区销售额" → database, "SELECT SUM(amount) FROM sales WHERE region='华东' AND quarter='Q3'"</span></span><span leaf=""><br></span><span><span leaf=""># 问题："最新的AI行业趋势" → web_search, "2025年AI行业发展趋势"</span></span><span leaf=""><br></span><span><span leaf=""># 问题："3.14乘以2.5" → calculator, "3.14 * 2.5"</span></span><span leaf=""><br></span><span><span leaf=""># 问题："你好" → direct_answer, ""</span></span><span leaf=""><br></span>
多工具Agent
<span><span leaf="">from</span></span><span leaf=""> langchain.agents </span><span><span leaf="">import</span></span><span leaf=""> create_agent</span><span leaf=""><br></span><span><span leaf="">from</span></span><span leaf=""> langchain_core.tools </span><span><span leaf="">import</span></span><span leaf=""> tool</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf="">@tool</span></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">search_knowledge_base</span></span><span><span leaf="">(query: str)</span></span><span leaf=""> -> str:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""搜索公司内部知识库，获取产品文档、政策流程、技术规范等信息。</span><span leaf=""><br></span><span leaf=""> 当用户问公司内部信息时使用。"""</span></span><span leaf=""><br></span><span leaf=""> results = retriever.invoke(query)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">"\n"</span></span><span leaf="">.join([r.page_content </span><span><span leaf="">for</span></span><span leaf=""> r </span><span><span leaf="">in</span></span><span leaf=""> results[:</span><span><span leaf="">5</span></span><span leaf="">]])</span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf="">@tool</span></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">query_database</span></span><span><span leaf="">(sql: str)</span></span><span leaf=""> -> str:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""执行SQL查询获取业务数据。</span><span leaf=""><br></span><span leaf=""> 当用户问销售数据、用户统计等结构化数据时使用。</span><span leaf=""><br></span><span leaf=""> 参数必须是合法的SQL语句。"""</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 实际项目对接数据库</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">try</span></span><span leaf="">:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># result = db.execute(sql)</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">f"查询结果：[模拟数据]"</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">except</span></span><span leaf=""> Exception </span><span><span leaf="">as</span></span><span leaf=""> e:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">f"查询失败：</span><span><span leaf="">{e}</span></span><span leaf="">"</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf="">@tool</span></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">search_web</span></span><span><span leaf="">(query: str)</span></span><span leaf=""> -> str:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""搜索互联网获取实时信息、新闻、外部知识。</span><span leaf=""><br></span><span leaf=""> 当用户问最新信息或知识库中没有的外部知识时使用。"""</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 实际项目对接搜索API</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">f"搜索结果：[模拟数据]"</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf="">@tool</span></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">calculate</span></span><span><span leaf="">(expression: str)</span></span><span leaf=""> -> str:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""执行数值计算。当用户需要数学计算时使用。</span><span leaf=""><br></span><span leaf=""> 参数是数学表达式，如 '3.14 * 2.5'。"""</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">try</span></span><span leaf="">:</span><span leaf=""><br></span><span leaf=""> result = eval(expression) </span><span><span leaf=""># 生产环境用ast.literal_eval或专门的解析器</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">f"计算结果：</span><span><span leaf="">{result}</span></span><span leaf="">"</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">except</span></span><span leaf=""> Exception </span><span><span leaf="">as</span></span><span leaf=""> e:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">f"计算失败：</span><span><span leaf="">{e}</span></span><span leaf="">"</span></span><span leaf=""><br></span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">create_agentic_rag</span></span><span><span leaf="">(llm)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""创建Agentic RAG Agent</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> Agent拥有多个工具，自主决定：</span><span leaf=""><br></span><span leaf=""> 1. 用哪个工具获取信息</span><span leaf=""><br></span><span leaf=""> 2. 是否需要多次检索</span><span leaf=""><br></span><span leaf=""> 3. 检索结果够不够</span><span leaf=""><br></span><span leaf=""> 4. 什么时候停止检索开始生成</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> agent = create_agent(</span><span leaf=""><br></span><span leaf=""> model=llm,</span><span leaf=""><br></span><span leaf=""> tools=[search_knowledge_base, query_database, search_web, calculate],</span><span leaf=""><br></span><span leaf=""> system_prompt=</span><span><span leaf="">"""你是一个智能助手，拥有多种信息获取工具。</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">工作流程：</span><span leaf=""><br></span><span leaf="">1. 分析用户问题的类型和所需信息</span><span leaf=""><br></span><span leaf="">2. 选择最合适的工具获取信息</span><span leaf=""><br></span><span leaf="">3. 如果第一次结果不够，换一个工具或换query再试</span><span leaf=""><br></span><span leaf="">4. 基于获取的信息生成回答</span><span leaf=""><br></span><span leaf="">5. 在回答末尾标注信息来源（知识库/数据库/网络）</span><span leaf=""><br></span><span leaf=""><br></span><span leaf="">重要规则：</span><span leaf=""><br></span><span leaf="">- 不要编造答案，所有信息必须基于工具返回的结果</span><span leaf=""><br></span><span leaf="">- 如果所有工具都找不到相关信息，明确告知用户</span><span leaf=""><br></span><span leaf="">- 最多使用5次工具调用，避免无限循环</span><span leaf=""><br></span><span leaf="">- 数值类问题用calculator，不要让LLM自己算"""</span></span><span leaf="">,</span><span leaf=""><br></span><span leaf=""> )</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> agent</span><span leaf=""><br></span>
坑2：Agent工具选择不稳定。同一个问题"上季度销售额"，Agent有时选database（对），有时选knowledge_base（错——知识库里可能有过期数据）。解决：在工具描述里写清楚适用场景，并在system_prompt里给明确的决策规则。上面代码的docstring就是给Agent看的"使用说明"。
延迟优化：Agentic RAG最大的工程挑战
Agentic RAG最大的问题是慢——基础RAG 200ms出结果，Agentic RAG可能要5-10秒。用户体验很差。
优化方案1：并行化
<span><span leaf="">import</span></span><span leaf=""> asyncio</span><span leaf=""><br></span><span leaf=""><br></span><span><span leaf="">async</span></span><span leaf=""> </span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">parallel_retrieve</span></span><span><span leaf="">(question, retrievers)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""并行检索多个数据源，取最快返回的结果</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 不等Agent决策完再检索，而是同时发起多个检索，</span><span leaf=""><br></span><span leaf=""> Agent决策完直接从已完成的检索结果里取。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> tasks = []</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">for</span></span><span leaf=""> name, retriever </span><span><span leaf="">in</span></span><span leaf=""> retrievers.items():</span><span leaf=""><br></span><span leaf=""> tasks.append(retriever.ainvoke(question))</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 并行执行，取全部结果</span></span><span leaf=""><br></span><span leaf=""> results = </span><span><span leaf="">await</span></span><span leaf=""> asyncio.gather(*tasks, return_exceptions=</span><span><span leaf="">True</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 过滤掉异常</span></span><span leaf=""><br></span><span leaf=""> valid_results = []</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">for</span></span><span leaf=""> name, result </span><span><span leaf="">in</span></span><span leaf=""> zip(retrievers.keys(), results):</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> </span><span><span leaf="">not</span></span><span leaf=""> isinstance(result, Exception):</span><span leaf=""><br></span><span leaf=""> valid_results.append((name, result))</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> valid_results</span><span leaf=""><br></span>
优化方案2：缓存+预判
<span><span leaf="">from</span></span><span leaf=""> functools </span><span><span leaf="">import</span></span><span leaf=""> lru_cache</span><span leaf=""><br></span><span><span leaf="">import</span></span><span leaf=""> hashlib</span><span leaf=""><br></span><span leaf=""><br></span><span><span><span leaf="">class</span></span><span leaf=""> </span><span><span leaf="">CachedAgenticRAG</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""带缓存的Agentic RAG</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 两层缓存：</span><span leaf=""><br></span><span leaf=""> 1. 答案缓存：相同问题直接返回（命中率高的问题重复问很多）</span><span leaf=""><br></span><span leaf=""> 2. 路由缓存：相同类型问题直接用之前的路由决策</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">__init__</span></span><span><span leaf="">(self, agent)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> self.agent = agent</span><span leaf=""><br></span><span leaf=""> self.answer_cache = {} </span><span><span leaf=""># question_hash → answer</span></span><span leaf=""><br></span><span leaf=""> self.route_cache = {} </span><span><span leaf=""># question_pattern → route</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">query</span></span><span><span leaf="">(self, question)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 1. 答案缓存</span></span><span leaf=""><br></span><span leaf=""> q_hash = hashlib.md5(question.encode()).hexdigest()</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> q_hash </span><span><span leaf="">in</span></span><span leaf=""> self.answer_cache:</span><span leaf=""><br></span><span leaf=""> print(</span><span><span leaf="">"命中答案缓存"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> self.answer_cache[q_hash], </span><span><span leaf="">0</span></span><span leaf=""> </span><span><span leaf=""># 0ms</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 2. 路由预判：根据问题模式快速选数据源</span></span><span leaf=""><br></span><span leaf=""> route = self._quick_route(question)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> route:</span><span leaf=""><br></span><span leaf=""> print(</span><span><span leaf="">f"命中路由缓存：</span><span><span leaf="">{route}</span></span><span leaf="">"</span></span><span leaf="">)</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 直接走对应数据源，跳过Agent决策</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 3. 正常Agent流程</span></span><span leaf=""><br></span><span leaf=""> answer = self.agent.invoke({</span><span><span leaf="">"messages"</span></span><span leaf="">: [{</span><span><span leaf="">"role"</span></span><span leaf="">: </span><span><span leaf="">"user"</span></span><span leaf="">, </span><span><span leaf="">"content"</span></span><span leaf="">: question}]})</span><span leaf=""><br></span><span leaf=""> result = answer[</span><span><span leaf="">"messages"</span></span><span leaf="">][</span><span><span leaf="">-1</span></span><span leaf="">].content</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 4. 写入缓存</span></span><span leaf=""><br></span><span leaf=""> self.answer_cache[q_hash] = result</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> result</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">_quick_route</span></span><span><span leaf="">(self, question)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""快速路由：基于规则匹配，不走LLM"""</span></span><span leaf=""><br></span><span leaf=""> rules = [</span><span leaf=""><br></span><span leaf=""> (</span><span><span leaf="">r"销售额|收入|利润|用户数"</span></span><span leaf="">, </span><span><span leaf="">"database"</span></span><span leaf="">),</span><span leaf=""><br></span><span leaf=""> (</span><span><span leaf="">r"最新|新闻|今天|昨天"</span></span><span leaf="">, </span><span><span leaf="">"web_search"</span></span><span leaf="">),</span><span leaf=""><br></span><span leaf=""> (</span><span><span leaf="">r"流程|政策|文档|规范"</span></span><span leaf="">, </span><span><span leaf="">"knowledge_base"</span></span><span leaf="">),</span><span leaf=""><br></span><span leaf=""> ]</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">import</span></span><span leaf=""> re</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">for</span></span><span leaf=""> pattern, route </span><span><span leaf="">in</span></span><span leaf=""> rules:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> re.search(pattern, question):</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> route</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">return</span></span><span leaf=""> </span><span><span leaf="">None</span></span><span leaf=""><br></span>
优化方案3：流式输出
<span><span leaf="">async</span></span><span leaf=""> </span><span><span><span leaf="">def</span></span><span leaf=""> </span><span><span leaf="">agentic_rag_stream</span></span><span><span leaf="">(question, agent)</span></span><span leaf="">:</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">"""流式输出：Agent决策时先输出"正在检索..."，不让用户干等</span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> 用户体验的关键不是绝对延迟，而是感知延迟。</span><span leaf=""><br></span><span leaf=""> 流式输出让用户看到进度，感知延迟降低50%。</span><span leaf=""><br></span><span leaf=""> """</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">yield</span></span><span leaf=""> </span><span><span leaf="">"正在分析问题...\n"</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># Agent第一步：路由判断</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">yield</span></span><span leaf=""> </span><span><span leaf="">"正在检索相关信息...\n"</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># Agent检索（这里简化，实际是Agent的中间步骤）</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># ...</span></span><span leaf=""><br></span><span leaf=""> </span><span leaf=""><br></span><span leaf=""> </span><span><span leaf=""># 生成答案时流式输出</span></span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">async</span></span><span leaf=""> </span><span><span leaf="">for</span></span><span leaf=""> chunk </span><span><span leaf="">in</span></span><span leaf=""> agent.astream({</span><span><span leaf="">"messages"</span></span><span leaf="">: [{</span><span><span leaf="">"role"</span></span><span leaf="">: </span><span><span leaf="">"user"</span></span><span leaf="">, </span><span><span leaf="">"content"</span></span><span leaf="">: question}]}):</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">if</span></span><span leaf=""> </span><span><span leaf="">"content"</span></span><span leaf=""> </span><span><span leaf="">in</span></span><span leaf=""> chunk:</span><span leaf=""><br></span><span leaf=""> </span><span><span leaf="">yield</span></span><span leaf=""> chunk[</span><span><span leaf="">"content"</span></span><span leaf="">]</span><span leaf=""><br></span>
CRAG vs Self-RAG：选哪个
| 维度
|
CRAG
|
Self-RAG
核心思路
|
检索后纠错
|
全流程反思
| |
判断次数
|
1次（评估检索质量）
|
3-4次（检索/相关/支撑/有用）
| |
延迟
|
中（1次检索+1次评估）
|
高（多次判断+可能重试）
| |
准确率
|
比基础RAG提升15-20%
|
比基础RAG提升20-30%
| |
工程复杂度
|
中（加一层评估+纠正）
|
高（LangGraph多节点流程）
| |
适合场景
|
知识库质量参差不齐
|
对准确率要求极高
|
我的建议：先上CRAG——在现有RAG基础上加一个检索质量评估节点，成本低效果明显。如果CRAG的准确率还不够（比如法律/医疗场景），再考虑Self-RAG。别一上来就Self-RAG，工程复杂度和延迟你都受不了。
要不要上Agentic RAG
| 你的情况
|
建议
|
原因
Advanced RAG准确率<80%
|
先别上Agentic
|
基础没打好，加Agent更乱
| |
准确率>80%但要更高
|
先上CRAG
|
加一层评估，成本低
| |
法律/医疗等高要求
|
再考虑Self-RAG
|
全流程反思，准确率最高
| |
延迟要求<2秒
|
别上Agentic
|
Agentic比基础慢10-20倍
| |
需要多数据源调度
|
上查询路由
|
投入产出比最高的一步
| |
预算有限
|
别上Self-RAG
|
一次查询3-8次LLM调用
|
本篇要点
| 要点
|
说明
Agentic RAG核心
|
自主决策检索策略 + 结果评估 + 自我纠正
| |
CRAG
|
检索结果分三档：correct直接用/incorrect换网络搜/ambiguous部分用+补充
| |
Self-RAG
|
四个反思节点：[Retrieve][IsRel][IsSup][IsUse]，全流程自我检查
| |
查询路由
|
根据问题类型选数据源：知识库/数据库/网络/计算器/直接回答
| |
多工具Agent
|
Agent拥有多个检索工具，自主决定用哪个、用几次
| |
延迟优化
|
并行检索 + 缓存 + 流式输出，把5-10秒降到可接受范围
| |
选择建议
|
先CRAG再Self-RAG，别一上来就搞复杂的
|
踩坑清单
-
重试死循环：Self-RAG的判断节点可能形成循环——不相关→重新检索→还是不相关→重新检索。必须在路由函数里加
retry_count限制，最多重试2次。 -
Agent工具选择不稳定：同一个问题，Agent有时选对工具有时选错。在工具docstring里写清楚适用场景，system_prompt里给明确决策规则。
-
延迟太大用户接受不了：Agentic RAG比基础RAG慢10-20倍。必须加流式输出让用户看到进度，加缓存减少重复计算。如果延迟还是太高，退回CRAG甚至基础RAG。
-
成本爆炸：一次查询3-8次LLM调用。1000次查询就是$30-80。用小模型做判断节点（路由、评估），大模型只做最终生成。
-
Agent调试困难：Agent的决策过程不透明，出了问题不知道哪一步走错了。用LangGraph的trace功能记录每步决策，或在每个节点加日志。
-
多数据源结果冲突：知识库说"退货时限7天"，网络搜索说"退货时限15天"。Agent不知道信谁。解决：给数据源设优先级（知识库 > 数据库 > 网络），或在system_prompt里指定冲突时的处理规则。
-
评估指标难定：Agentic RAG的评估比基础RAG复杂——除了检索准确率，还要评估Agent的决策质量（路由对不对、重试值不值）。用RAGAS + 人工审核路由日志结合评估。
下篇预告
至此RAG系列的10篇正文全部完成。下一篇是系列总结——把10篇文章的知识点串成一张图谱，给你决策树和速查表，帮你快速定位"我的RAG问题出在哪、该怎么调"。
你的RAG项目在考虑上Agentic RAG吗？还是基础RAG够用了？评论区说说你的场景。
觉得有用就点个在看，下一篇是RAG系列总结——一张图串通全流程。
- 原文作者：知识铺
- 原文链接：https://index.zshipu.com/ai001/post/20260822/%E8%AE%A9RAG%E5%AD%A6%E4%BC%9A%E8%87%AA%E5%B7%B1%E7%BA%A0%E9%94%99Agentic-RAG%E5%AE%9E%E6%88%98%E6%8C%87%E5%8D%97/
- 版权声明：本作品采用知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议进行许可，非商业转载请注明出处（作者，原文链接），商业转载请联系作者获得授权。
- 免责声明：本页面内容均来源于站内编辑发布，部分信息来源互联网，并不意味着本站赞同其观点或者证实其内容的真实性，如涉及版权等问题，请立即联系客服进行更改或删除，保证您的合法权益。转载请注明来源，欢迎对文章中的引用来源进行考证，欢迎指出任何有错误或不够清晰的表达。也可以邮件至 sblig@126.com
