Agentic RAG：让 Agent 自主决定何时检索什么

[进阶选读] 本篇适合已掌握基础 RAG 和 LangGraph 基础、希望构建智能化 RAG 系统的读者。Agentic RAG 让 Agent 自主判断"这个问题需要检索吗"、"检索结果够用吗"、"需要多次检索吗"，是 RAG 走向智能化的关键一步。

---

1.1 标准 RAG 的"强迫症"问题

Agentic RAG 决策流程——Agent 根据意图自主判断是否检索及选择检索策略

想象一位助手，无论你问他"今天几号？"还是"公司差旅报销流程是什么？"，他都先跑去档案室翻文件，再回来回答。这显然效率低下——有些问题根本不需要查资料。

标准 RAG 就是这样一位"强迫症助手"。它的处理逻辑是固定的：无论什么问题，先检索，再生成。这个固定管道在简单场景下工作良好，但随着企业应用复杂度提升，局限越来越明显。

具体来说，标准 RAG 存在三个核心问题：

问题一：不该检索的时候也检索。 用户问"Python 的 for 循环怎么写？"——这是公开的编程知识，任何 LLM 都掌握，根本无需检索企业知识库。强制检索不仅增加延迟，还可能检索到不相关内容干扰模型。

问题二：检索一次往往不够。 用户问"根据 Q3 财报，我们的产品线中哪条盈利增长最快，这和竞品相比如何？"——这个问题需要先检索财报数据，再检索竞品分析，两次检索的结果还需要综合推理。标准 RAG 的单次检索无法应对这类多步骤问题。

问题三：不验证检索结果是否够用。 标准 RAG 检索到什么就用什么，不会评估"这几段文字能回答用户的问题吗？还需要补充信息吗？"

Agentic RAG 正是为了解决这三个问题而生的。它的核心思想是：给 RAG 系统装上一个"大脑"，让它能自主判断、自主规划、自主验证。

---

1.2 Agentic RAG 的三个核心能力

能力一：自主判断——这个问题需要检索吗？

Agent 在执行检索前先评估问题性质。常识性问题、计算问题、编程语法问题，直接用模型自身知识回答；涉及私有数据、实时信息、专业文档的问题，才触发检索。这一步节省了大量不必要的检索开销。

能力二：自主规划——需要检索哪些内容，分几步？

对于复杂问题，Agent 会分解任务。例如"对比我们的 A 产品和竞争对手 B 产品的定价策略"，Agent 会拆分为：第一步检索内部产品定价文档，第二步检索竞品信息，第三步综合两次结果生成对比分析。

能力三：自主验证——检索结果够用吗？

每次检索后，Agent 评估："拿到的这些文档能回答用户的问题吗？"如果不够，它会调整检索策略，换个角度或换个关键词再检索，而不是直接用不充分的信息生成可能出错的答案。

---

1.3 标准 RAG vs Agentic RAG 对比

维度	标准 RAG	Agentic RAG
检索时机	每次必检索	按需检索，可跳过
检索次数	固定 1 次	动态 1～N 次
检索规划	无规划，直接用原始问题	自主分解问题，多步检索
结果验证	不验证，照单全收	评估质量，不够则继续
知识库选择	固定单一知识库	可路由到不同知识库
适用场景	简单问答，单跳推理	复杂推理，多源信息融合
实现复杂度	低，固定管道	高，需要状态机/图结构
延迟	稳定（固定开销）	可变（简单问题更快，复杂问题更慢）

---

1.4 三种 Agentic RAG 模式

1.4.1 模式一：Router 模式（路由到正确的知识库）

企业通常有多个知识库：产品文档、技术文档、HR 政策、财务数据。Router 模式的 Agent 先判断问题属于哪个领域，再路由到对应的知识库检索。

类比：去图书馆查资料，先到咨询台问"这类书在几楼"，再去对应楼层找，而不是把每层都搜一遍。

适用场景： 多知识库、多部门的企业系统；问题类型明显分类的场景。

1.4.2 模式二：Multi-hop 模式（多跳推理）

Multi-hop（多跳推理）是指第二次检索依赖第一次检索的结果。"跳"是一个比喻，就像在信息之间跳格子——从一个信息跳到下一个关联信息。

