让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：差在哪

核心权衡：Agentic RAG用更多的时间和成本换取更高的准确率。适合对准确率要求高、对延迟不敏感的场景（如企业知识库问答、法律咨询）。不适合实时对话场景（延迟要求<2秒）。

Corrective RAG（CRAG）：检索结果打分 + 自动纠正

CRAG是2024年提出的方案，核心思路：给检索结果打分，分三档处理。

检索结果 → 质量评估 → ① Correct（相关）：直接用
                    → ② Incorrect（不相关）：丢掉，去网络搜索
                    → ③ Ambiguous（不确定）：部分用 + 网络搜索补充

质量评估：怎么判断检索结果好不好

from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from enum import Enum

class RetrievalGrade(str, Enum):
    """检索结果质量等级"""
    CORRECT = "correct"      # 相关：直接用
    INCORRECT = "incorrect"  # 不相关：丢弃
    AMBIGUOUS = "ambiguous"  # 不确定：部分使用

class GradeResult(BaseModel):
    """检索结果评估"""
    grade: RetrievalGrade = Field(description="检索结果质量等级")
    reason: str = Field(description="判断理由")
    relevant_parts: str = Field(description="如果ambiguous，哪些部分是相关的", default="")

def grade_retrieval(question, retrieved_docs, llm):
    """评估检索结果质量

    CRAG的核心：不是所有检索结果都值得用。
    先让LLM判断检索结果跟问题的相关度，再决定怎么处理。
    """
    grade_prompt = ChatPromptTemplate.from_template(
        """你是一个检索质量评估专家。请判断以下检索结果是否能回答用户问题。

用户问题：{question}

检索结果：
{documents}

判断标准：
- correct：检索结果直接包含回答问题所需的信息
- incorrect：检索结果跟问题完全无关
- ambiguous：检索结果部分相关，但不完整或不确定

{format_instructions}"""
    )

    # 拼接检索结果
    doc_text = "\n---\n".join([
        f"[文档{i+1}] {doc.page_content}"
        for i, doc in enumerate(retrieved_docs)
    ])

    from langchain_core.output_parsers import PydanticOutputParser
    parser = PydanticOutputParser(pydantic_object=GradeResult)

    chain = grade_prompt | llm | parser
    result = chain.invoke({
        "question": question,
        "documents": doc_text,
        "format_instructions": parser.get_format_instructions(),
    })

    return result

CRAG的纠正策略

def crag_search(question, retriever, llm, web_search_fn=None):
    """Corrective RAG：检索 → 评估 → 纠正

    三档处理：
    - correct → 知识转换（提取关键信息，去噪）
    - incorrect → 丢弃，用网络搜索补充
    - ambiguous → 提取相关部分 + 网络搜索补充
    """
    # Step 1: 初始检索
    retrieved_docs = retriever.invoke(question)

    # Step 2: 质量评估
    grade = grade_retrieval(question, retrieved_docs, llm)

    print(f"检索评估: {grade.grade} - {grade.reason}")

    if grade.grade == RetrievalGrade.CORRECT:
        # 相关：知识转换（提取关键信息，去掉噪声）
        refined_docs = knowledge_transformation(retrieved_docs, question, llm)
        return refined_docs, "knowledge_base"

    elif grade.grade == RetrievalGrade.INCORRECT:
        # 不相关：丢弃，去网络搜索
        if web_search_fn:
            print("检索结果不相关，启用网络搜索...")
            web_results = web_search_fn(question)
            return web_results, "web_search"
        else:
            return [], "no_result"

    else:  # AMBIGUOUS
        # 部分相关：提取相关部分 + 网络搜索补充
        relevant_docs = [doc for doc in retrieved_docs
                        if any(keyword in doc.page_content
                               for keyword in grade.relevant_parts.split())]

        if web_search_fn:
            web_results = web_search_fn(question)
            return relevant_docs + web_results, "hybrid"
        else:
            return relevant_docs, "partial"

