基于LangGraph的Agentic RAG实战：让检索会思考、能纠错、可联网

普通RAG 用久了，谁没遇到过几个尴尬瞬间：用户问一个跨了五份合同的问题，返回的是三份文档拼出来的“缝合怪”答案；用户问“今天上午发布会公布了什么”，本地知识库里根本不可能有；多轮对话里问一句“那第二个方案呢”，检索直接把这句话扔给向量库，召回自然是一塌糊涂。这些问题不是模型笨，也不是 Embedding 质量差，而是流程结构坏了——传统 RAG 是一条笔直的单向管线，没地方插纠错逻辑，没地方做二次判断，更没地方临时调工具。

Agentic RAG 要解决的就是这件事。它把“检索”从一次快照查询，变成可规划、可执行、可验证、可重试的智能体任务。这篇文章我会用 LangGraph 从零手把手搭一套会思考、会纠错、会联网的 Agentic RAG，把路由、查询改写、知识召回、Web 搜索兜底、自反思循环这些关键结构全部落到可运行代码上。适合已经跑通过基础 RAG、想往生产可用的问答案系统再进一步的同学，也适合刚接触 LangGraph、被网上碎片化教程绕晕的新手。

1. 先聊聊“死板 RAG”到底死板在哪

1.1 传统 RAG 的链路长什么样

常规 RAG 的核心逻辑非常简单：用户提问 → 向量化 → TopK 召回 → 拼 Prompt → 让大模型根据检索片段回答。看起来没什么问题，因为在小规模文档、单轮简单问答的场景下，这条链路确实够用。但它的本质是一次性的“检索快照”，整个过程没有任何决策点。

代码上通常长这样：

from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings, ChatOpenAI # 加载向量库 vectorstore = FAISS.load_local("kb_index", OpenAIEmbeddings()) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # 检索 docs = retriever.invoke("今年公司发票报销的流程是什么？") # 拼 Prompt prompt = f""" 基于以下资料回答问题： { "".join(doc.page_content for doc in docs) } 问题：今年公司发票报销的流程是什么？ """ # 生成 llm = ChatOpenAI(model="gpt-4o-mini") answer = llm.invoke(prompt) print(answer.content)

流程是直的，代码也好懂。问题在于：如果第一轮检索就错了，整条链路没有任何补救机制。

1.2 三个让我崩溃的经典场景

我实际项目里遇到最多的是这三类：

第一类是跨文档组合问题。比如“A 项目的合同金额对比 B 项目的付款条款”，向量检索往往会把 A 和 B 的片段分别召回，但大模型只能拼出一个逻辑混乱的答案。因为传统 RAG 没有拆解问题的能力，更没有“先查 A 再查 B 最后对照”的规划能力。

第二类是实时性需求。用户问“最近一次系统故障公告发了吗”，你知识库更新到昨天，答案根本不存在。传统 RAG 只会得出一个“未找到相关内容”，实际上这种问题应该触发 Web 搜索去查官网或公告栏。

第三类是多轮对话的指代消解。用户先问“公司的年假制度是什么”，再问“那产假呢”，如果直接把“那产假呢”拿去检索，几乎必然召回错东西。正确做法要先判断“那”指代的是“公司人事制度”，再改写查询条件。

1.3 问题的本质：只检索不思考

把这三个场景放一起看，根因很清楚：传统 RAG 把“检索”当成了一个物理操作，而不是一个需要决策的行为。它不判断问题类型，不规划查询顺序，不验证答案质量，更不会在答案质量差时重新来过。

用大白话说，传统 RAG 是个执行力很强但不动脑子的工具。你把问题丢给它，它拿去向量库“捞”一段文本出来，至于捞得对不对、够不够、要不要换个方式再捞，没有任何机制在管。Agentic RAG 的本质就是给这个流程加上“大脑”——让大模型来调度检索动作，而不是只做最后一个生成步骤。

2. Agentic RAG 的“智能”到底体现在哪几个维度

2.1 从“查了再说”到“想了再查”

我理解的 Agentic RAG，不是某一篇论文里给出的固定定义，而是一类“由大模型驱动检索决策”的架构模式。最核心的变化，是把用户的问题当作一个任务来对待，而不是一条查询语句。