示例： 用户问"负责 A 项目的团队，他们最近提交的需求文档在哪里？"

• 第一跳：检索"A 项目负责团队是谁" → 得到"B 团队"
• 第二跳：用"B 团队 + 需求文档"再检索 → 得到具体文档

适用场景： 需要通过中间结果推导最终答案的问题；知识库中有引用关系的文档体系。

1.4.3 模式三：Iterative 模式（迭代检索）

Iterative（迭代）模式是"检索 → 评估 → 不够用 → 再检索"的循环。每次检索后，Agent 评估信息的完整性和相关性，直到满足条件才停止。

这就像写论文时查文献——查到一篇，发现它引用了另一篇更重要的文章，于是再去找那篇，如此反复，直到觉得资料充分为止。

适用场景： 问题复杂度未知，无法预判需要几次检索；需要高质量、高置信度输出的场景。

---

1.5 Agentic RAG 决策流程图

---

1.6 用 LangGraph 实现 Agentic RAG

LangGraph 是 LangChain 生态中专门用于构建有状态 Agent 工作流的框架（详见第 5 章 LangGraph 基础）。它用"节点"（每个决策步骤）和"边"（步骤之间的流转逻辑）描述 Agent 的决策流程，天然适合实现 Agentic RAG。

以下为代码示例，非程序员可跳过代码，重点看文字说明。

整体设计思路：

• 每个决策步骤是一个"节点"（Node）
• 节点之间的跳转逻辑是"条件边"（Conditional Edge）
• 整个流程的运行状态存储在一个"状态字典"中

from typing import TypedDict, Literal, List
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.messages import HumanMessage, SystemMessage

# ============================================================
# 第一步：定义状态结构
# 状态就像一个"工作台"，记录整个检索过程中的所有信息
# ============================================================
class AgenticRAGState(TypedDict):
    question: str                  # 用户的原始问题
    needs_retrieval: bool          # 是否需要检索
    retrieved_docs: List[str]      # 已检索到的文档内容
    retrieval_quality: str         # 检索质量评估结果: "sufficient" / "insufficient"
    iteration_count: int           # 已迭代次数（防止无限循环）
    search_query: str              # 当前使用的检索关键词
    answer: str                    # 最终生成的答案

# ============================================================
# 第二步：初始化工具
# ============================================================
llm = ChatOpenAI(model="gpt-4o", temperature=0)
embeddings = OpenAIEmbeddings()