def knowledge_transformation(docs, question, llm):
    """知识转换：提取跟问题最相关的信息，去掉噪声

    CRAG的一个细节：即使是"correct"的检索结果，
    也不是整篇用，而是提取最相关的部分。
    """
    transform_prompt = ChatPromptTemplate.from_template(
        """从以下文档中提取与问题最相关的信息，去掉无关内容。

问题：{question}

文档：
{documents}

提取关键信息（保留原文表述，不要改写）："""
    )

    doc_text = "\n".join([doc.page_content for doc in docs])
    chain = transform_prompt | llm

    refined = chain.invoke({"question": question, "documents": doc_text})

    # 返回精炼后的文档
    from langchain_core.documents import Document
    return [Document(page_content=refined.content, metadata={"source": "crag_refined"})]

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的推理流程

用户问题 → [Retrieve]判断 → 需要检索 → 检索 → [IsRel]判断 → 相关 → 生成 → [IsSup]判断 → 有支撑 → [IsUse]判断 → 有用 → 输出
                            → 不需要   → 直接生成 → 输出
                                       → 不相关   → 重新检索或直接生成
                                                    → 无支撑   → 重新生成

工程实现：用LangGraph模拟Self-RAG

真正的Self-RAG需要微调一个能输出反思token的模型。但工程上我们可以用LangGraph模拟这个流程——用普通LLM做判断节点：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class SelfRAGState(TypedDict):
    """Self-RAG的状态"""
    question: str                              # 用户问题
    need_retrieval: bool                       # 是否需要检索
    documents: Annotated[list, operator.add]   # 检索到的文档
    is_relevant: bool                          # 检索结果是否相关
    answer: str                                # 生成的答案
    is_supported: bool                         # 答案是否有文档支撑
    is_useful: bool                            # 答案是否有用
    retry_count: int                           # 重试次数

def judge_need_retrieval(state):
    """[Retrieve] 判断是否需要检索

    简单问题（闲聊、通用知识）不需要检索，
    复杂问题（公司政策、产品细节）需要检索。
    """
    question = state["question"]

    judge_prompt = f"""判断以下问题是否需要从知识库检索信息。

问题：{question}

判断标准：
- 需要：涉及具体事实、公司政策、产品细节、技术文档
- 不需要：闲聊、通用知识、观点表达

回答（需要/不需要）："""

    response = llm.invoke(judge_prompt).content
    need = "需要" in response

    return {"need_retrieval": need, "retry_count": state.get("retry_count", 0)}

def retrieve_documents(state):
    """执行检索"""
    docs = retriever.invoke(state["question"])
    return {"documents": docs}

def judge_relevance(state):
    """[IsRel] 判断检索结果是否相关"""
    grade = grade_retrieval(state["question"], state["documents"], llm)
    is_relevant = grade.grade != RetrievalGrade.INCORRECT
    return {"is_relevant": is_relevant}

def generate_answer(state):
    """生成答案"""
    context = "\n".join([doc.page_content for doc in state["documents"]])

    prompt = f"""基于以下上下文回答问题。如果上下文不足以回答，请说明。

上下文：
{context}

问题：{state["question"]}

回答："""

    answer = llm.invoke(prompt).content
    return {"answer": answer}

def judge_support(state):
    """[IsSup] 判断答案是否有文档支撑

    关键：防止LLM编造答案。
    检查答案中的每个论断是否都能在检索到的文档中找到依据。
    """
    answer = state["answer"]
    docs = state["documents"]
    doc_text = "\n".join([doc.page_content for doc in docs])

    support_prompt = f"""判断以下答案是否被文档支撑。

文档：
{doc_text}

答案：
{answer}

判断标准：
- supported：答案中的信息都能在文档中找到依据
- unsupported：答案包含文档中没有的信息（可能是编造的）

回答（supported/unsupported）："""

    response = llm.invoke(support_prompt).content
    is_supported = "supported" in response.lower()

    return {"is_supported": is_supported}

def judge_useful(state):
    """[IsUse] 判断答案是否有用"""
    answer = state["answer"]
    question = state["question"]

    useful_prompt = f"""判断以下答案对用户问题是否有用。

问题：{question}
答案：{answer}

判断标准：
- useful：直接回答了问题，信息完整
- not_useful：没有回答问题，或答非所问

回答（useful/not_useful）："""

    response = llm.invoke(useful_prompt).content
    is_useful = "useful" in response.lower()

    return {"is_useful": is_useful}