举个例子。用户问“对比 Python 和 Java 在内存管理上的主要差异”，传统 RAG 会把整句话送给向量库，把 TopK 片段返回给大模型。Agentic RAG 的第一步是先让 LLM 分析这个任务，拆成两个子查询：“Python 内存管理机制”和“Java 内存管理机制”，分别检索后再汇总对比。结果质量高一个档次。

2.2 Agentic RAG 的四大核心能力

我把工程上真正的 Agentic RAG 拆成四个能力维度：

• 路由（Routing）：先判断问题属于哪种类型。是事实型查询？是总结型查询？是需要联网的实时查询？还是闲聊？不同类型走不同路径，而不是一股脑全进向量库。
• 规划（Planning）：复杂问题拆成多步子任务，按顺序执行，甚至并行走多路检索。
• 工具调用（Tool Use）：把向量检索、Web 搜索、SQL 查询、文档解析都封装成工具，LLM 按需调用。
• 反思（Reflection）：生成答案后自己检查一遍，打分不合格就重新检索、重新生成，形成循环。

这四项能力不是孤立存在的，它们要通过一个可编程的流程编排框架组合起来，这就是为什么我选了 LangGraph。

2.3 一个反直觉的事实：Agentic 不等于模型越大越好

很多人以为 Agentic RAG 必须用最强的大模型，因为要“思考”。我的实测结论恰恰相反：路由、改写、打分这类任务用小模型反而更稳、更快、更便宜。

我会用 GPT-4o-mini 或者 DeepSeek 这类性价比模型做路由和反思节点，用大模型只做最后一步答案生成。原因很简单：路由和打分本质是“选择”和“判断”，不需要太强的推理能力；真正需要生成质量的是最终答案，而那个环节才值得用旗舰模型。这个取舍能省一大笔成本，还能降低延迟。

3. 为什么我选 LangGraph，而不是裸写 LangChain 链

3.1 LangChain 和 LangGraph 到底是什么关系

LangChain 最早提出 Chain 的概念，把 Prompt、LLM、Retriever 串成线性流水线。它的问题是：一旦流程出现分支、循环、条件跳转，代码会变得极其别扭。你可能会在 RunnableLambda 里塞 if-else，最后维护成本飙高。

LangGraph 是 LangChain 团队后来推出的低层编排框架，核心模型是图，而不是链。节点是函数或 Runnable，边负责控制流转方向，支持条件边、循环、人工介入、持久化。LangChain 适合快速写一条线性链；LangGraph 适合构建有状态、有分支、能反复迭代的 Agent。

一句话总结我的选型建议：如果流程直来直去，用 LangChain 就够了；如果流程里有“决定走哪条路”和“不行就重来”，上 LangGraph。

3.2 核心概念就三个：State、Node、Edge

LangGraph 上手不需要理解太复杂的东西，抓住三个概念就能干活。

• State（状态）：整个图运行期间共享的数据结构，通常是一个 TypedDict 或 Pydantic Model。所有节点读写同一份 State，这是多轮循环的基础。
• Node（节点）：一个普通的 Python 函数。入参是当前 State，返回值会合并进 State。
• Edge（边）：节点之间的连接。可以简单直连，也可以加条件判断，返回值决定下一个节点是谁。

我项目里的 State 大概长这样：

from typing import TypedDict, List class AgentState(TypedDict): question: str # 原始问题 rewritten_question: str # 改写后的查询 route: str # 路由结果 documents: List[str] # 检索到的文档 web_results: List[str] # 联网检索结果 answer: str # 最终答案 score: float # 自评分数 retries: int # 重试次数

是不是很简单？关键是这个 State 会在节点之间来回传递和更新，这让“重新检索”“修正查询”变得非常自然。

3.3 环状图才是关键：让 AI 能“反悔”

LangGraph 最打动我的能力是支持环。节点 A 到节点 B，B 发现答案不好，可以再回到节点 A，这种环状结构在 LangChain 里写起来非常痛苦，但在 LangGraph 里只是加一条条件边的事。

