上个月我给一个老RAG项目加日志翻到一条很典型的失败记录用户问“LangGraph和LangChain到底有什么区别我该怎么选”系统检索了一堆LangChain的入门文档最后一本正经地回答了工具链的安装流程。答非所问语气还特别自信。这种场景在传统RAG里太常见了——检索一次、拼给大模型、模型硬着头皮回答整个链路没有任何反馈回路。后来我把项目重构成了Agentic RAG用LangGraph把一个“一次性检索动作”变成“可规划、可评估、可纠错、可联网”的循环工作流。这篇文章不聊概念直接给你看完整实现三个核心节点怎么写、怎么用条件边让系统自我修正、怎么接入实时搜索以及我实际踩过的坑。1. 传统RAG的“死法”清单为什么一次检索定生死注定翻车1.1 一个让人血压升高的真实问答我那个老项目是典型的三段式RAG用户输入问题向量库按相似度取Top-K文档拼进Prompt让大模型生成回答。听起来没问题但实际跑起来翻车案例一个接一个。除了上面那个LangChain vs LangGraph的例子还有更典型的用户问“2025年Q3的销量相比去年同期增长了多少”知识库索引的文档只更新到2024年。系统依然能给出一个数字因为它检索到了2024年Q3的销售报告然后模型硬凑了一个“增长”。用户要是没去核对原文就被带偏了。用户问“把这个会议纪要对预算的影响总结一下”会议纪要里提到“下季度要扩大欧洲团队”。传统向量检索把“预算”当成核心关键词召回的却是另一份预算表完全忽略了“欧洲团队扩张”这个隐含条件。这类问题的共性是一次检索定的生死检索错了后续再强的大模型也救不回来。模型不是不知道自己在胡说而是这个流程根本没给它“发现自己没答好”的机会。1.2 死板RAG的三个结构性缺陷我复盘了一下传统RAG的死板本质上是三个结构性缺陷叠加第一线性管线没有回路。Retrieve - Augment - Generate 是一条走到底的流水线。它假设“第一次检索一定是对的”但实际检索结果受Embedding模型、分块策略、查询改写质量等因素影响失败概率远比你想象的高。第二无法处理多跳问题。“比较LangGraph和LangChain的Agent能力”这种问题需要先查A再查B再对比是典型的multi-hop。传统RAG只做一轮相似度检索视角天然受限。第三没有自我评估机制。模型永远直接回答从不判断证据是否充分。这就好比一个学生考试时写完就交卷从来不检查——错了也不知道错。1.3 Agentic RAG把“检索动作”升级为“检索策略”Agentic RAG的核心变化在于把大模型从“回答器”变成“调度员”。它不是直接回答用户而是先规划要查什么、实际去查、然后评估查到的内容够不够、不够就改写查询重查、判断问题有时效性就临时联网搜索最后把收集到的证据汇总成答案。这套思路拆解下来就是四个核心能力规划Planning把用户问题拆解成多个检索子问题。评估Grading判断检索到的文档是否真的能回答问题。反思Reflection评估不通过时改写查询重新检索。工具调用Tool Use必要时调用外部API比如实时搜索。如果这几个步骤用传统代码硬写状态管理会非常痛苦。所以我选了LangGraph——它天生就是干这个的。下一个章节详细说为什么是它。2. 为什么挑LangGraph它和LangChain到底哪里不一样2.1 LangChain与LangGraph两代编排思路很多新手看到LangGraph教程都会困惑这不是LangChain出的吗为什么不直接用LangChain我直接用表格说明维度LangChainLangGraph编排方式线性链式调用、LCEL表达式有向图支持环、分支状态管理隐式传递显式State TypedDict条件分支能力有限条件边Conditional Edge检查点无Checkpointer人工介入弱支持原生支持interrupt适合场景流程固定的轻量任务复杂Agent循环工作流LangChain更像乐高积木盒里面全是好用的积木块比如LLM封装、向量库接口、Prompt模板。LangGraph则像施工图纸加上执行引擎它告诉你积木怎么连、什么时候该回头返工、状态放在哪里。两者不是替代关系而是LangGraph调用LangChain的组件来干活。我的项目就是LangGraph做骨架LangChain做工具层。2.2 理解LangGraph的五个核心概念如果你想跟人聊LangGraph或者准备面试这五个概念是绕不开的StateGraph整个工作流的“总图”所有节点和边的容器。State节点之间共享的“黑板”。它是一个TypedDict每个节点读写其中的字段后面的节点能看到前面的节点写下的内容。Node执行单元一个Python函数。函数接收当前State返回一个字典表示要对State做的修改。Edge / Conditional Edge普通边表示“无条件走到下一个节点”条件边是“根据当前State的内容决定去哪个节点”。条件边是让Agent具备自我纠错能力的开关。Checkpointer状态快照把每一步的中间状态存下来支持断点续跑、回溯、人工干预。用一个生活类比State是你书桌上的文件夹Node是不同学科的作业普通边是“做完数学做语文”条件边是“如果语文作文没写到800字就重写”。Checkpointer就是桌上的录音笔随时能回放。2.3 环境准备与最小可运行骨架先装依赖pip install langgraph langchain-openai langchain-community tavily-python numpy设置环境变量export OPENAI_API_KEYsk-xxxx export TAVILY_API_KEYtvly-xxxxLangGraph的最小可运行骨架比你想的更简单from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): count: int def plus_one(state: State): return {count: state[count] 1} graph StateGraph(State) graph.add_node(add, plus_one) graph.add_edge(START, add) graph.add_edge(add, END) app graph.compile() print(app.invoke({count: 0})) # 输出{count: 1}就这么点代码一个“加一节点”的可执行图就出来了。注意plus_one返回的是{count: ...}LangGraph会把它合并到State里。这个“返回增量修改”的模式是LangGraph的灵魂后面所有节点都遵循这个约定。3. 第一块积木Plan节点让Agent先想清楚“要查什么”3.1 让模型输出检索计划而非直接回答“会思考”的第一步是不让系统直接用用户原话去检索而是让LLM先输出一个检索计划。比如用户问“LangGraph支持哪些持久化后端和LangChain的Memory机制有什么区别”如果直接把整句话丢给向量库Top-K结果大概率是LangChain Memory的文档因为“Memory”这个关键词权重太高。但正确的检索策略应该是拆成三个独立的子查询“LangGraph checkpointer支持哪些后端”“LangChain Memory机制”“LangGraph与LangChain记忆机制对比”我把这个逻辑做成plan_nodeimport json import re from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o, temperature0) PLAN_PROMPT 你是一个检索策略规划器。 用户问题是{question} 请把该问题拆解成最多3个独立的检索子问题。 只输出JSON数组每个元素是一个检索查询字符串不要输出任何解释。 def parse_json_array(text: str): text re.sub(r(?:json)?, , text).strip() start text.find([) end text.rfind(]) return json.loads(text[start:end 1]) def plan_node(state): resp llm.invoke(PLAN_PROMPT.format(questionstate[question])) plan parse_json_array(resp.content) return {plan: plan}注意几个细节temperature0防止模型自由发挥产出不稳定的查询。要求“只输出JSON数组”方便程序解析。拆解个数限制在3个以内避免子查询太多导致下游检索成本爆炸。3.2 多路检索把计划变成文档Plan节点产生的子查询每个都要走一遍向量检索。为了演示清晰我这里用一个极简的内存向量库先用Embedding模型给文档和查询分别做向量再用余弦相似度取Top-K。import numpy as np from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 项目里的示例文档集 documents [ LangGraph 支持 MemorySaver、SqliteSaver、PostgresSaver 等检查点后端..., LangChain 的 Memory 机制通过 ConversationBufferMemory 管理历史消息..., LangGraph 的 StateGraph 支持条件边可实现动态路由..., # ... 更多业务文档 ] def cosine_similarity(a, b): return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))) def retrieve_docs(query: str, top_k: int 3): q_vec embeddings.embed_query(query) scored [] for doc in documents: d_vec embeddings.embed_query(doc) scored.append((cosine_similarity(q_vec, d_vec), doc)) scored.sort(keylambda x: x[0], reverseTrue) return [doc for _, doc in scored[:top_k]]然后retrieve_node对plan里的每个查询都做一次检索合并结果并去重def retrieve_node(state): all_docs [] for query in state[plan]: all_docs.extend(retrieve_docs(query, top_k3)) # 保留首次出现的顺序并去重 deduped list(dict.fromkeys(all_docs)) return {documents: deduped}3.3 为什么要拆解单次Top-K的视野盲区多路检索的核心价值在于打破单次Top-K的视野盲区。传统RAG一次检索只取3个片段如果这3个片段都来自同一篇文档那答案的覆盖范围就非常窄。拆解成多个子查询后每个子查询独立走一次检索相当于在不同语义方向上分别取样本。我用前面的对比问题实测过不拆解时Top-3文档全是LangChain Memory的拆解后LangGraph检查点、LangChain Memory、两者对比三个方向各有一个子查询命中对应文档最终答案质量完全不是一个量级。所以我的原则是Plan节点可以拆解到“每个子查询只负责一个孤立事实”的粒度这样才能保证下游评估节点有东西可评估。4. 第二块积木Grade节点与“不合格就重来”的反思回路4.1 把“文档质量”交给模型裁判“会纠错”的关键节点是grade_node。它负责回答一个问题当前检索到的文档到底够不够回答用户问题我让LLM输出两个字段verdict表示文档是否充分need_web表示是否因为时效性问题需要联网搜索。GRADE_PROMPT 你是检索质量评审。你的任务是判断下面的文档是否足以回答用户问题。 【用户问题】 {question} 【当前检索文档】 {documents} 请严格基于文档内容判断 1. verdict如果文档能直接或组合后回答用户问题输出 yes否则输出 no。 2. need_web如果问题疑似包含实时信息如版本更新、最新事件、2025年数据且当前文档明显滞后输出 yes否则输出 no。 只输出JSON格式{{verdict: yes|no, need_web: yes|no}} def parse_json_obj(text: str): text re.sub(r(?:json)?, , text).strip() start text.find({) end text.rfind(}) return json.loads(text[start:end 1]) def grade_node(state): docs_text \n---\n.join(state[documents]) resp llm.invoke(GRADE_PROMPT.format(questionstate[question], documentsdocs_text)) result parse_json_obj(resp.content) return { verdict: result.get(verdict, no), need_web: result.get(need_web, no), tries: state.get(tries, 0) 1, }这里有一个容易被忽视的设计tries计数放在grade_node里而不是放在检索或联网节点里。因为只有grade是每次循环必经的“裁判席”在这里计数最干净。4.2 条件边支撑的三种出口有了verdict和need_webLangGraph的条件边就能决定下一步去哪三个方向之一def decide_next(state): # 如果文档充分直接生成 if state[verdict] yes: return generate # 重试超过3次强制生成兜底避免死循环 if state[tries] 3: return generate # 问题可能有时效性且知识库不够就去联网 if state[need_web] yes: return web_search # 其余情况改写查询重新检索 return rewrite对应到图上就是三个出口generate、web_search、rewrite。其中rewrite会回到检索节点形成一个环web_search完成搜索后会回到grade_node做二次评估也形成一个环。这两个环就是“纠错”与“联网”的物理载体。4.3 让反思真正稳定的几个细节我实际调试过程中发现Grade节点如果写得不精细很容易变成摆设。下面是几个关键经验第一必须是结构化输出。如果让模型自由文本回答“文档充分吗”它可能输出“基于现有信息我认为可以……”解析非常痛苦。强制JSON格式后条件边拿到的永远是稳定的yes/no字段。第二评估标准要写具体。像“文档是否相关”这种描述太模糊模型几乎永远输出yes。我后来改成“文档是否能直接回答量化结论、对比结论或给出明确操作步骤”评估质量立刻提升。第三必须设置重试上限。没有tries上限的反思回路遇到坏文档时会无限循环费用烧到你哭。我用3次作为上限宁可最后硬答也不浪费时间和token。第四temperature0。裁判需要确定性不需要创造性。所有评估类节点我都固定用0。5. 第三块积木Web Search节点让Agent在知识库之外联网自救5.1 什么时候该联网让Grade决定而不是硬编码很多读者会想既然要联网那我干脆一开始就并排做向量检索加联网搜索不就行了我的答案是不建议。原因有三企业内部知识库可能已经覆盖90%的常见问题每次都联网纯粹浪费成本。网络搜索返回的内容质量参差不齐混进一大批不相关网页会稀释本地证据的权重。延迟上涨。每次Tavily搜索通常要1到3秒多一次搜索用户就多等几秒。正确的做法是把它作为“兜底自救”手段本地向量检索结果不合格并且模型判断问题可能有时效性时才触发联网。也就是5.2中条件边里写的那样由need_web字段动态决定。5.2 Tavily接入与文档合并Tavily是我给这个项目选的搜索API因为它是专门给LLM设计的返回的是干净网页摘要文本不需要自己写爬虫。from tavily import TavilyClient import os tavily TavilyClient(api_keyos.getenv(TAVILY_API_KEY)) def web_search_node(state): # 用原始问题来搜索而不是子查询因为实时问题通常整体理解更准 query state[question] result tavily.search(queryquery, max_results5) web_contents [item[content] for item in result[results]] # 合并到现有文档集后续grade会重新评估 combined state[documents] web_contents # 标注来源方便生成节点引用 tagged combined [[Web] item[content] for item in result[results]] return {documents: list(dict.fromkeys(tagged))}注意我做了两件事一是把网络结果拼接到本地文档后面二是给每条网络结果加上[Web]前缀。这很关键因为后续generate_node要能区分“本地知识”和“实时信息”在Prompt里告诉用户哪些来自官网、哪些来自实时搜索避免两种信息源互相打架。web_search_node执行完后图的边会把它指回grade_node让裁判再评估一次合并后的文档。如果这次够了就进入生成节点如果还不够会继续进入rewrite流程直到达到上限。5.3 联网后的成本与效果平衡加了联网之后系统立刻“活了”但成本变化也要心里有数。我实测下来每个问题平均会额外产生环节额外LLM调用额外API延迟Tavily搜索0次1-3秒二次Grade评估1次约1秒网络文档token0次但Prompt token增加生成时变慢一个原本2秒响应的RAG问题加了联网自救后可能变成5到8秒。这在可以接受的范围内但如果你的场景对延迟极其敏感建议把need_web的判断再加一层规则只有用户问题中出现“最新”“2025”“相比去年同期”等时效性关键词时才允许联网否则即使grade判了need_webyes也只走rewrite。另外max_results不要贪多我测试过5条和10条5条已经能给足关键信息10条只会让token成本翻倍。6. 拼成图完整代码、执行链路与踩坑实录6.1 把六个节点接成一张有环图到这里六个节点就齐了plan_node、retrieve_node、grade_node、web_search_node、rewrite_node、generate_node。现在把它们组装成完整的LangGraph应用。from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class RAGState(TypedDict): question: str plan: list documents: list verdict: str need_web: str generation: str tries: int def rewrite_node(state): REWRITE_PROMPT 根据检索反馈重新设计查询。 原始问题{question} 已有文档{documents} 请输出一个更精准的JSON检索查询数组最多2个查询。 docs_text \n---\n.join(state[documents]) resp llm.invoke(REWRITE_PROMPT.format( questionstate[question], documentsdocs_text[:2000] # 防止prompt过长 )) new_plan parse_json_array(resp.content) return {plan: new_plan} def generate_node(state): GENERATE_PROMPT 请基于下面的文档回答问题要求 1. 优先使用文档中的信息引用的关键数据标注文档来源。 2. 如果信息不足直接说明知识库和网络均未覆盖不要编造。 3. 语言严谨、分点回答。 【问题】{question} 【文档】{documents} docs_text \n---\n.join(state[documents]) resp llm.invoke(GENERATE_PROMPT.format( questionstate[question], documentsdocs_text )) return {generation: resp.content} graph StateGraph(RAGState) graph.add_node(plan, plan_node) graph.add_node(retrieve, retrieve_node) graph.add_node(grade, grade_node) graph.add_node(web_search, web_search_node) graph.add_node(rewrite, rewrite_node) graph.add_node(generate, generate_node) graph.add_edge(START, plan) graph.add_edge(plan, retrieve) graph.add_edge(retrieve, grade) graph.add_conditional_edges( grade, decide_next, {generate: generate, web_search: web_search, rewrite: rewrite} ) graph.add_edge(web_search, grade) graph.add_edge(rewrite, retrieve) graph.add_edge(generate, END) app graph.compile(checkpointerMemorySaver())执行config {configurable: {thread_id: demo-001}} result app.invoke({question: LangGraph和LangChain在记忆机制上的区别是什么}, config) print(result[generation])6.2 运行效果与状态可视化第一次运行时我建议先不要直接invoke而是用stream观察每个节点的输出确认流程真的在“思考”for update in app.stream( {question: LangGraph和LangChain在记忆机制上的区别是什么}, configconfig, stream_modeupdates ): print(update)输出会是类似这样的一串字典按时间顺序展示每个节点的修改结果。你可以亲眼看到plan拆解出了两个子查询grade初判“no”然后触发web_search合并网络结果后再评估变成“yes”最后generate给出回答。如果怀疑某个中间状态不对可以用get_state查看快照current_state app.get_state(config) print(current_state.values)这比打印日志好用得多因为你能直接看到整个“黑板”上每个字段的真实值。6.3 我在实战中踩过的五个坑坑一State字段漏定义导致死循环。最早我把tries写在plan_node里但plan_node只在第一次执行循环中tries一直没有递增系统但凡走一次rewrite就永远出不来。解决方法是把计数统一放在必经的grade_node上——这是LangGraph状态设计里最容易想错的地方。坑二条件边映射的字符串和节点名不一致。add_conditional_edges第三个参数的字典key是decide_next的返回值value是节点名。我写过一次把web_search误拼成web-search运行时直接报“Node web-search not found”。这种错误在纯代码里很难发现建议提前把所有节点名集中在一个常量类里。坑三多会话状态串掉。使用MemorySaver时如果不指定thread_id或者所有请求都用同一个thread_id不同用户的状态就会互相覆盖。生产环境我改成根据用户ID生成thread_id并逐步迁移到SqliteSaver做到持久化。坑四Grade prompt写得“太松”。我第一版只写了“请判断文档是否相关”结果几乎每个问题都直接yes反思和联网形同虚设。后来把prompt改成“请检查文档是否包含量化结论、版本号、对比信息或明确步骤文档缺失这些关键元素时输出no”这才让批判回路真正生效。坑五Web搜索结果和本地文档矛盾。有次本地文档说某框架支持某功能实时网页却显示该功能已经废弃。两条信息合并后生成节点“和稀泥”式地把两句话都写进答案用户反而看不懂。解决方案是在generate_node的Prompt里明确要求“如果不同来源信息冲突指出冲突并分别标注来源”让矛盾暴露出来而不是掩盖。6.4 再往深走一步这套Agentic RAG的骨架稳定之后你可以继续加装能力多Agent子图把“检索Agent”和“写作Agent”拆成两个子图各自管理一套状态通过supervisor节点调度。Human-in-the-loop在grade判定no时插入interrupt()让用户确认是否需要联网搜索避免费用失控。持久化升级把MemorySaver换成PostgresSaver支持分布式部署和会话回溯。模型替换把ChatOpenAI换成本地Ollama模型配合向量库整体私有化部署解决企业数据合规问题。按我个人经验来看Agentic RAG带来的最大提升不是某个准确率指标而是系统变“诚实”了。它不再假装自己什么都知道而是在检索不足时主动说“我需要更多信息”并且用重试和联网去补足。这种“会自救”的系统比任何精心调过的固定流程都更接近我对智能问答的预期。最后分享一个小技巧如果你第一次跑通这套流程试着把grade_node的Prompt里“verdict”改成三级打分——yes、no、partial你会发现系统对半答状态的处理会细腻很多。这个改造留给你的下一个周末去折腾。