def direct_generate(state):
    """不需要检索，直接生成"""
    prompt = f"回答以下问题：{state['question']}"
    answer = llm.invoke(prompt).content
    return {"answer": answer, "is_supported": True, "is_useful": True}

def should_retrieve(state):
    """路由：是否检索"""
    if state.get("need_retrieval", True):
        return "retrieve"
    else:
        return "direct_generate"

def should_regenerate(state):
    """路由：答案无支撑时是否重新生成"""
    retry = state.get("retry_count", 0)
    if not state.get("is_supported", True) and retry 2:
        return "regenerate"
    return "end"

# 构建LangGraph流程图
def build_self_rag_graph():
    """构建Self-RAG的LangGraph流程"""

    workflow = StateGraph(SelfRAGState)

    # 添加节点
    workflow.add_node("judge_retrieval", judge_need_retrieval)
    workflow.add_node("retrieve", retrieve_documents)
    workflow.add_node("judge_relevance", judge_relevance)
    workflow.add_node("generate", generate_answer)
    workflow.add_node("judge_support", judge_support)
    workflow.add_node("judge_useful", judge_useful)
    workflow.add_node("direct_generate", direct_generate)

    # 设置入口
    workflow.set_entry_point("judge_retrieval")

    # 条件路由
    workflow.add_conditional_edges(
        "judge_retrieval",
        should_retrieve,
        {"retrieve": "retrieve", "direct_generate": "direct_generate"},
    )

    workflow.add_edge("retrieve", "judge_relevance")

    workflow.add_conditional_edges(
        "judge_relevance",
        lambda state: "generate" if state["is_relevant"] else "retrieve",
        {"generate": "generate", "retrieve": "retrieve"},  # 不相关就重新检索
    )

    workflow.add_edge("generate", "judge_support")

    workflow.add_conditional_edges(
        "judge_support",
        should_regenerate,
        {"regenerate": "generate", "end": "judge_useful"},
    )

    workflow.add_edge("judge_useful", END)
    workflow.add_edge("direct_generate", END)

    return workflow.compile()

坑1：重试死循环。Self-RAG的判断节点可能形成循环——检索不相关→重新检索→还是不相关→重新检索……必须在路由函数里加重试次数限制。上面代码的 should_regenerate 里有 retry < 2 限制，最多重试2次。

Agentic RAG的查询路由：多数据源调度

Agentic RAG和基础RAG的另一个区别：多数据源。Agent根据问题类型选择不同的检索源。

from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional

class DataSource(str, Enum):
    """数据源类型"""
    KNOWLEDGE_BASE = "knowledge_base"  # 向量知识库
    DATABASE = "database"              # SQL数据库
    WEB_SEARCH = "web_search"          # 网络搜索
    CALCULATOR = "calculator"          # 计算器
    DIRECT_ANSWER = "direct_answer"    # 直接回答（不需要检索）

class QueryRoute(BaseModel):
    """查询路由结果"""
    source: DataSource = Field(description="数据源")
    rewritten_query: str = Field(description="针对该数据源优化后的查询")
    reason: str = Field(description="选择该数据源的理由")

def route_query(question, llm):
    """查询路由：判断问题该走哪个数据源

    这是Agentic RAG的"大脑"——决定了信息从哪来。
    """
    route_prompt = ChatPromptTemplate.from_template(
        """你是一个查询路由专家。请分析用户问题，选择最合适的数据源。

可用数据源：
- knowledge_base：公司内部知识库（产品文档、政策流程、技术文档）
- database：业务数据库（销售数据、用户数据、财务数据，支持SQL查询）
- web_search：网络搜索（实时信息、新闻、外部知识）
- calculator：计算器（数值计算、统计）
- direct_answer：直接回答（闲聊、通用知识）

用户问题：{question}

请选择数据源并给出优化后的查询。"""
    )

    chain = route_prompt | llm.with_structured_output(QueryRoute)
    return chain.invoke({"question": question})