想一想“纠错”的本质：模型发现自己第一轮检索不对，得重新走一遍检索流程。没有环，就只能靠人工异常处理；有了环，一个条件判断就能完成流程回路。这就是 LangGraph 在 Agentic RAG 里无可替代的原因。

4. 手把手实现：会思考、会纠错、会联网的 Agentic RAG

4.1 环境准备与依赖安装

我先装依赖，建议用 Python 3.10+：

pip install langgraph langchain langchain-openai langchain-community langchain-text-splitters faiss-cpu beautifulsoup4 requests httpx

我用 FAISS 做向量库示例，OpenAI 做 Embedding 和 LLM。如果你用其他模型，替换对应类就行，图结构完全不用改。

4.2 定义全局状态 State

按照 3.2 里展示的 State 定义，我再加一个chat_history字段用来处理多轮对话：

from typing import TypedDict, List class AgentState(TypedDict): chat_history: List[tuple] # 对话历史 question: str rewritten_question: str route: str documents: List[str] web_results: List[str] answer: str score: float retries: int

这个 State 是整张图的“数据总线”。每个节点读取自己关心的字段，把输出写回 State，下个节点就能看到。

4.3 实现规划节点：查询改写与检索决策

规划节点是我所有智能性的起点。它做两件事：一是结合多轮历史改写问题，消除指代；二是判断该走本地知识库还是联网搜索。

from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate llm_planner = ChatOpenAI(model="gpt-4o-mini", temperature=0) planner_prompt = PromptTemplate.from_template(""" 你是一个检索规划器。根据对话历史和当前问题，输出一个 JSON： - rewritten_question: 适合向量库检索的独立问题 - route: "local" 或 "web" 如果问题涉及实时信息、最新公告、突发新闻，route 必须为 web。 否则 route 为 local。 对话历史： {chat_history} 当前问题： {question} 只返回 JSON，不要额外说明。 """) def plan_node(state: AgentState) -> dict: chain = planner_prompt | llm_planner resp = chain.invoke({ "chat_history": state.get("chat_history", []), "question": state["question"], }) # 这里用 json 解析，生产环境建议加 try-except 和格式校验 import json parsed = json.loads(resp.content.strip().strip("```").lstrip("json")) return { "rewritten_question": parsed["rewritten_question"], "route": parsed["route"], }

这个节点的设计思路是把“决策”收敛到一次结构化输出里，后续流程只要读route字段就能分流。

4.4 实现本地检索与知识召回

本地检索节点读取改写后的问题，去向量库做相似度检索：

from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings vectorstore = FAISS.load_local("kb_index", OpenAIEmbeddings(), allow_dangerous_deserialization=True) local_retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) def local_retrieve_node(state: AgentState) -> dict: query = state["rewritten_question"] docs = local_retriever.invoke(query) return {"documents": [doc.page_content for doc in docs]}

这里有个细节：allow_dangerous_deserialization=True是因为 FAISS 本地文件存在反序列化风险，自己项目里只在可信环境初始化时使用，不要对外部上传的索引文件直接加载。

4.5 实现 Web 搜索节点：联网兜底

联网节点本质是一个工具调用。我用一个搜索函数模拟：实际使用时会替换成你选用的搜索服务 API。

import requests def web_search(query: str, top_k: int = 5) -> list[str]: # 这里以通用搜索 API 为例，实际换成你的服务商 params = { "q": query, "count": top_k, } # resp = requests.get("https://your-search-api.example/search", params=params) # data = resp.json() # 解析结果片段，返回纯文本列表 # 演示环境先跳过真实请求，避免依赖外部服务 return [] def web_search_node(state: AgentState) -> dict: query = state["rewritten_question"] results = web_search(query) return {"web_results": results}

真实项目中，我一般用两个渠道：搜索引擎网页结果做广度召回，再抓取命中页面正文做深度阅读。这样能拿到更完整的信息。

4.6 实现回答生成与自查评估

回答生成节点把本地文档和联网结果合在一起，让 LLM 基于所有材料生成答案：

