第一次听说 LangGraph 的时候我其实没有太当回事。当时手里已经用 LangChain 写了不少自动化流程链式调用用得也算顺手觉得无非又是生态里多了一个新概念。真正开始被它吸引是某次在做多步骤任务编排的时候被一个非常基础的问题反复折磨流程走到一半怎么让系统根据上一步的结果决定接下来走哪条路是循环调用工具还是直接结束用传统链式写法代码会越堆越别扭判断逻辑散落在各个环节状态也理不清楚。后来认真把 LangGraph 源码和官方文档啃了一遍才发现这东西解决的不只是“分支判断”这么简单——它等于把一套完整的状态机机制引入了 AI 应用开发让 Agent 的行为变得可控、可回溯、可持久化。这篇就聊聊我对 LangGraph 的理解和实际使用经验从核心概念到工程落地尽量把你需要知道的东西讲透。1. 从链式调用到状态图LangGraph 到底解决了什么问题1.1 链式调用的线性局限先回忆一下我们最熟悉的方式。用 LangChain 写一个简单的问答流程通常是这样加载文档、切块、向量化、检索、扔给大模型生成答案。每一步的输出是下一步的输入整条链路是固定的、单向的、线性的代码看起来也足够清晰。但一旦任务复杂起来问题就出现了。假设你要做一个能够自主决定“要不要调用联网搜索、要不要查数据库、要不要重写问题”的 Agent这个时候流程不再是直线了而是带条件分支、带循环、甚至需要多轮工具调用才能收敛的图结构。你当然可以用 if-else 硬写但每加一个分支代码的可读性就下降一截。更麻烦的是状态管理每一轮执行到哪了、已经调用过哪些工具、当前会话的上下文是什么这些信息如果不集中管理很快就会失控。LangGraph 的核心理念就是把这些流程显式地建模成一张图。节点是你要执行的动作边是状态之间的转移条件状态则贯穿整个图的运行周期。听起来像传统的状态机但它和 LLM 应用结合在一起之后能做的东西就远比传统状态机灵活得多。1.2 用图结构管理 Agent 运行状态我们可以用一个生活化的类比来理解。做饭的时候流程通常不是一条直线走到黑的。你要先看看冰箱里有什么菜再决定做什么菜、需不需要下楼买调料甚至可能炒到一半发现锅不够用得调整菜谱。每一步都依赖当前的“厨房状态”食材列表、炊具情况、时间预算。如果你把所有决策都用 if-else 硬编码写在一条长长的菜谱里换一个人来就很难维护。LangGraph 的做法是把“厨房状态”定义成一个结构化的对象把每个动作洗菜、切菜、炒菜、决定下一步定义成节点再把节点之间的转换关系定义成边。这样整个烹饪过程就是状态在这些节点之间流动的过程。图结构的好处是天然支持分支、回环、暂停和恢复而且状态是唯一的、可序列化的随时可以保存下来下次接着用。1.3 LangGraph 的定位与适用人群LangGraph 并不是 LangChain 的替代品而是它面向复杂 Agent 场景的补充。如果你只是做简单的文档问答、基础提示词模板LangChain 足够用一旦你想实现一个自主规划、多工具调度、可人工介入、需要记忆上下文的智能体不上图结构管理基本会在某个阶段被复杂度反噬。适合学习 LangGraph 的读者大概是这几类已经用过 LangChain、对 Agent 概念有基础认识但苦于流程编排的朋友准备做生产级智能体应用、需要持久化和断点容错的工程师以及对状态机模式感兴趣、想看看如何把传统软件工程思路搬进 LLM 应用里的开发者。这篇文章会尽量把基础概念讲到小白也能理解同时保留足够的深度给有经验的读者。2. LangGraph 五大核心概念拆解2.1 State所有节点共享的一块黑板理解 LangGraph第一个要抓住的概念就是 State。它是贯穿整个图运行过程的全局数据结构所有节点都可以读取它、修改它、并且把修改后的内容传回给图的状态管理器。在 LangGraph 中State 通常用 TypedDict 定义。比如from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): question: str messages: Annotated[List[str], operator.add] intermediate_steps: List[dict] final_answer: str这段代码里有几个细节值得注意。question和final_answer是普通的字段后一个节点写进去会把前一个节点的值覆盖。而messages字段用了Annotated[List[str], operator.add]这是 LangGraph 的约定了这个字段的合并方式是“把新值追加到旧值后面”而不是替换。也就是说多个节点往messages里写内容时它们会自动累积成列表。这种设计思路我特别喜欢因为它把状态更新的语义显式地放在了类型定义层面。你不需要在每个节点里手动把旧消息取出来再拼接框架帮你把常见的 reducer 逻辑自动化了。我一开始踩过一个很小的坑忘了给messages加 reducer结果多次节点运行后前面的消息全部被覆盖了。后来才意识到普通的字段语义是“写入”带 reducer 的字段语义才是“累积”。如果你希望某个状态字段在多个节点之间共享并持续增长记得加上合适的 reducer。2.2 Node单个业务动作的封装单位节点就是图里的一个函数。它接收当前完整的 State执行一段逻辑然后返回一个字典字典里是要更新的字段值。一个最基本的节点长这样def retrieve_node(state: AgentState) - dict: question state[question] docs retriever.invoke(question) return {intermediate_steps: [{type: retrieve, content: docs}]}函数签名很干净入参是 State出参是状态更新。好处是每个节点都是独立的、可测试的单元。你可以单独调用一个节点函数传入一个模拟的 State检查它返回的更新是否符合预期不需要把整张图跑起来。我在实际项目里的习惯是一个节点只做一件事。比如专门负责改写问题专门负责调用检索专门负责生成最终回复。这不只是代码风格问题而是状态流可观测性的基础。节点越细出错时就越容易定位是哪一步坏了。很多 LangGraph 示例项目喜欢把几个动作合在一个节点里以图省事但工程上还是建议拆开。2.3 Edge连接节点与条件路由的“水管”边决定了节点之间的流转方式。LangGraph 里有两种边普通边和条件边。普通边表示固定转移A 结束后一定走 B。条件边则引入了一段判断函数它接收 State返回下一个要进入的节点名称。这个函数可以用任意 Python 逻辑实现也可以调用大模型来做决策这也是 LangGraph 能表达“Agent 自主决定行为”的关键。from langgraph.graph import END def route_after_tools(state: AgentState) - str: if state.get(need_more_info, False): return retrieve_node return generate_node条件边返回的是字符串这个字符串必须和图中注册的节点名严格对应。如果你返回了不存在的节点名运行时会直接报错这点要注意。我习惯把所有节点名定义成常量避免手打字符串出错。2.4 StateGraph构图、编译与运行的完整流程把节点和边组装到一起的就是 StateGraph。构建流程分成四步创建图对象、添加节点、添加边、编译。from langgraph.graph import StateGraph graph StateGraph(AgentState) graph.add_node(query_rewrite, rewrite_node) graph.add_node(retrieve_node, retrieve_node) graph.add_node(generate_node, generate_node) graph.set_entry_point(query_rewrite) graph.add_edge(query_rewrite, retrieve_node) graph.add_conditional_edges( retrieve_node, route_after_tools, {retrieve_node: retrieve_node, generate_node: generate_node} ) graph.add_edge(generate_node, END) compiled_graph graph.compile()compile 之后得到的对象可以直接调用。编译过程会做校验所有边的目标节点是否存在、是否设置了入口点、是否可能有无法到达的死角。在开发和调试阶段会把明显错误前置暴露出来而不是运行到一半才爆炸。运行方式也很简单result compiled_graph.invoke({question: 什么是状态机})invoke 传入的就是初始 State 里需要填充的字段返回值是最终 State。如果想拿中间过程LangGraph 是支持开启 stream 模式的后面我也会讲到调试时怎么用这个能力。2.5 和 LangChain 生态的关系不是替代是升级LangGraph 在底层复用了很多 LangChain 的概念和组件工具接口、消息抽象、模型封装等。它的节点函数里完全可以自由使用 LangChain 的链、模型调用、工具装饰器。两者更像是同一个生态在两个不同复杂度层次上的呈现。简单说LangChain 提供了丰富的积木LangGraph 则提供了把这些积木灵活组合成“图”的框架。我的建议是不要把它们割裂开看在实际项目中经常是混着用的。3. 手把手构建一个可复用的检索问答 Agent3.1 场景设定与整体架构这部分我们用一个小而完整的案例来走一遍流程。场景是这样的构建一个能够围绕内部知识库做问答的 Agent它具备两个行为第一直接基于已有知识回答第二如果判断知识不足或需要联网补充就调用检索工具再基于检索结果回答。这个 Agent 需要支持多轮对话并能记录当前用户问题的历史上下文。整体图结构设计如下入口节点检查是否需要检索条件分支需要则进入检索节点不需要则进入生成节点检索节点调用向量检索工具并把结果写入状态生成节点整合历史消息和检索结果调用大模型生成答案结束节点输出最终答案这个结构很简单却是 LangGraph 条件路由和状态管理的最小完整案例后续无论多复杂的 Agent 都是在此基础上的扩展。3.2 定义状态与检索工具先定义状态和工具函数。我用的是伪代码风格重点在于逻辑结构清晰你可以直接替换成自己环境里的实现。from typing import TypedDict, Annotated, List import operator class ChatState(TypedDict): question: str chat_history: Annotated[List[dict], operator.add] retrieved_docs: List[str] needs_search: bool answer: str def check_search_node(state: ChatState) - dict: last_message state[chat_history][-1] if state[chat_history] else needs_search should_search(state[question], last_message) return {needs_search: needs_search} def retrieve_node(state: ChatState) - dict: docs knowledge_retriever(state[question]) return {retrieved_docs: docs} def generate_node(state: ChatState) - dict: context \n\n.join(state.get(retrieved_docs, [])) if not context: reply normal_llm_chat(state[question]) else: reply rag_llm_chat(state[question], context) return {answer: reply}这里should_search可以是一个简单规则比如检测到“最新”“实时”“内部文档没提到”等信号也可以用一个小模型做意图判断。在实际生产项目中我倾向于用后者准确率高很多。3.3 编排条件路由循环接下来把节点组装成图并加上条件路由。这里有一个易踩的坑LangGraph 的条件边是我们自己写判断逻辑判断逻辑不要在多个节点里各处都有尽量集中在条件函数里否则状态流会变成一团乱麻。from langgraph.graph import StateGraph, END def decide_next(state: ChatState) - str: if not state[needs_search]: return generate_node return retrieve_node graph StateGraph(ChatState) graph.add_node(check_search_node, check_search_node) graph.add_node(retrieve_node, retrieve_node) graph.add_node(generate_node, generate_node) graph.set_entry_point(check_search_node) graph.add_conditional_edges( check_search_node, decide_next, {generate_node: generate_node, retrieve_node: retrieve_node} ) graph.add_edge(retrieve_node, generate_node) graph.add_edge(generate_node, END) app graph.compile()这里有一个细节条件边的映射字典里键是decide_next返回的字符串值才是真正的节点名。如果你希望返回值和节点名保持一致可以直接这样写。我自己的习惯是让返回值和节点名解耦这样以后改节点名时不用动判断函数安全性高一些。3.4 编译运行与参数调整运行这段代码result app.invoke({ question: 我们的退款政策最近有什么变化, chat_history: [], }) print(result[answer])如果你开启了检索模式最终答案会综合知识库内容和问题来生成。这里有一个参数上的取舍需要说明retrieved_docs的条数、chat_history保留的长度都会直接影响上下文窗口占用和模型成本。我在实际项目中会把检索出的文档数限制在 4 到 6 条历史消息限制在最近 10 轮并且按 token 截断。控制上下文长度不是可选项是生产级应用必须考虑的。另外一个小技巧invoke 支持传入config参数里面可以带recursion_limit控制图最多执行多少步。这个参数在调试无限循环时特别好用比如一场意外的循环卡死如果不设上限请求会挂很久。默认上限足够但我在复杂图结构里都会主动设一个更小的值比如 20让异常快速暴露。4. 持久化、Checkpoint 与断点续跑4.1 记忆从哪来Checkpoint 机制解读单纯的图执行是一次性的invoke 结束所有状态都丢了。但实际业务里用户不可能每次都把所有历史重新说一遍。LangGraph 提供的答案是 Checkpoint。Checkpoint 机制简单说就是图上每执行到一个节点框架就把当前完整 State 保存成一份快照。这份快照存在内存或数据库里关联一个线程 ID。下次要恢复这个会话只要传入同一个线程 ID图就能从上次中断处继续跑。这会带来两个很明显的好处。第一是多轮对话的天然记忆上一轮的状态在下一轮自动可用。第二是容错如果某一步执行失败或用户主动终止了流程后续可以从断点恢复不用重头开始。这在长任务场景里价值巨大。4.2 多轮会话的 thread_id 用法使用 Checkpoint 并不需要大幅改代码主要是在编译和调用时多传一个配置config {configurable: {thread_id: user_123_session_456}} result1 app.invoke({question: 你们的退货周期是几天}, configconfig) result2 app.invoke({question: 如果超过这个时间怎么办}, configconfig)第二次调用时输入的 State 会自动带上第一轮产生的历史消息。你会发现chat_history字段里自动积累了前面的内容而你的生成节点只需要正常读取即可。这个机制解决了我之前手动维护历史记录的大量样板代码。使用 thread_id 有个原则需要想清楚thread_id 是线程的标识不是用户 ID。一个用户可以有多个线程比如围绕不同问题开的会话。把用户 ID 和会话 ID 拼起来作为 thread_id 是比较常见的做法。如果只拿用户 ID 当 thread_id会导致所有历史混在一个会话里后续状态会越来越乱。除了内存级 Checkpoint 外LangGraph 也支持把状态存到外部持久化存储里。这样即使服务重启历史也能恢复。我建议生产环境一定要配置外部持久化否则服务一挂用户的历史对话全没了体验会很糟糕。4.3 Human-in-the-loop让人插一脚的设计Checkpoint 机制还解锁了一个冷门但重要的能力人机协同。LangGraph 允许你在图中设置断点比如在执行某个敏感动作之前暂停等人工确认后再继续。这非常适合审核、审批、危险操作确认等场景。实现思路大概是在编译时把某个节点设置成断点节点运行时图执行到该节点前会自动暂停并返回当前待执行步骤的信息。你可以自行做人工审核流程然后再调接口让图继续跑。我在实际项目里用这个模式实现过一个生成内容的发布前审核流程大模型生成初稿进入审核节点时暂停运营人员在后台查看、修改或退回确认后图继续执行发布动作。整个过程用户无感知但对业务方来说风险控制能力增强了一个量级。5. 常见问题与排查技巧实录5.1 三个高频踩坑场景场景一状态字段被覆盖而不是累积这是最容易踩的坑。如果某个字段需要由多个节点往里追加内容比如历史消息列表一定要在定义 State 时给它加上Annotated和 reducer。否则每个节点写入时都会整体覆盖最终只保留最后一个节点的内容。排查这种问题可以把最终 State 打印出来看看字段值是预期的累积列表还是只有单条内容。场景二条件路由返回了不存在的节点名条件边函数返回的内容如果拼错了一个字母运行时会直接报错 “Node not found”。这种问题在字符串没有常量管理时非常容易发生而且错误信息只有在实际执行到该条件边时才出现。我强烈建议把所有节点名字定义成常量并且条件函数返回时直接用常量不要手写字符串字面量。场景三循环节点停不下来某些业务流程里为了处理复杂任务会让图在“检索—生成—工具调用—再检索”之间循环。如果循环终止条件写得太宽松或依赖的模型判断不稳定就可能出现死循环。LangGraph 有 recursion_limit但默认值很高不会及时触发。排查时先把 recursion_limit 调小一些比如 20让循环快速中断再检查终止条件。另一个思路是在状态里加一个计数器字段超过阈值强制跳转到最终节点这是工程上比较稳的兜底方案。5.2 调试建议与日志技巧调试 LangGraph 应用有一个明显优势图结构本身就是很好的调试地图。出错时可以先确认是哪两个节点之间的转换出了问题再针对节点单独测试。我在项目里通常会在每个节点的入口打一条结构化日志记录当前节点名和关键状态字段的摘要在节点出口再打一条日志记录返回的更新内容。这样整个执行路径一目了然。不要只记录节点名还要记录状态里的关键值否则日志只能告诉你“走到哪了”不能告诉你“为什么走了这条路”。LangGraph 还提供了app.stream()方法可以逐节点流式输出执行过程。调试阶段我会经常用它替代 invoke能直观看到每一步的状态变化for event in app.stream({question: ...}, configconfig): for node_name, state in event.items(): print(f节点: {node_name}) print(f状态摘要: {state})流式执行在线上也能派上用场用户侧可以实时看到 Agent 当前在做什么体验上会“智能感”强很多。但要注意不要直接把完整状态流给用户里面可能包含检索中间结果、内部判断逻辑容易暴露实现细节。5.3 问题速查表现象可能原因排查思路历史消息不累积State 字段没配 reducer检查 TypedDict 中的 Annotated 定义条件路由报节点不存在返回字符串与节点名不一致用常量管理节点名图运行超时循环终止条件过弱调小 recursion_limit加计数器兜底多轮对话串号thread_id 设计粒度不对检查是否把用户 ID 直接当 thread_id某节点状态为 None 引发异常前置节点未返回该字段在节点入口加字段默认值修改 State 后下游读不到新值返回的字段名与状态字段不同检查节点返回的 dict 键名这张速查表基本覆盖了早期使用 LangGraph 的高频问题。很多问题形式上五花八门本质都是对 State 更新语义或节点命名约定理解不到位。6. 工具选型解析与进阶方向6.1 什么场景必须上 LangGraph不是所有项目都需要 LangGraph。判断标准主要看三点有没有分支决策、有没有循环依赖、需不需要持久化和断点恢复。如果你的应用只是“接一个模型给一个答案”LangChain 或其他轻量封装完全够用。一旦你的 Agent 需要根据中间结果决定下一步、需要调用多个工具且调用的顺序不固定、需要在长时间任务中反复比较和筛选信息那么图结构的价值就体现出来了。从维护角度说图结构让流程可见。一份代码接手过来只要画出节点和边的连接关系就能理解业务逻辑。而 if-else 嵌套的链式代码这种可见性是很差的。团队协作时这个优势会越来越明显。6.2 与其他编排方案的对比市面上做 Agent 流程编排的选项不少大致有三类思路。一类是在原生 Python 代码里自己维护状态机和循环一类是用 LangGraph 这类专门的图编排库另一类是拖拽式流程编排平台。单说 LangGraph 和自维护代码的区别。自维护代码最大的问题是状态不集中、恢复逻辑要自己造轮子、异常处理分散。你可能自己写出了一个能用的流程但等到要加“断点恢复”“人工审批”“历史会话”就会发现每一块都要手搓。LangGraph 把这些从框架层面解决了代码量不一定会少但复杂度可控得多。与其他图编排库相比LangGraph 的优势是离 LangChain 生态近、对 AI 场景的抽象更贴切比如消息、工具、模型调用这些概念都是内置的。局限是它要求你按照它的思维模式去设计状态对没有状态机经验的开发者来说有一定的学习成本。6.3 可以继续扩展的方向LangGraph 本身的能力边界还在快速扩展。几个方向值得大家保持关注。一个是多智能体协作模式。单个 Agent 处理不了的任务可以拆成多个子 Agent 并行或接力完成。LangGraph 的状态机制很自然地支持这种组织结构每个子图可以作为一个节点挂在更大的图上。另一个是与外部事件流的集成。结合消息队列和事件触发器图节点可以由外部事件激活而不再局限于用户请求驱动。这个方向比较适合构建自动化运维、异常处理这类响应式系统。还有一个是图的动态生成。目前多数应用的图结构是静态的开发时写死了。如果能让图结构随着任务变化而动态生成通用性和灵活度会大幅提升。这也意味着 Agent 的自主规划能力会上一个新台阶当然工程达成难度也要高不少。最后再分享一个小技巧。如果你刚开始学 LangGraph不要一上来就追求“大而全”的复杂图先拿一个三四个节点的最小流程图跑通再逐步加入条件路由、持久化、人工介入这些机制。状态机的思维和传统面向过程编程很不一样需要一段时间适应。等你想清楚“状态是什么、变化如何发生、谁能改变状态”这三件事LangGraph 就不再是一个需要学的新框架而是一个顺手的表达工具。
第一次听说 LangGraph 的时候我其实没有太当回事。当时手里已经用 LangChain 写了不少自动化流程链式调用用得也算顺手觉得无非又是生态里多了一个新概念。真正开始被它吸引是某次在做多步骤任务编排的时候被一个非常基础的问题反复折磨流程走到一半怎么让系统根据上一步的结果决定接下来走哪条路是循环调用工具还是直接结束用传统链式写法代码会越堆越别扭判断逻辑散落在各个环节状态也理不清楚。后来认真把 LangGraph 源码和官方文档啃了一遍才发现这东西解决的不只是“分支判断”这么简单——它等于把一套完整的状态机机制引入了 AI 应用开发让 Agent 的行为变得可控、可回溯、可持久化。这篇就聊聊我对 LangGraph 的理解和实际使用经验从核心概念到工程落地尽量把你需要知道的东西讲透。1. 从链式调用到状态图LangGraph 到底解决了什么问题1.1 链式调用的线性局限先回忆一下我们最熟悉的方式。用 LangChain 写一个简单的问答流程通常是这样加载文档、切块、向量化、检索、扔给大模型生成答案。每一步的输出是下一步的输入整条链路是固定的、单向的、线性的代码看起来也足够清晰。但一旦任务复杂起来问题就出现了。假设你要做一个能够自主决定“要不要调用联网搜索、要不要查数据库、要不要重写问题”的 Agent这个时候流程不再是直线了而是带条件分支、带循环、甚至需要多轮工具调用才能收敛的图结构。你当然可以用 if-else 硬写但每加一个分支代码的可读性就下降一截。更麻烦的是状态管理每一轮执行到哪了、已经调用过哪些工具、当前会话的上下文是什么这些信息如果不集中管理很快就会失控。LangGraph 的核心理念就是把这些流程显式地建模成一张图。节点是你要执行的动作边是状态之间的转移条件状态则贯穿整个图的运行周期。听起来像传统的状态机但它和 LLM 应用结合在一起之后能做的东西就远比传统状态机灵活得多。1.2 用图结构管理 Agent 运行状态我们可以用一个生活化的类比来理解。做饭的时候流程通常不是一条直线走到黑的。你要先看看冰箱里有什么菜再决定做什么菜、需不需要下楼买调料甚至可能炒到一半发现锅不够用得调整菜谱。每一步都依赖当前的“厨房状态”食材列表、炊具情况、时间预算。如果你把所有决策都用 if-else 硬编码写在一条长长的菜谱里换一个人来就很难维护。LangGraph 的做法是把“厨房状态”定义成一个结构化的对象把每个动作洗菜、切菜、炒菜、决定下一步定义成节点再把节点之间的转换关系定义成边。这样整个烹饪过程就是状态在这些节点之间流动的过程。图结构的好处是天然支持分支、回环、暂停和恢复而且状态是唯一的、可序列化的随时可以保存下来下次接着用。1.3 LangGraph 的定位与适用人群LangGraph 并不是 LangChain 的替代品而是它面向复杂 Agent 场景的补充。如果你只是做简单的文档问答、基础提示词模板LangChain 足够用一旦你想实现一个自主规划、多工具调度、可人工介入、需要记忆上下文的智能体不上图结构管理基本会在某个阶段被复杂度反噬。适合学习 LangGraph 的读者大概是这几类已经用过 LangChain、对 Agent 概念有基础认识但苦于流程编排的朋友准备做生产级智能体应用、需要持久化和断点容错的工程师以及对状态机模式感兴趣、想看看如何把传统软件工程思路搬进 LLM 应用里的开发者。这篇文章会尽量把基础概念讲到小白也能理解同时保留足够的深度给有经验的读者。2. LangGraph 五大核心概念拆解2.1 State所有节点共享的一块黑板理解 LangGraph第一个要抓住的概念就是 State。它是贯穿整个图运行过程的全局数据结构所有节点都可以读取它、修改它、并且把修改后的内容传回给图的状态管理器。在 LangGraph 中State 通常用 TypedDict 定义。比如from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): question: str messages: Annotated[List[str], operator.add] intermediate_steps: List[dict] final_answer: str这段代码里有几个细节值得注意。question和final_answer是普通的字段后一个节点写进去会把前一个节点的值覆盖。而messages字段用了Annotated[List[str], operator.add]这是 LangGraph 的约定了这个字段的合并方式是“把新值追加到旧值后面”而不是替换。也就是说多个节点往messages里写内容时它们会自动累积成列表。这种设计思路我特别喜欢因为它把状态更新的语义显式地放在了类型定义层面。你不需要在每个节点里手动把旧消息取出来再拼接框架帮你把常见的 reducer 逻辑自动化了。我一开始踩过一个很小的坑忘了给messages加 reducer结果多次节点运行后前面的消息全部被覆盖了。后来才意识到普通的字段语义是“写入”带 reducer 的字段语义才是“累积”。如果你希望某个状态字段在多个节点之间共享并持续增长记得加上合适的 reducer。2.2 Node单个业务动作的封装单位节点就是图里的一个函数。它接收当前完整的 State执行一段逻辑然后返回一个字典字典里是要更新的字段值。一个最基本的节点长这样def retrieve_node(state: AgentState) - dict: question state[question] docs retriever.invoke(question) return {intermediate_steps: [{type: retrieve, content: docs}]}函数签名很干净入参是 State出参是状态更新。好处是每个节点都是独立的、可测试的单元。你可以单独调用一个节点函数传入一个模拟的 State检查它返回的更新是否符合预期不需要把整张图跑起来。我在实际项目里的习惯是一个节点只做一件事。比如专门负责改写问题专门负责调用检索专门负责生成最终回复。这不只是代码风格问题而是状态流可观测性的基础。节点越细出错时就越容易定位是哪一步坏了。很多 LangGraph 示例项目喜欢把几个动作合在一个节点里以图省事但工程上还是建议拆开。2.3 Edge连接节点与条件路由的“水管”边决定了节点之间的流转方式。LangGraph 里有两种边普通边和条件边。普通边表示固定转移A 结束后一定走 B。条件边则引入了一段判断函数它接收 State返回下一个要进入的节点名称。这个函数可以用任意 Python 逻辑实现也可以调用大模型来做决策这也是 LangGraph 能表达“Agent 自主决定行为”的关键。from langgraph.graph import END def route_after_tools(state: AgentState) - str: if state.get(need_more_info, False): return retrieve_node return generate_node条件边返回的是字符串这个字符串必须和图中注册的节点名严格对应。如果你返回了不存在的节点名运行时会直接报错这点要注意。我习惯把所有节点名定义成常量避免手打字符串出错。2.4 StateGraph构图、编译与运行的完整流程把节点和边组装到一起的就是 StateGraph。构建流程分成四步创建图对象、添加节点、添加边、编译。from langgraph.graph import StateGraph graph StateGraph(AgentState) graph.add_node(query_rewrite, rewrite_node) graph.add_node(retrieve_node, retrieve_node) graph.add_node(generate_node, generate_node) graph.set_entry_point(query_rewrite) graph.add_edge(query_rewrite, retrieve_node) graph.add_conditional_edges( retrieve_node, route_after_tools, {retrieve_node: retrieve_node, generate_node: generate_node} ) graph.add_edge(generate_node, END) compiled_graph graph.compile()compile 之后得到的对象可以直接调用。编译过程会做校验所有边的目标节点是否存在、是否设置了入口点、是否可能有无法到达的死角。在开发和调试阶段会把明显错误前置暴露出来而不是运行到一半才爆炸。运行方式也很简单result compiled_graph.invoke({question: 什么是状态机})invoke 传入的就是初始 State 里需要填充的字段返回值是最终 State。如果想拿中间过程LangGraph 是支持开启 stream 模式的后面我也会讲到调试时怎么用这个能力。2.5 和 LangChain 生态的关系不是替代是升级LangGraph 在底层复用了很多 LangChain 的概念和组件工具接口、消息抽象、模型封装等。它的节点函数里完全可以自由使用 LangChain 的链、模型调用、工具装饰器。两者更像是同一个生态在两个不同复杂度层次上的呈现。简单说LangChain 提供了丰富的积木LangGraph 则提供了把这些积木灵活组合成“图”的框架。我的建议是不要把它们割裂开看在实际项目中经常是混着用的。3. 手把手构建一个可复用的检索问答 Agent3.1 场景设定与整体架构这部分我们用一个小而完整的案例来走一遍流程。场景是这样的构建一个能够围绕内部知识库做问答的 Agent它具备两个行为第一直接基于已有知识回答第二如果判断知识不足或需要联网补充就调用检索工具再基于检索结果回答。这个 Agent 需要支持多轮对话并能记录当前用户问题的历史上下文。整体图结构设计如下入口节点检查是否需要检索条件分支需要则进入检索节点不需要则进入生成节点检索节点调用向量检索工具并把结果写入状态生成节点整合历史消息和检索结果调用大模型生成答案结束节点输出最终答案这个结构很简单却是 LangGraph 条件路由和状态管理的最小完整案例后续无论多复杂的 Agent 都是在此基础上的扩展。3.2 定义状态与检索工具先定义状态和工具函数。我用的是伪代码风格重点在于逻辑结构清晰你可以直接替换成自己环境里的实现。from typing import TypedDict, Annotated, List import operator class ChatState(TypedDict): question: str chat_history: Annotated[List[dict], operator.add] retrieved_docs: List[str] needs_search: bool answer: str def check_search_node(state: ChatState) - dict: last_message state[chat_history][-1] if state[chat_history] else needs_search should_search(state[question], last_message) return {needs_search: needs_search} def retrieve_node(state: ChatState) - dict: docs knowledge_retriever(state[question]) return {retrieved_docs: docs} def generate_node(state: ChatState) - dict: context \n\n.join(state.get(retrieved_docs, [])) if not context: reply normal_llm_chat(state[question]) else: reply rag_llm_chat(state[question], context) return {answer: reply}这里should_search可以是一个简单规则比如检测到“最新”“实时”“内部文档没提到”等信号也可以用一个小模型做意图判断。在实际生产项目中我倾向于用后者准确率高很多。3.3 编排条件路由循环接下来把节点组装成图并加上条件路由。这里有一个易踩的坑LangGraph 的条件边是我们自己写判断逻辑判断逻辑不要在多个节点里各处都有尽量集中在条件函数里否则状态流会变成一团乱麻。from langgraph.graph import StateGraph, END def decide_next(state: ChatState) - str: if not state[needs_search]: return generate_node return retrieve_node graph StateGraph(ChatState) graph.add_node(check_search_node, check_search_node) graph.add_node(retrieve_node, retrieve_node) graph.add_node(generate_node, generate_node) graph.set_entry_point(check_search_node) graph.add_conditional_edges( check_search_node, decide_next, {generate_node: generate_node, retrieve_node: retrieve_node} ) graph.add_edge(retrieve_node, generate_node) graph.add_edge(generate_node, END) app graph.compile()这里有一个细节条件边的映射字典里键是decide_next返回的字符串值才是真正的节点名。如果你希望返回值和节点名保持一致可以直接这样写。我自己的习惯是让返回值和节点名解耦这样以后改节点名时不用动判断函数安全性高一些。3.4 编译运行与参数调整运行这段代码result app.invoke({ question: 我们的退款政策最近有什么变化, chat_history: [], }) print(result[answer])如果你开启了检索模式最终答案会综合知识库内容和问题来生成。这里有一个参数上的取舍需要说明retrieved_docs的条数、chat_history保留的长度都会直接影响上下文窗口占用和模型成本。我在实际项目中会把检索出的文档数限制在 4 到 6 条历史消息限制在最近 10 轮并且按 token 截断。控制上下文长度不是可选项是生产级应用必须考虑的。另外一个小技巧invoke 支持传入config参数里面可以带recursion_limit控制图最多执行多少步。这个参数在调试无限循环时特别好用比如一场意外的循环卡死如果不设上限请求会挂很久。默认上限足够但我在复杂图结构里都会主动设一个更小的值比如 20让异常快速暴露。4. 持久化、Checkpoint 与断点续跑4.1 记忆从哪来Checkpoint 机制解读单纯的图执行是一次性的invoke 结束所有状态都丢了。但实际业务里用户不可能每次都把所有历史重新说一遍。LangGraph 提供的答案是 Checkpoint。Checkpoint 机制简单说就是图上每执行到一个节点框架就把当前完整 State 保存成一份快照。这份快照存在内存或数据库里关联一个线程 ID。下次要恢复这个会话只要传入同一个线程 ID图就能从上次中断处继续跑。这会带来两个很明显的好处。第一是多轮对话的天然记忆上一轮的状态在下一轮自动可用。第二是容错如果某一步执行失败或用户主动终止了流程后续可以从断点恢复不用重头开始。这在长任务场景里价值巨大。4.2 多轮会话的 thread_id 用法使用 Checkpoint 并不需要大幅改代码主要是在编译和调用时多传一个配置config {configurable: {thread_id: user_123_session_456}} result1 app.invoke({question: 你们的退货周期是几天}, configconfig) result2 app.invoke({question: 如果超过这个时间怎么办}, configconfig)第二次调用时输入的 State 会自动带上第一轮产生的历史消息。你会发现chat_history字段里自动积累了前面的内容而你的生成节点只需要正常读取即可。这个机制解决了我之前手动维护历史记录的大量样板代码。使用 thread_id 有个原则需要想清楚thread_id 是线程的标识不是用户 ID。一个用户可以有多个线程比如围绕不同问题开的会话。把用户 ID 和会话 ID 拼起来作为 thread_id 是比较常见的做法。如果只拿用户 ID 当 thread_id会导致所有历史混在一个会话里后续状态会越来越乱。除了内存级 Checkpoint 外LangGraph 也支持把状态存到外部持久化存储里。这样即使服务重启历史也能恢复。我建议生产环境一定要配置外部持久化否则服务一挂用户的历史对话全没了体验会很糟糕。4.3 Human-in-the-loop让人插一脚的设计Checkpoint 机制还解锁了一个冷门但重要的能力人机协同。LangGraph 允许你在图中设置断点比如在执行某个敏感动作之前暂停等人工确认后再继续。这非常适合审核、审批、危险操作确认等场景。实现思路大概是在编译时把某个节点设置成断点节点运行时图执行到该节点前会自动暂停并返回当前待执行步骤的信息。你可以自行做人工审核流程然后再调接口让图继续跑。我在实际项目里用这个模式实现过一个生成内容的发布前审核流程大模型生成初稿进入审核节点时暂停运营人员在后台查看、修改或退回确认后图继续执行发布动作。整个过程用户无感知但对业务方来说风险控制能力增强了一个量级。5. 常见问题与排查技巧实录5.1 三个高频踩坑场景场景一状态字段被覆盖而不是累积这是最容易踩的坑。如果某个字段需要由多个节点往里追加内容比如历史消息列表一定要在定义 State 时给它加上Annotated和 reducer。否则每个节点写入时都会整体覆盖最终只保留最后一个节点的内容。排查这种问题可以把最终 State 打印出来看看字段值是预期的累积列表还是只有单条内容。场景二条件路由返回了不存在的节点名条件边函数返回的内容如果拼错了一个字母运行时会直接报错 “Node not found”。这种问题在字符串没有常量管理时非常容易发生而且错误信息只有在实际执行到该条件边时才出现。我强烈建议把所有节点名字定义成常量并且条件函数返回时直接用常量不要手写字符串字面量。场景三循环节点停不下来某些业务流程里为了处理复杂任务会让图在“检索—生成—工具调用—再检索”之间循环。如果循环终止条件写得太宽松或依赖的模型判断不稳定就可能出现死循环。LangGraph 有 recursion_limit但默认值很高不会及时触发。排查时先把 recursion_limit 调小一些比如 20让循环快速中断再检查终止条件。另一个思路是在状态里加一个计数器字段超过阈值强制跳转到最终节点这是工程上比较稳的兜底方案。5.2 调试建议与日志技巧调试 LangGraph 应用有一个明显优势图结构本身就是很好的调试地图。出错时可以先确认是哪两个节点之间的转换出了问题再针对节点单独测试。我在项目里通常会在每个节点的入口打一条结构化日志记录当前节点名和关键状态字段的摘要在节点出口再打一条日志记录返回的更新内容。这样整个执行路径一目了然。不要只记录节点名还要记录状态里的关键值否则日志只能告诉你“走到哪了”不能告诉你“为什么走了这条路”。LangGraph 还提供了app.stream()方法可以逐节点流式输出执行过程。调试阶段我会经常用它替代 invoke能直观看到每一步的状态变化for event in app.stream({question: ...}, configconfig): for node_name, state in event.items(): print(f节点: {node_name}) print(f状态摘要: {state})流式执行在线上也能派上用场用户侧可以实时看到 Agent 当前在做什么体验上会“智能感”强很多。但要注意不要直接把完整状态流给用户里面可能包含检索中间结果、内部判断逻辑容易暴露实现细节。5.3 问题速查表现象可能原因排查思路历史消息不累积State 字段没配 reducer检查 TypedDict 中的 Annotated 定义条件路由报节点不存在返回字符串与节点名不一致用常量管理节点名图运行超时循环终止条件过弱调小 recursion_limit加计数器兜底多轮对话串号thread_id 设计粒度不对检查是否把用户 ID 直接当 thread_id某节点状态为 None 引发异常前置节点未返回该字段在节点入口加字段默认值修改 State 后下游读不到新值返回的字段名与状态字段不同检查节点返回的 dict 键名这张速查表基本覆盖了早期使用 LangGraph 的高频问题。很多问题形式上五花八门本质都是对 State 更新语义或节点命名约定理解不到位。6. 工具选型解析与进阶方向6.1 什么场景必须上 LangGraph不是所有项目都需要 LangGraph。判断标准主要看三点有没有分支决策、有没有循环依赖、需不需要持久化和断点恢复。如果你的应用只是“接一个模型给一个答案”LangChain 或其他轻量封装完全够用。一旦你的 Agent 需要根据中间结果决定下一步、需要调用多个工具且调用的顺序不固定、需要在长时间任务中反复比较和筛选信息那么图结构的价值就体现出来了。从维护角度说图结构让流程可见。一份代码接手过来只要画出节点和边的连接关系就能理解业务逻辑。而 if-else 嵌套的链式代码这种可见性是很差的。团队协作时这个优势会越来越明显。6.2 与其他编排方案的对比市面上做 Agent 流程编排的选项不少大致有三类思路。一类是在原生 Python 代码里自己维护状态机和循环一类是用 LangGraph 这类专门的图编排库另一类是拖拽式流程编排平台。单说 LangGraph 和自维护代码的区别。自维护代码最大的问题是状态不集中、恢复逻辑要自己造轮子、异常处理分散。你可能自己写出了一个能用的流程但等到要加“断点恢复”“人工审批”“历史会话”就会发现每一块都要手搓。LangGraph 把这些从框架层面解决了代码量不一定会少但复杂度可控得多。与其他图编排库相比LangGraph 的优势是离 LangChain 生态近、对 AI 场景的抽象更贴切比如消息、工具、模型调用这些概念都是内置的。局限是它要求你按照它的思维模式去设计状态对没有状态机经验的开发者来说有一定的学习成本。6.3 可以继续扩展的方向LangGraph 本身的能力边界还在快速扩展。几个方向值得大家保持关注。一个是多智能体协作模式。单个 Agent 处理不了的任务可以拆成多个子 Agent 并行或接力完成。LangGraph 的状态机制很自然地支持这种组织结构每个子图可以作为一个节点挂在更大的图上。另一个是与外部事件流的集成。结合消息队列和事件触发器图节点可以由外部事件激活而不再局限于用户请求驱动。这个方向比较适合构建自动化运维、异常处理这类响应式系统。还有一个是图的动态生成。目前多数应用的图结构是静态的开发时写死了。如果能让图结构随着任务变化而动态生成通用性和灵活度会大幅提升。这也意味着 Agent 的自主规划能力会上一个新台阶当然工程达成难度也要高不少。最后再分享一个小技巧。如果你刚开始学 LangGraph不要一上来就追求“大而全”的复杂图先拿一个三四个节点的最小流程图跑通再逐步加入条件路由、持久化、人工介入这些机制。状态机的思维和传统面向过程编程很不一样需要一段时间适应。等你想清楚“状态是什么、变化如何发生、谁能改变状态”这三件事LangGraph 就不再是一个需要学的新框架而是一个顺手的表达工具。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