# 路由示例
# 问题："公司退货流程是什么" → knowledge_base, "退货流程 退换货政策"
# 问题："上季度华东区销售额" → database, "SELECT SUM(amount) FROM sales WHERE region='华东' AND quarter='Q3'"
# 问题："最新的AI行业趋势" → web_search, "2025年AI行业发展趋势"
# 问题："3.14乘以2.5" → calculator, "3.14 * 2.5"
# 问题："你好" → direct_answer, ""

多工具Agent

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search_knowledge_base(query: str) -> str:
    """搜索公司内部知识库，获取产品文档、政策流程、技术规范等信息。
    当用户问公司内部信息时使用。"""
    results = retriever.invoke(query)
    return "\n".join([r.page_content for r in results[:5]])

@tool
def query_database(sql: str) -> str:
    """执行SQL查询获取业务数据。
    当用户问销售数据、用户统计等结构化数据时使用。
    参数必须是合法的SQL语句。"""
    # 实际项目对接数据库
    try:
        # result = db.execute(sql)
        return f"查询结果：[模拟数据]"
    except Exception as e:
        return f"查询失败：{e}"

@tool
def search_web(query: str) -> str:
    """搜索互联网获取实时信息、新闻、外部知识。
    当用户问最新信息或知识库中没有的外部知识时使用。"""
    # 实际项目对接搜索API
    return f"搜索结果：[模拟数据]"

@tool
def calculate(expression: str) -> str:
    """执行数值计算。当用户需要数学计算时使用。
    参数是数学表达式，如 '3.14 * 2.5'。"""
    try:
        result = eval(expression)  # 生产环境用ast.literal_eval或专门的解析器
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算失败：{e}"

def create_agentic_rag(llm):
    """创建Agentic RAG Agent

    Agent拥有多个工具，自主决定：
    1. 用哪个工具获取信息
    2. 是否需要多次检索
    3. 检索结果够不够
    4. 什么时候停止检索开始生成
    """
    agent = create_agent(
        model=llm,
        tools=[search_knowledge_base, query_database, search_web, calculate],
        system_prompt="""你是一个智能助手，拥有多种信息获取工具。

工作流程：
1. 分析用户问题的类型和所需信息
2. 选择最合适的工具获取信息
3. 如果第一次结果不够，换一个工具或换query再试
4. 基于获取的信息生成回答
5. 在回答末尾标注信息来源（知识库/数据库/网络）

重要规则：
- 不要编造答案，所有信息必须基于工具返回的结果
- 如果所有工具都找不到相关信息，明确告知用户
- 最多使用5次工具调用，避免无限循环
- 数值类问题用calculator，不要让LLM自己算""",
    )

    return agent

坑2：Agent工具选择不稳定。同一个问题"上季度销售额"，Agent有时选database（对），有时选knowledge_base（错——知识库里可能有过期数据）。解决：在工具描述里写清楚适用场景，并在system_prompt里给明确的决策规则。上面代码的docstring就是给Agent看的"使用说明"。

延迟优化：Agentic RAG最大的工程挑战

Agentic RAG最大的问题是慢——基础RAG 200ms出结果，Agentic RAG可能要5-10秒。用户体验很差。

优化方案1：并行化

import asyncio

async def parallel_retrieve(question, retrievers):
    """并行检索多个数据源，取最快返回的结果

    不等Agent决策完再检索，而是同时发起多个检索，
    Agent决策完直接从已完成的检索结果里取。
    """
    tasks = []
    for name, retriever in retrievers.items():
        tasks.append(retriever.ainvoke(question))

    # 并行执行，取全部结果
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 过滤掉异常
    valid_results = []
    for name, result in zip(retrievers.keys(), results):
        if not isinstance(result, Exception):
            valid_results.append((name, result))

    return valid_results

优化方案2：缓存+预判

from functools import lru_cache
import hashlib