from langchain_core.runnables import RunnableLambda llm_answer = ChatOpenAI(model="gpt-4o", temperature=0.2) answer_prompt = PromptTemplate.from_template(""" 你是企业知识助手。请基于以下资料回答问题。 如果资料不足以回答，请明确说明“根据现有资料无法回答”， 不要编造。 本地资料： {context} 联网资料： {web_context} 问题： {question} 回答： """) def generate_answer_node(state: AgentState) -> dict: context = "\n".join(state.get("documents", [])) web_context = "\n".join(state.get("web_results", [])) chain = answer_prompt | llm_answer resp = chain.invoke({ "context": context, "web_context": web_context, "question": state["rewritten_question"], }) return {"answer": resp.content}

生成完答案还不够，我需要一个检查节点来判断答案可不可信。这里用一个打分节点：

judge_prompt = PromptTemplate.from_template(""" 你是一个答案质量评审员。请判断以下回答是否完整、基于给定资料、 并且能回答用户问题。输出一个 JSON，包含 score(0-1) 和 reason。 用户问题：{question} 回答：{answer} 本地资料：{context} 联网资料：{web_context} 只输出 JSON，不要额外说明。 """) def judge_node(state: AgentState) -> dict: chain = judge_prompt | llm_planner resp = chain.invoke({ "question": state["rewritten_question"], "answer": state["answer"], "context": "\n".join(state.get("documents", [])), "web_context": "\n".join(state.get("web_results", [])), }) import json parsed = json.loads(resp.content.strip().strip("```").lstrip("json")) return {"score": float(parsed["score"])}

如果打分低，就走纠错回路。

4.7 搭图：用条件边把流程串起来

这是整个实战场最核心的一段，用 LangGraph 把前面所有节点挂载成图：

from langgraph.graph import StateGraph, END def route_after_plan(state: AgentState) -> str: # 根据路由结果，决定去本地检索还是联网搜索 if state.get("route") == "web": return "web_search" return "local_retrieve" def route_after_generate(state: AgentState) -> str: # 自评分过低且重试次数未超限，则回到重新检索 if state["score"] < 0.6 and state["retries"] < 2: return "plan" # 重新规划 return END graph = StateGraph(AgentState) graph.add_node("plan", plan_node) graph.add_node("local_retrieve", local_retrieve_node) graph.add_node("web_search", web_search_node) graph.add_node("generate_answer", generate_answer_node) graph.add_node("judge", judge_node) graph.set_entry_point("plan") graph.add_edge("plan", "local_retrieve", condition=route_after_plan) graph.add_edge("local_retrieve", "generate_answer") graph.add_edge("web_search", "generate_answer") graph.add_edge("generate_answer", "judge") graph.add_edge("judge", "generate_answer", condition=route_after_generate) app = graph.compile()

注意：第三版 LangGraph 里add_edge的条件参数写法可能是add_conditional_edges，不同版本 API 有差异。我现在的写法基于较新版本，如果报add_edge() got an unexpected keyword argument 'condition'，就改成：

graph.add_conditional_edges("plan", route_after_plan, { "local_retrieve": "local_retrieve", "web_search": "web_search", })

4.8 跑通第一版

调用方式很简单：

result = app.invoke({ "chat_history": [], "question": "请对比 A 项目和 B 项目的付款条款差异", }) print(result["answer"])

如果你把verbose=True传入 compile，可以看到完整执行流程，判断到底走了本地检索还是联网搜索、重试了几次。这一步对调试极其重要。

5. 让 RAG 真正“会思考”的三个关键设计

5.1 查询路由：先判意图再决定走哪条路

很多人以为路由就是“判断需不需要联网”，实际没那么简单。我一般在规划节点里把问题分成五类：

路由类型	触发场景	处理方式
local	知识库内可回答的事实型问题	直接向量检索
web	实时信息、新闻、最新政策	走 Web 搜索
sql	需要查结构化数据的指标问题	调用 SQL 工具
summary	需要概括多个文档的问题	先检索再专门做摘要
chat	闲聊、问候、无关内容	不检索，直接回复

多分几个路由，图的维护复杂度会上升，但在真实项目里值得。原因是每一种问题的最优处理方式都不同，强行合并会两头不讨好。

5.2 多轮对话：把上下文压缩成检索条件

多轮对话最容易翻车，核心原因不是模型不行，而是把“用户问题”直接拿去做检索了。解决思路是改写，不是把历史全部塞给 LLM，而是让规划节点提炼“独立的问题描述”。