# 假设已有一个向量数据库（见第01章：如何构建向量库）
vectorstore = Chroma(
    collection_name="company_docs",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# ============================================================
# 第三步：定义各个节点函数
# ============================================================

def assess_retrieval_need(state: AgenticRAGState) -> AgenticRAGState:
    """
    节点1：判断是否需要检索
    LLM评估问题类型，决定是否触发检索流程
    """
    response = llm.invoke([
        SystemMessage(content="""你是一个智能路由助手。
        判断用户问题是否需要检索企业知识库。
        需要检索的情况：涉及公司内部政策、产品信息、业务流程、私有数据。
        不需要检索的情况：通用常识、数学计算、编程语法、历史知识。
        只回答 yes 或 no。"""),
        HumanMessage(content=f"问题：{state['question']}")
    ])

    needs_retrieval = response.content.strip().lower() == "yes"
    return {**state, "needs_retrieval": needs_retrieval, "iteration_count": 0}


def generate_search_query(state: AgenticRAGState) -> AgenticRAGState:
    """
    节点2：生成优化的检索关键词
    将用户问题转化为更适合向量检索的查询语句
    """
    response = llm.invoke([
        SystemMessage(content="将用户问题转化为适合文档检索的简洁关键词，只输出关键词，不要解释。"),
        HumanMessage(content=state["question"])
    ])
    search_query = response.content.strip()
    return {**state, "search_query": search_query}


def retrieve_documents(state: AgenticRAGState) -> AgenticRAGState:
    """
    节点3：执行检索
    """
    docs = retriever.invoke(state["search_query"])
    doc_contents = [doc.page_content for doc in docs]
    # 合并新旧检索结果（迭代检索时会累积）
    all_docs = state.get("retrieved_docs", []) + doc_contents
    return {**state, "retrieved_docs": all_docs}


def evaluate_retrieval_quality(state: AgenticRAGState) -> AgenticRAGState:
    """
    节点4：评估检索质量
    判断已有文档是否足以回答用户问题
    """
    docs_text = "\n---\n".join(state["retrieved_docs"])

    response = llm.invoke([
        SystemMessage(content="""评估检索到的文档是否足以回答用户问题。
        如果文档包含足够的相关信息，回答 sufficient。
        如果文档信息不足、不相关或缺少关键内容，回答 insufficient。
        只回答 sufficient 或 insufficient。"""),
        HumanMessage(content=f"用户问题：{state['question']}\n\n检索到的文档：\n{docs_text}")
    ])

    quality = response.content.strip().lower()
    return {**state, "retrieval_quality": quality}


def generate_answer(state: AgenticRAGState) -> AgenticRAGState:
    """
    节点5：生成最终答案
    """
    docs_text = "\n---\n".join(state.get("retrieved_docs", []))
    context_prompt = f"参考文档：\n{docs_text}\n\n" if docs_text else ""

    response = llm.invoke([
        SystemMessage(content=f"你是企业知识库助手。根据提供的文档回答问题，如信息不完整请说明。{context_prompt}"),
        HumanMessage(content=state["question"])
    ])
    return {**state, "answer": response.content}


def refine_search_query(state: AgenticRAGState) -> AgenticRAGState:
    """
    节点6：优化检索策略（用于迭代检索）
    在前一次检索不够时，换一个角度重新生成查询词
    """
    response = llm.invoke([
        SystemMessage(content="前次检索结果不足以回答问题。请从不同角度生成新的检索关键词。只输出关键词。"),
        HumanMessage(content=f"原始问题：{state['question']}\n已用查询词：{state['search_query']}")
    ])
    new_query = response.content.strip()
    return {
        **state,
        "search_query": new_query,
        "iteration_count": state["iteration_count"] + 1
    }

# ============================================================
# 第四步：定义条件边（决策逻辑）
# ============================================================

def route_after_assessment(state: AgenticRAGState) -> Literal["generate_search_query", "generate_answer"]:
    """判断是否跳过检索直接生成"""
    if state["needs_retrieval"]:
        return "generate_search_query"
    return "generate_answer"


def route_after_evaluation(state: AgenticRAGState) -> Literal["generate_answer", "refine_search_query"]:
    """判断是否需要继续检索"""
    MAX_ITERATIONS = 3  # 最多迭代3次，防止无限循环
    if state["retrieval_quality"] == "sufficient" or state["iteration_count"] >= MAX_ITERATIONS:
        return "generate_answer"
    return "refine_search_query"

# ============================================================
# 第五步：构建图结构
# ============================================================
graph_builder = StateGraph(AgenticRAGState)

# 添加节点
graph_builder.add_node("assess_retrieval_need", assess_retrieval_need)
graph_builder.add_node("generate_search_query", generate_search_query)
graph_builder.add_node("retrieve_documents", retrieve_documents)
graph_builder.add_node("evaluate_retrieval_quality", evaluate_retrieval_quality)
graph_builder.add_node("generate_answer", generate_answer)
graph_builder.add_node("refine_search_query", refine_search_query)

# 添加边（流程连接）
graph_builder.add_edge(START, "assess_retrieval_need")
graph_builder.add_conditional_edges("assess_retrieval_need", route_after_assessment)
graph_builder.add_edge("generate_search_query", "retrieve_documents")
graph_builder.add_edge("retrieve_documents", "evaluate_retrieval_quality")
graph_builder.add_conditional_edges("evaluate_retrieval_quality", route_after_evaluation)
graph_builder.add_edge("refine_search_query", "retrieve_documents")  # 迭代回到检索节点
graph_builder.add_edge("generate_answer", END)

# 编译成可执行的图
agentic_rag = graph_builder.compile()

# ============================================================
# 第六步：使用
# ============================================================
result = agentic_rag.invoke({
    "question": "我们公司差旅报销的上限是多少，出租车票需要发票吗？",
    "retrieved_docs": [],
    "search_query": "",
    "answer": ""
})

print(result["answer"])

代码关键点说明：

route_after_assessment 函数是第一个关键决策点——它检查 needs_retrieval 字段，决定是直接调用 LLM 生成答案（绕过整个检索流程），还是进入检索规划环节。

route_after_evaluation 函数是第二个关键决策点——检索质量评估后，如果文档不够用且迭代次数未超限，就回到检索节点（refine_search_query → retrieve_documents），形成一个循环；直到质量评估通过或达到最大迭代次数才退出循环生成答案。

MAX_ITERATIONS = 3 是一个安全阀，防止 Agent 陷入无限检索循环，这在实际生产中非常重要。

---

1.7 与第 11 章 Self-RAG、CRAG 的关系

本系列第 11 章介绍了 Self-RAG 和 CRAG，第 8 章前沿方向也涉及 Agentic RAG 的概念。这几篇文章有所重叠，但侧重点不同：

• 第 11 章（Self-RAG/CRAG）：聚焦"自我纠错"机制，即如何评估和修正检索结果的质量；更偏学术原理。
• 第 8 章前沿方向：从宏观角度介绍 Agentic RAG 的发展趋势和前沿进展。
• 本章（第 15 章）：聚焦工程实现，用 LangGraph 构建完整的 Agentic RAG 系统；更偏生产落地。

三者结合看，能对 Agentic RAG 形成完整认识：为什么要做（第 11 章的问题）→ 未来方向（第 8 章）→ 怎么做（本章）。

---

1.8 实战案例：企业内部问答系统

以下是一个接近真实场景的企业问答系统实现，展示 Agentic RAG 如何处理不同类型的问题。

以下为代码示例，非程序员可跳过代码，重点看文字说明。

# 场景模拟：HR 助手，对接公司 HR 知识库
# 展示三种不同问题的处理路径

test_questions = [
    # 场景1：不需要检索（LLM直接回答）
    "Python 列表和元组的区别是什么？",

    # 场景2：简单检索（一次检索即可）
    "公司的年假天数政策是什么？",

    # 场景3：多跳推理（需要多步检索）
    "上海研发中心的试用期考核标准和北京总部一样吗？",
]

for question in test_questions:
    print(f"\n{'='*50}")
    print(f"问题：{question}")

    result = agentic_rag.invoke({
        "question": question,
        "retrieved_docs": [],
        "search_query": "",
        "answer": ""
    })

    print(f"是否触发检索：{result['needs_retrieval']}")
    print(f"检索迭代次数：{result['iteration_count']}")
    print(f"答案：{result['answer'][:200]}...")

实际运行效果说明：

第一个问题（Python 语法）——Agent 判断不需要检索，needs_retrieval=False，直接由 LLM 回答，整个流程跳过了向量数据库，延迟从约 1.5 秒降低到约 0.5 秒。

第二个问题（年假政策）——触发检索，一次检索即评估为"足够"，iteration_count=0，正常流程完成。

第三个问题（跨地区对比）——触发检索，第一次只检索到上海政策，评估为"不够"，第二次补充检索北京政策，两次结果合并后生成对比答案，iteration_count=1。

---

1.9 注意事项与最佳实践

关于延迟： Agentic RAG 的延迟不稳定。简单问题因跳过检索而更快，复杂问题因多轮检索而更慢。建议在 UI 上展示"正在思考/正在检索"的进度提示，提升用户体验。

关于成本： 每次评估检索质量都要调用一次 LLM，成本比标准 RAG 高。可以用小模型（如 GPT-4o-mini）做评估，用大模型只做最终答案生成，以此控制成本。

关于幻觉防控： 迭代上限（MAX_ITERATIONS）非常重要。超过上限后应明确告知用户"信息不足，以下答案可能不完整"，而不是用不足的信息强行生成看起来完整的答案。

关于适用场景： 不是所有 RAG 系统都需要升级为 Agentic RAG。如果业务场景简单、问题类型固定，标准 RAG 的稳定性和可预测性反而是优点。Agentic RAG 适合：问题类型多样、需要多步推理、有多个知识库需要路由的场景。

---

Agentic RAG 要解决的核心问题是：标准 RAG 不会思考。它不管问什么都先检索，检索到什么都往 Prompt 里塞，不验证够不够用。加上决策能力后，它能判断要不要检索、检索什么、检索够了吗——对应 Router、Multi-hop、Iterative 三种模式。

LangGraph 的图结构适合做这件事，节点执行、条件边决策、状态字典传上下文。