如果你已经受够了那种“一问一答”的聊天机器人想在项目里做出一个真正会查数据、会判断、会回头修正答案的智能体Agent那你一定会遇到同一个问题流程到底怎么编排我在过去几个项目里反复试过链式调用、硬编码状态机、甚至自己写事件循环……直到把 LangGraph 用起来才感觉复杂智能体编排这件事终于有了顺手的工具。这篇内容不是 API 文档翻译而是我从第一个 Demo 到生产环境部署的全过程总结一步步带你从零看懂 LangGraph 的核心机制适合所有想把 Agent 落到真实业务里的开发者。我尽量把“为什么这么设计”“当时怎么踩的坑”也一起讲清楚你照着走比我当时少走很多弯路。1. 先聊聊为什么“链式” Agent 撑不住真实业务这一步想不通后面都白搭1.1 一次“流程写不出来”的翻车经历我最早做客服助手时需求特别简单用户问“我的快递到哪了”系统先去订单接口查状态再把结果翻译成友好回复。第一版用链式写法代码只有三步感觉毫无压力。可产品经理紧接着提了一个需求当系统识别出用户在问订单、但没有给出订单号时需要追问“请提供订单号”拿到后再继续查。问题来了。链式的本质是“上一步的输出给下一步当输入”它不会停在半路等你更不会查不到之后换个思路重来。我在代码里塞了一堆 flag 和全局变量勉强跑通了可一上多轮对话、并发用户状态就串了。这个经历让我意识到Agent 程序真正需要的不是一条笔直的管道而是一张可以“绕路”、可以“停下来等一等”、甚至可以“退回上一个路口”的图。1.2 普通调用链与“有状态控制流”的本质差异普通调用链是用户输入 → 模型生成回复 → 结束。最多加一层检索或工具调用顺序固定不变像一个单向传送带。真实 Agent 是模型先判断意图 → 决定要不要调用某个工具 → 拿到工具结果 → 基于结果决定是回答还是继续查 → 如果回答不满意可能还要重新表述或重新检索。这里每一步都依赖“上一步结果 当前整体状态”而且允许循环、分支、回退。在没有明确状态机的前提下这种逻辑靠 if-else 硬写写不到第五个分支就会乱。我们经常说 Agent LLM 工具 记忆 编排。前三个都好解决真正难的是编排因为编排决定了一个 Agent 在边界条件下是否还能稳定输出。而 LangGraph 这个框架恰恰就是冲着“编排”来的。1.3 LangGraph 到底改变了什么编排层的出现LangGraph 的思路非常朴素把复杂的 Agent 流程画成一张有向图节点是一个个函数每条边决定一个函数执行结束后下一步去哪里。与 LangChain 里 AgentExecutor 这种“黑盒执行器”相比显式图带来几个实打实的好处流程可观测每一步执行完都能打印状态出问题一眼定位。控制流可编程条件、循环、回退、人工审核都是图的边和节点不再靠补丁。状态可持久化把当前“台账”存下来进程重启后还能接着跑。团队可复用同一套图结构可以跨项目复制只改节点内部逻辑。想明白这些再去看 LangGraph 的 API 就会觉得顺理成章。如果你现在还不太理解 LangChain 和 LangGraph 的关系也完全可以LangGraph 不依赖 LangChain 那些 Chain 抽象它可以当作一个独立的状态图执行引擎来用。2. StateGraph 的三个核心概念先记住“节点、状态、边”就赢了一半2.1 节点就是一个接收 State 并返回补丁的普通 Python 函数在 LangGraph 里节点不需要继承任何基类也不需要装饰器。一个节点函数长这样def some_node(state: MyState) - dict: result do_something(state[field]) return {field: result}它接收当前状态返回一个字典。这个字典不会整个替换状态而是作为“补丁”合入。我真没有夸张——只要理解了这句话后面所有节点逻辑都通了。因为节点就是普通函数你可以塞任意逻辑调用模型、查数据库、调外部 API、跑一段算法。这也意味着你熟悉的 Python 工具链全都能直接搬进来不需要学什么“节点语法”之类花里胡哨的东西。2.2 状态一张贯穿全图的“总台账”State 是 LangGraph 的命根子。它本质是一个 TypedDict 或 dataclass用来规定图里有哪些数据、什么类型。我习惯把它想象成流水线上的一张随工单每个工位都会读它、在它上面填内容填完之后传给下一个工位。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 对话历史 order_id: str # 当前处理的订单号 retry_count: int # 已经重试了几次为什么messages要包一层Annotated这是 LangGraph 的 reducer 机制后面第 4 节单独讲。你先把它理解成“这个字段有自己的合并规则”而不是简简单单覆盖掉。2.3 边顺序边是“默认剧情”条件边是“剧情分叉”边决定了执行方向。最基础的是add_edge(node_a, node_b)表示 A 执行完必然去 B。而add_conditional_edges则是在一个节点后面挂一个“路由函数”函数根据 state 返回下一个节点名。打个比方顺序边像地铁的固定线路一站一站走条件边像岔路口的路标根据当前情况决定往左还是往右。状态、节点、边这三个概念组合起来理论上能表达任何你能画出来的业务流程图——循环就是一条边从后面指回前面这也就是 LangGraph 敢叫“复杂智能体编排秘钥”的原因。2.4 为什么图模式比 AgentExecutor 可靠用过 AgentExecutor 的人可能都有一种感觉它能跑但内部到底怎么走的像个黑盒。而图模式把所有控制流都摊开放在你面前循环不会莫名其妙出现也不会偷偷消失。另一个好处是测试。函数节点可以独立单测你不需要启动整个 Agent 才能验证一步逻辑。把复杂的编排拆成图每一个节点的输入输出边界都清晰这对团队协作的价值极大。3. 从零跑通第一个图安装、写节点、编图、执行3.1 环境准备不需要装一堆东西我的建议是装两个基础包就够了pip install langgraph langchain-coreLangGraph 负责图执行langchain-core 提供消息类型、工具注解这些基础类型。等到你真的要接某个语言模型再去装对应的模型封装包。有人说是不是还要装整个 LangChain其实不需要LangGraph 对 LangChain 的依赖比你想象中轻。Python 版本建议 3.9 以上我自己用的是 3.11 环境体验最稳。3.2 第一个 Demo一个会自我修正重写的迷你图我写一个特别小的图用来演示“节点 条件边 循环”三者怎么配合。这个图会模拟一个流程系统生成一句话 → 检查这句话是否太长 → 太长就重写 → 重写后再检查直到长度合格。from typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): question: str # 当前问题 complexity: int # 用长度模拟复杂度 def generate(state: DemoState) - dict: # 假设这里是一轮模型生成生成后的文本长度超过20就认为“过于复杂需要重写” return {question: langgraph is a framework for building stateful agent workflows} def check(state: DemoState) - dict: # 用长度代替真正的复杂度评估 return {complexity: len(state[question])} def rewrite(state: DemoState) - dict: # 截断到20个字符模拟一次“重新组织语言” return {question: state[question][:20]} builder StateGraph(DemoState) builder.add_node(generate, generate) builder.add_node(check, check) builder.add_node(rewrite, rewrite) builder.add_edge(START, generate) builder.add_edge(generate, check) def route(state: DemoState) - str: if state[complexity] 20: return need_rewrite return finish builder.add_conditional_edges(check, route, { need_rewrite: rewrite, finish: END, }) builder.add_edge(rewrite, check) # 重写之后回到check再来一轮 graph builder.compile() result graph.invoke({question: }) print(result)这段代码里有个值得注意的细节route函数返回的不是节点本身而是一个分支别名真正的对应关系在add_conditional_edges的第三个参数里声明。这样你把判断逻辑和跳转目标解耦了后面改流程时会非常爽。3.3 执行结果与逐步观察跑完result的question会被截断成前 20 个字符。为什么因为 generate 出来后长度超过 20check 判定为需要重写rewrite 截断再回到 check此时长度等于 20不满足大于 20循环才结束。整个过程虽然简单但已经把“判断条件 回环修正”这个 Agent 核心骨架练了一遍。如果你和我一样有“黑盒恐惧”用 stream 看每一步for step in graph.stream({question: }, {recursion_limit: 10}): print(step)控制台会按执行顺序打印每个节点返回的补丁你就能清清楚楚看到 generate → check → rewrite → check → END 的完整轨迹。这个观察能力在复杂 Agent 里能救命调试的时候一眼就能看出是哪一步没有按预期走。我建议你把那个 20 改成 5再把rewrite的截断长度也改小观察循环次数的变化。亲手调一次参数比看十遍文档都有用。4. 状态合并与 Reducer为什么“各写各的”还能不打架4.1 节点返回的是补丁不是全量替换这个理解太关键了我必须单独开一节。很多第一次用 LangGraph 的人以为节点返回{field: x}后state 里就只剩field一个字段了。实际上 LangGraph 做的是浅合并返回的字典里每个 key 对应更新到 state 的对应 key没提到的 key 保持原样。举一个实际例子节点 A 更新了order_id节点 B 只返回{answer: 已发货}。执行完 B 后order_id还在并不会丢。这个设计让每个节点可以只关注自己负责的字段协作起来不会互相覆盖。你完全不用像以前那样手动维护一个全局大字典。4.2 消息列表面临的问题多个节点写同一个字段如果 state 里有messages这个列表字段麻烦来了。节点 A 加了一条 AI 消息节点 B 加了一条工具消息它们都返回{messages: [xxx]}。按照普通浅合并逻辑后执行的会整体覆盖messages前一个节点的工作就白做了。所以 LangGraph 引入了 reducer。Annotated[list, add_messages]的意思就是当某个节点往messages字段写入新值时不要覆盖而是调用add_messages这个合并函数把新旧列表拼起来。你可以把它理解为操作系统的追加写模式同一份文件多个进程都能安全地在末尾追加内容。4.3 自定义 Reducer不止节省代码还防坑不一定非得用现成的add_messages你完全可以自己写合并规则from typing import Annotated def merge_scores(left: list, right: list) - list: # 两边都保留但通过去重保证顺序 seen set(left) for item in right: if item not in seen: left.append(item) return left class State(TypedDict): scores: Annotated[list, merge_scores]或者更偷懒的方式直接复用操作符。import operator class State(TypedDict): steps: Annotated[list, operator.add]我的一点经验凡是会被多个节点更新的列表字段一律加 reducer哪怕当前只有一个节点写它。因为需求增长速度永远比你重构速度快等第三个节点开始往里写消息时你已经忘了最初为什么不加。到时候排查“消息怎么只剩最后一条了”会花掉你整整一个下午。5. 条件边与循环是智能体的灵魂从“一条道走到黑”到“会拐弯会回头”5.1 模型其实不会调工具它只会“表达意图”很多新手对“Agent 调用工具”有误解以为模型直接执行代码。实际上LLM 只是根据指令输出一个结构化的调用意图里面包含工具名和参数。真正执行工具的是你写好的函数LangGraph 的 ToolNode 帮你把这件事接到图里。理解这一点后图里最关键的一个分支就出现了最后一条 AI 消息里有 tool_calls → 走工具节点没有 → 直接生成最终回答。这个分支就是条件边的经典范例几乎所有工具型 Agent 都长这样。5.2 一个标准的“判断→执行→回来”循环结构我用文字描述一下这个图的路径你可以照着搭入口 START 先进 agent 节点agent 节点把当前消息列表交给模型模型返回一条可能带 tool_calls 的 AI 消息。接着走条件边如果有工具调用意图去 tools 节点tools 节点执行真正的工具函数把执行结果作为 ToolMessage 追加到 messages然后边指回 agent。模型看到工具结果后再做一轮决策。如果没有工具意图条件边直接连到 END。自定义路由函数长这样def should_continue(state): last_message state[messages][-1] if last_message.tool_calls: return use_tool return respond然后注册边builder.add_conditional_edges(agent, should_continue, { use_tool: tools, respond: END, }) builder.add_edge(tools, agent)这套结构其实就是 ReAct 模式只是由 LangGraph 替你把控制流全部落地了。你可以在这个基础上随意加“人工审批”“超时重试”“查询失败转人工”等各种分支。5.3 循环必须设上限recursion_limit 是保命底线图允许节点反复执行但实际生产里最怕“模型陷入工具调用循环”。模型一直认为“我需要再查一次”永远不返回最终答案图就会无限跑。LangGraph 默认的递归上限一般够用超过会抛异常但你最好主动设一个业务合理的值。result graph.invoke(input_state, config{recursion_limit: 15})如果业务上确实需要很多步就把上限调大但每次循环里最好带一个“计数节点”在 State 里加一个step_count字段每次回环增加 1超过阈值就强制收尾。这种兜底逻辑我是强烈建议加的等模型抽风时你就知道它多重要。5.4 无限循环的场景真出现过吗出现过而且不止一次。有一次我用一个弱一点的开源模型做 Agent它在“检查答案是否满意”这个节点上频繁返回“需要重写”一直循环到上限。加了计数节点后第三次重写还不行就强制结束把当前答案返回给用户并附带一句“生成过程不太稳定请核对关键信息”。这比无限循环体面多了。6. 持久化与多轮记忆一条 config 让 Agent 变成“有记忆的人”6.1 每次 invoke 都是“失忆”的这问题得靠 Checkpointer 解决直接用compile()出来的图每次 invoke 都是独立执行。你第一轮告诉它“我叫小明”第二轮再问“我叫什么”它根本不知道。原因是 LangGraph 默认没有保存每条路径的执行状态。要解决必须给图挂一个 Checkpointer也就是“检查点存储器”。你可以理解成游戏存档每跑完一步把整个 state 存一份下次继续时从最新存档读进度而不是从零开局。6.2 三种常见 Checkpointer 怎么选类型持久范围适合场景我的建议MemorySaver仅当前进程本地 Demo、单元测试上手学概念用SqliteSaver本地单文件单机服务、小规模并发生产起步首选PostgresSaver数据库多实例、需要跨节点恢复团队线上服务MemorySaver 用起来最简单from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)6.3 thread_id不同用户的“账本”怎么隔离挂上 Checkpointer 后invoke 的时候要传一个 config告诉 LangGraph“这次对话属于哪本账”config {configurable: {thread_id: session-001}} graph.invoke({messages: [(user, 我叫小明)]}, config) graph.invoke({messages: [(user, 我叫什么名字)]}, config)第二次 invoke 时因为使用了同一个 thread_idLangGraph 会先把上次的 state 载入再把你这次的新消息追加进去模型就能想起来你叫小明。换一个 thread_id就是完全独立的会话互不干扰。这就是多轮记忆的最小实现。一定要记住thread_id 不要用手机号、身份证号这类敏感信息用主键 ID、UUID 都好。调试线上问题时能精确定位会话又不泄露隐私才是合格的会话键。如果你直接把用户手机号塞进去日志一打出来全是隐私。6.4 我的一个误用教训我最早做多轮记忆时以为只要挂上 MemorySaver所有历史就自动保留。结果重启服务后全忘了。后来才意识到 MemorySaver 存在进程内存里一重启就清零。想跨进程、跨重启保留就得换 SqliteSaver 或 PostgresSaver。Demo 阶段无所谓生产环境一定要提前想好持久层。7. 实战拆解把一个能查订单的客服智能体编成一张图7.1 业务需求怎么翻译成图我拿一个真实项目里最常见的需求来讲用户问订单状态系统要识别意图、必要时调订单接口、最后给出友好回复。画出来的图特别简单就三个角色节点职责输入依赖输出agent让模型决定要不要调工具当前 messagesAI 消息可能带 tool_callstools执行实际订单查询消息中的 tool_callsToolMessage条件边判断去 tools 还是 END最后一条 AI 消息路由目标业务上真正体现编排价值的是这个循环tools 跑完结果会回到 agent模型看到真实查询结果后再生成自然语言回复而不是在 tools 节点里硬拼接话术。原始数据是什么样、用户看到的回复是什么样两者可以完全分离。7.2 完整代码骨架from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition from langchain_core.tools import tool class AgentState(TypedDict): messages: Annotated[list, add_messages] tool def query_order_status(order_id: str) - str: 根据订单号查询订单当前状态返回可读的状态描述。 # 实际项目里替换成你的订单接口 return f订单 {order_id} 已于昨天发货当前状态运输中。 # 这里的 llm 来自你自己的模型封装只要能 bind_tools 即可 llm get_llm() llm_with_tools llm.bind_tools([query_order_status]) tools_node ToolNode([query_order_status]) def agent_node(state: AgentState): ai_message llm_with_tools.invoke(state[messages]) return {messages: [ai_message]} builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, tools_node) builder.add_edge(START, agent) builder.add_conditional_edges(agent, tools_condition, { tools: tools, __end__: END, }) builder.add_edge(tools, agent) graph builder.compile()tools_condition是 LangGraph 预置的条件边函数逻辑就是“最后一条 AI 消息有 tool_calls 就走 tools没有就结束”。它省的不只是几行代码还能避免你手写路由函数时漏掉一些边界情况。7.3 调用与结果验证result graph.invoke({messages: [(user, 你好帮我查一下订单 1024 的状态)]}) print(result[messages][-1].content)正常会得到类似“订单 1024 已于昨天发货当前状态运输中”的回答。如果用户问的是“今天天气怎么样”模型没有产生 tool_calls条件边会直接放行到 END最后模型凭自身能力回答。7.4 这一节踩过的坑第一版跑的时候我踩了两个坑都是典型的“看起来能跑但一上真实数据就翻车”。第一个工具函数没写 docstring模型完全不知道该在什么场景下调用它导致它宁可自己瞎编也不调工具。后来把 docstring 写清楚模型才老实调用。工具的函数名、参数名、描述都是模型决定是否调用的关键信息不要偷懒。第二个模型偶发返回一个空列表的 tool_calls我的自定义路由直接判断if ai_message.tool_calls空列表时反而误入了 END 分支用户收到“抱歉我暂时查不到”这种话。后来统一用tools_condition这类边界它内部已经处理干净了。8. 进阶编排技巧人工介入、状态检查和生产化排错8.1 审批类 Agentinterrupt_before 让流程“半路刹车”不是所有 Agent 都应该全自动。比如“自动回复售后工单”这个需求技术上能全自动但业务上要求“人类审核后回复”。LangGraph 提供了中断钩子graph builder.compile(checkpointercheckpointer, interrupt_before[human_review])执行到human_review节点之前invoke 会中断返回图不再往下走。你可以在自己的业务系统里弹一个人工审核框等审核人点了同意再用同一个 config 调用一次graph.invoke(None)图会从断开位置继续。半自动流程不需要手写任何“等待状态”的 hack这就是有状态编排的威力。我也用过 interrupt_after比如在模型生成回复后先展示给人工修改改完再继续发送。两种钩子覆盖了“执行前审核”和“执行后编辑”两类常见需求。8.2 get_state查图走到哪一步了中断和异常恢复都离不开状态检查。graph.get_state(config)返回一个快照对象其中next字段表示下一步即将执行的节点values里是当前 state 内容。调试多轮对话时我特别喜欢用它先看最后几条消息再看 next十秒钟就能定位问题是状态错了还是路由错了。如果想查看历史存档用graph.get_state_history(config)它按时间倒序返回。万一某次执行结果不满意你可以拿旧 checkpoint 继续跑相当于给 Agent 加了“时光机”。这在人工审核场景特别有用审核人觉得这条回复不太好可以直接回到上一步重新生成。8.3 invoke / stream / astream 怎么选单次要最终结果用 invoke。想给用户展示“正在打字”的中间过程或想逐步观察节点输出用 stream。异步接口是 astream 和 ainvokeWeb 服务里配合 async 事件循环更合适。我自己的习惯是写单元测试全用 invoke联调阶段改 stream 看中间消息上生产再根据前端交互需求决定。如果用户能接受几秒的等待invoke 最简单如果希望边生成边显示stream 是唯一解。8.4 我给生产项目的几条硬建议现在回到开头那个问题为什么大家说 LangGraph 是复杂智能体编排的秘钥因为它把“流程”变成了看得见、可调试、能持久化的图。最后分享几条我踩过坑后沉淀下来的生产建议第一图里的节点尽量保持幂等。因为有检查点恢复机制同一个节点可能因为恢复被再次执行节点内部不要依赖“只跑一次”的副作用。第二一个图的节点数超过 15 个就考虑拆子图或者重新设计。图太大会让条件边变成一张蜘蛛网别人接手时根本不敢改。第三日志里务必带上 thread_id 和 checkpoint_id。线上多用户并发时没有会话标识的日志等于没有日志你根本不知道是哪次执行出了问题。第四工具调用失败不要静默吞掉单独做一个“重试或转人工”的分支节点比在工具函数里 try-except 到底更具可观测性。用户看到一句“稍等我正在重新查询”比突然断线体面得多。最后再补一句可能反常识的体会LangGraph 的上手曲线其实不在 API而在思维转换。你不需要先学完所有概念再动手直接拿一个最简单的图跑起来然后不断加字段、加节点、加条件边遇到状态不对再看 reducer你会发现自己很快就能把脑子里那张业务流程图翻译成一张真的图。我就是这么从第一个 Demo 走到生产环境的希望这篇保姆级教程也能让你少走几圈弯路。