一个实用的技巧：设置 max_history=4，只保留最近几轮，避免历史太长干扰改写。同时要求改写结果必须不包含“这个”“那个”“它”这类指代词。

5.3 知识切块：被大多数人忽略的召回上限

这是我想重点提醒的地方：Agentic RAG 的“思考”能力再强，也恢复不了已经被切碎丢掉的语义信息。切块策略直接决定了召回上限。

我踩过的坑是拿固定 chunk_size 切全部文档，结果一个完整表格被切成两半，检索时只能召回表头，答案自然是空的。后来我用了“基于文档结构切块”：先按 Markdown 标题切分，再对每个段落做语义切分，表格单独处理。

from langchain_text_splitters import MarkdownHeaderTextSplitter splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "H1"), ("##", "H2"), ("###", "H3"), ] ) splits = splitter.split_text(markdown_text)

如果你存的是表格类内容，建议把整张表格转成 Markdown 或 JSON 格式后作为一个独立 chunk，不要强行切段。另一半“系列产品表格怎么存入 RAG 知识库”问的正是这个问题，我目前最稳的方案是表格一行一条结构化记录，检索时先召回相关表，再在 Prompt 里把表完整贴上。

6. 会纠错是怎么做到的：自我反思循环

6.1 简单 N 步：打分决定是否重新检索

第 4 节的代码里已经实现了纠错循环：judge 节点打分，低于 0.6 回 plan 重新规划。这个过程本质上模拟了人的行为——写完了自己检查一遍，不满意就推翻重来。

实际执行中，我不喜欢让模型只输出一个 0-1 分数，太粗糙。我的方案是让 judge 输出三个子分数：相关性、完整性、忠实度。三个分数加权平均，相关性和忠实度权重高一些。如果一个回答很流畅但内容都是模型编的，忠实度会很低，这时候必须重新检索。

6.2 避坑：死循环与成本控制

我最担心的问题不是“不纠错”，而是“无限纠错”。所以retries字段不是摆设，我的经验是单轮最多重试 2 次。理由很简单：如果第二次重新检索后评分还是低于阈值，说明问题要么超出知识库范围，要么检索本身有硬伤，再试下去只会浪费 token。

还应该加一个总步数上限，LangGraph 里可以设置recursion_limit：

result = app.invoke(initial_state, {"recursion_limit": 20})

超过步数直接抛异常，防止生产环境出现“隐形成本黑洞”。

6.3 实战中的纠错案例

我做过一个企业规章制度问答系统，用户问“迟到三次会被开除吗”。第一遍检索召回的是“员工出勤管理规定”的摘要，模型给出的答案含糊其辞。judge 模型发现摘要没有提到“开除”对应的处理等级，忠实度打分很低，就把改写后的问题改成“迟到三次 处理措施 解除劳动合同”，重新检索到了具体条款，答案变得非常精准。

这个案例的关键在于：重试不是简单重复 “检索-生成”，而是通过 judge 反馈，让规划节点知道“哪里不行”，从而改写查询词。这是 Agentic RAG 和“暴力重试”最本质的区别。

7. 会联网：从本地知识库到实时信息

7.1 为什么不能只靠本地知识库

很多知识库系统做出来之后，最大的抱怨是“回答不了新东西”。本地知识库天然是静态的，你再怎么切块也不能让向量数据库长出刚发生的新闻。但用户不在乎这些，他们默认你是一个“什么都知道的助手”。

所以 Agentic RAG 必须具备联网能力，而且不是“无脑搜”，而是只在规划节点判定需要实时信息时才去搜。这样既省成本，也避免搜索引擎噪音污染普通问答的质量。

7.2 工具封装：Web 搜索 API 作为工具

在刚才的 web_search_node 里，我只留了接口。真实项目中，我通常会把搜索封装成一个带缓存的工具类：

import time from functools import lru_cache @lru_cache(maxsize=512) def web_search_with_cache(query: str) -> tuple: # 缓存相同关键词的搜索结果，减少 API 调用 # 实际调用搜索服务，返回结果列表 pass def web_search_node(state: AgentState) -> dict: query = state["rewritten_question"] cached = web_search_with_cache(query) results = list(cached) return {"web_results": results}

