普通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(modelgpt-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、EdgeLangGraph 上手不需要理解太复杂的东西抓住三个概念就能干活。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 到节点 BB 发现答案不好可以再回到节点 A这种环状结构在 LangChain 里写起来非常痛苦但在 LangGraph 里只是加一条条件边的事。想一想“纠错”的本质模型发现自己第一轮检索不对得重新走一遍检索流程。没有环就只能靠人工异常处理有了环一个条件判断就能完成流程回路。这就是 LangGraph 在 Agentic RAG 里无可替代的原因。4. 手把手实现会思考、会纠错、会联网的 Agentic RAG4.1 环境准备与依赖安装我先装依赖建议用 Python 3.10pip 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(modelgpt-4o-mini, temperature0) 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_deserializationTrue) 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_deserializationTrue是因为 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, paramsparams) # 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(modelgpt-4o, temperature0.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, conditionroute_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, conditionroute_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])如果你把verboseTrue传入 compile可以看到完整执行流程判断到底走了本地检索还是联网搜索、重试了几次。这一步对调试极其重要。5. 让 RAG 真正“会思考”的三个关键设计5.1 查询路由先判意图再决定走哪条路很多人以为路由就是“判断需不需要联网”实际没那么简单。我一般在规划节点里把问题分成五类路由类型触发场景处理方式local知识库内可回答的事实型问题直接向量检索web实时信息、新闻、最新政策走 Web 搜索sql需要查结构化数据的指标问题调用 SQL 工具summary需要概括多个文档的问题先检索再专门做摘要chat闲聊、问候、无关内容不检索直接回复多分几个路由图的维护复杂度会上升但在真实项目里值得。原因是每一种问题的最优处理方式都不同强行合并会两头不讨好。5.2 多轮对话把上下文压缩成检索条件多轮对话最容易翻车核心原因不是模型不行而是把“用户问题”直接拿去做检索了。解决思路是改写不是把历史全部塞给 LLM而是让规划节点提炼“独立的问题描述”。一个实用的技巧设置 max_history4只保留最近几轮避免历史太长干扰改写。同时要求改写结果必须不包含“这个”“那个”“它”这类指代词。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_limitresult 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(maxsize512) 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 一次完整的“本地检索 联网兜底”流程理想的生产流程是这样的用户提问 “2025 年最新的员工福利政策有变化吗”规划节点判断涉及“最新”路由到 web。web 搜索返回公告正文。生成节点综合网页内容回答。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 学费希望这篇能帮你省下这笔钱。