class CachedAgenticRAG:
    """带缓存的Agentic RAG

    两层缓存：
    1. 答案缓存：相同问题直接返回（命中率高的问题重复问很多）
    2. 路由缓存：相同类型问题直接用之前的路由决策
    """

    def __init__(self, agent):
        self.agent = agent
        self.answer_cache = {}      # question_hash → answer
        self.route_cache = {}       # question_pattern → route

    def query(self, question):
        # 1. 答案缓存
        q_hash = hashlib.md5(question.encode()).hexdigest()
        if q_hash in self.answer_cache:
            print("命中答案缓存")
            return self.answer_cache[q_hash], 0  # 0ms

        # 2. 路由预判：根据问题模式快速选数据源
        route = self._quick_route(question)
        if route:
            print(f"命中路由缓存：{route}")
            # 直接走对应数据源，跳过Agent决策

        # 3. 正常Agent流程
        answer = self.agent.invoke({"messages": [{"role": "user", "content": question}]})
        result = answer["messages"][-1].content

        # 4. 写入缓存
        self.answer_cache[q_hash] = result

        return result

    def _quick_route(self, question):
        """快速路由：基于规则匹配，不走LLM"""
        rules = [
            (r"销售额|收入|利润|用户数", "database"),
            (r"最新|新闻|今天|昨天", "web_search"),
            (r"流程|政策|文档|规范", "knowledge_base"),
        ]

        import re
        for pattern, route in rules:
            if re.search(pattern, question):
                return route
        return None

优化方案3：流式输出

async def agentic_rag_stream(question, agent):
    """流式输出：Agent决策时先输出"正在检索..."，不让用户干等

    用户体验的关键不是绝对延迟，而是感知延迟。
    流式输出让用户看到进度，感知延迟降低50%。
    """
    yield "正在分析问题...\n"

    # Agent第一步：路由判断
    yield "正在检索相关信息...\n"

    # Agent检索（这里简化，实际是Agent的中间步骤）
    # ...

    # 生成答案时流式输出
    async for chunk in agent.astream({"messages": [{"role": "user", "content": question}]}):
        if "content" in chunk:
            yield chunk["content"]

CRAG vs Self-RAG：选哪个

我的建议：先上CRAG——在现有RAG基础上加一个检索质量评估节点，成本低效果明显。如果CRAG的准确率还不够（比如法律/医疗场景），再考虑Self-RAG。别一上来就Self-RAG，工程复杂度和延迟你都受不了。

要不要上Agentic RAG

本篇要点

踩坑清单

1. 重试死循环：Self-RAG的判断节点可能形成循环——不相关→重新检索→还是不相关→重新检索。必须在路由函数里加 retry_count 限制，最多重试2次。

1. Agent工具选择不稳定：同一个问题，Agent有时选对工具有时选错。在工具docstring里写清楚适用场景，system_prompt里给明确决策规则。

1. 延迟太大用户接受不了：Agentic RAG比基础RAG慢10-20倍。必须加流式输出让用户看到进度，加缓存减少重复计算。如果延迟还是太高，退回CRAG甚至基础RAG。

1. 成本爆炸：一次查询3-8次LLM调用。1000次查询就是$30-80。用小模型做判断节点（路由、评估），大模型只做最终生成。

1. Agent调试困难：Agent的决策过程不透明，出了问题不知道哪一步走错了。用LangGraph的trace功能记录每步决策，或在每个节点加日志。

1. 多数据源结果冲突：知识库说"退货时限7天"，网络搜索说"退货时限15天"。Agent不知道信谁。解决：给数据源设优先级（知识库 > 数据库 > 网络），或在system_prompt里指定冲突时的处理规则。

1. 评估指标难定：Agentic RAG的评估比基础RAG复杂——除了检索准确率，还要评估Agent的决策质量（路由对不对、重试值不值）。用RAGAS + 人工审核路由日志结合评估。

下篇预告

至此RAG系列的10篇正文全部完成。下一篇是系列总结——把10篇文章的知识点串成一张图谱，给你决策树和速查表，帮你快速定位"我的RAG问题出在哪、该怎么调"。

___

你的RAG项目在考虑上Agentic RAG吗？还是基础RAG够用了？评论区说说你的场景。

觉得有用就点个在看，下一篇是RAG系列总结——一张图串通全流程。