缓存有个额外好处：多轮对话里用户重复问相似问题时，直接命中缓存，响应快很多。注意缓存时间设 TTL，比如 10 分钟，否则“实时”会变“过期内容”。

7.3 一次完整的“本地检索 + 联网兜底”流程

理想的生产流程是这样的：

1. 用户提问 “2025 年最新的员工福利政策有变化吗？”
2. 规划节点判断：涉及“最新”，路由到 web。
3. web 搜索返回公告正文。
4. 生成节点综合网页内容回答。
5. judge 检查信息完整性，答案带来源链接。

另一条路径：用户提问 “公司的带薪年假是几天？” 规划节点路由到 local，本地知识库直接命中，不需要联网。两条路径共用同一个生成和评估节点，这就是图结构带来的高度复用。

8. 踩坑实录与调优清单

8.1 我踩过的三个高频坑

第一个坑是State 字段名不一致导致数据丢失。LangGraph 的 State 合并逻辑依据字段名，如果你在节点里返回了rewrited_question而其他地方读取的是rewritten_question，数据不会报错，但下一环全是空值。排查方法很笨：在每个节点入口打印一下关键字段，跑一轮就能发现问题。

第二个坑是检索为空不报错。向量库没有命中时，很多 retriever 返回空列表，后续 Prompt 依然会生成一个貌似合理的幻觉答案。解决方法是加一个空文档检查节点，文档为空时直接给用户反馈“知识库未命中”，并建议走联网。

第三个坑是condition 边写错导致死循环。LangGraph 的条件边如果返回值不在映射表中，它会抛异常，但如果映射错了方向，比如把低分节点配置成“回到 generate_answer”而不是“回到 plan”，就会在同一个节点里空转。我后来统一约定：低分必须回到一个可以改变检索上下文的节点，绝不回生成节点。

8.2 效果调优的核心指标

我评估一套 Agentic RAG，不看单次回答漂不漂亮，看三个硬指标：

• 健康率：judge 打分 ≥ 0.7 的比例，低于 60% 说明检索或路由有问题。
• 重试率：一次通过的比例最好在 70% 以上，如果经常要重试，说明第一轮检索质量太差。
• 无答案反悔率：模型明确说“无法回答”的次数，不能为零。一个永远回答的 RAG 一定在幻觉。

这三个指标同时要求你有一个稳定的 judge 模型。judge 模型不要求太强，但 prompt 必须固定，最好先拿 100 条人工标注数据验证过打分稳定性。

8.3 什么时候别用 Agentic RAG

这句话可能有点泼冷水，但 Agentic RAG 不是银弹。如果你的场景只是“固定文档库里的简单 QA”，比如词典查询、参数查询，传统 RAG 加上一个不错的切块策略就够了。Agentic RAG 引入的延迟和成本，在小问题上不划算。

另外一个反例是：如果你的知识库本身质量就很差，Agentic RAG 的纠错能力也救不回来。它只能在检索环节做优化，不能让错误内容变成正确内容。

8.4 常见问题：RAG 和 MCP 怎么选

最近总有人拿 RAG 和 MCP 做对比，其实两者根本不是一层面的东西。RAG 是“怎么让模型获取知识”，MCP 是“怎么让模型使用工具”。一套 Agentic RAG 完全可以基于 MCP 协议来封装检索工具和 Web 搜索工具——用 MCP 暴露检索能力，用 LangGraph 编排调用流程，这两者天然互补，不是二选一。

我自己的经验是：如果团队已经有统一工具接入标准，用 MCP 封装检索服务会很舒服；如果只是做一个独立问答机器人，直接用 LangGraph 的函数调用就够了，没必要为引入协议而引入协议。最后再分享一个我自己调试时的习惯：给图里的每个节点加上node_time字段记录耗时，LangGraph 的状态回放功能会把每一次节点执行的输入输出都打印出来，线上出问题时不至于两眼一抹黑。这套链路我从最开始的三节点线性图，迭代到现在的路由加反思九节点图，中间改过的每个坑都是实打实的 token 学费，希望这篇能帮你省下这笔钱。