做 AI Agent 项目做得多了你一定会撞上一个非常现实的问题单轮问答做得再漂亮一旦要求它处理多步操作、来回确认、多轮记忆代码结构就容易失控。我今年接的不少需求都卡在这一环最后促使我把工作流编排从原来的 Chain 方式换成了 LangGraph才算真正弄明白“有状态的 AI Agent 工作流”应该怎么设计。LangGraph 本质上是一个基于图结构的编排框架核心思路是把 Agent 的每个处理步骤建模成节点和边把对话历史、中间结果、业务标记全部显式放进 State 里从而实现可控、可回放、可恢复的多步工作流。这篇内容适合已经会调大模型接口、但一遇到多分支任务就头疼的开发者不管你是做客服机器人、代码助手还是内部数据问答下面这套从入门到实战的设计思路都能直接用上。1. 为什么需要一张“图”来编排 Agent1.1 从 Chain 到 Graph编排方式的演变早期我写智能客服流程很简单用户消息进来拼一个 Prompt调一次模型返回答案完事。这种线性 Chain 在处理“读问题、回答”这类固定顺序任务时完全够用但业务一旦复杂问题就暴露了。比如客服系统需要先判断用户属于技术问题、账单问题还是售后问题再走对应处理通道代码助手需要先分析项目状态再决定要不要调用检索工具。这些流程天然不是一条直线你用 Chain 硬写也行只是需要塞进一堆 if else写到最后就变成一棵庞大的决策树嵌套在业务代码里。后续想改一个节点所有调用点都要跟着动测试难度也直线上升。LangGraph 的处理方式完全不同。它把工作流定义成一张有向图入口是 StateGraph。节点负责执行具体任务比如“判断分类”“生成回答”“调用外部服务”边负责决定执行顺序普通边直接连接上下游条件边根据当前状态动态选择下一个节点。运行时图引擎会把状态一步一步往下传节点返回更新结果后自动合并进全局状态再继续下一跳。我也不是建议所有项目都抛弃 Chain。如果一个任务只有三步且顺序固定用 Chain 反而更直接。真正值得迁移到图上的是你发现自己需要“根据上一步的输出决定下一步”或者要“反复执行直到某个条件满足”的时候。Graph 解决的是会转弯、会循环、会分叉的复杂任务不要杀鸡用牛刀。1.2 有状态 Agent 的四个真实痛点既然标题重点在“有状态”我先分享实际过程中最常碰到的四类状况后面的代码也都围绕这四类展开。上下文覆盖。多个节点都想往同一个字段里写数据后面的把前面的覆盖掉导致前面步骤的中间结果丢失。分支不可控。模型第一步分类的结果要影响后面走向但业务代码里全是散落的 if else整体流程没法一眼看清。流程无法回放。工作流跑到第三步挂了你想知道前两步到底产生了什么状态完全没有记录。多轮对话失忆。用户上一轮报了订单号这一轮接着问进度Agent 完全不记得只能靠你手动把完整历史塞进 Prompt。LangGraph 解决这些问题的办法是把“状态”当一等公民显式定义、逐步更新、依赖检查点持久化。你可以把 State 理解成一张流水线上的工单节点每次拿到工单修改几个字段再交回传送带传送带有一套合并规则不会因为多个节点同时改同一个字段就把数据丢掉。这样构建出来的 Agent行为可预测出了问题也能把现场状态调出来复盘。在真实项目里我见过最多失败案例并不是模型回答不好而是编排层把状态搞丢了。模型能力上升很快但每一步执行后上个环节判断出来的字段存在哪里、下一个环节该读什么很多团队没有想清楚。LangGraph 强制先声明状态字段等于逼着你在动手前先把业务数据模型设计好这在多人协作的场景里尤其有价值。节点之间有了明确的传参接口不会再出现“我改了 A 节点的返回B 节点就神秘报错”的问题。2. LangGraph 核心概念状态、节点和边2.1 状态State流水线上的工单在 LangGraph 里State 通常使用 Python 的 TypedDict 定义每个字段表示工作流中需要维护的一份数据。一个简化的示例如下from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] category: str resolved: bool这里声明了三类数据messages 存放对话消息历史category 表示工单分类结果resolved 表示任务是否完结。关键在Annotated[list, add_messages]这是消息字段的 reducer 声明告诉图引擎当多个节点都返回新消息时不要覆盖整份列表而是把新旧内容合并起来。缺了它后执行的节点会把之前所有历史消息清空多轮对话立刻失忆。普通字段比如 category 和 resolved则遵循“后写覆盖”的默认规则。也就是说如果先后有两个节点都返回 category最终保留后一个节点的值。这提醒我们一个设计要点多个并行节点最好不要同时修改同一个普通字段否则运行结果具有不确定性。我实际做项目时会把状态字段分成三类会话消息、业务字段、临时控制字段。会话消息统一用 add_messages 管理业务字段用普通覆盖但每个字段尽量只由一个节点写入临时控制字段比如“当前是否处于人工介入模式”用完后在合适节点清空或翻转避免污染下一轮对话。这个分类思路能规避大量状态竞争问题值得刻意练习。2.2 节点Node与边Edge零件和路线图节点就是一个普通 Python 函数签名固定接收一个 state 字典返回一个要更新的字段字典。它不需要返回 state 的全部字段只返回想更新的部分。比如一个分类节点可以写成def classify_node(state: AgentState) - dict: result model.invoke(...) return {category: billing}边则分两种。普通边直接把一个节点接到另一个节点graph.add_edge(classify, billing)表示分类做完后必定进入 billing。条件边则通过路由函数动态选择下一跳def route_after_classify(state: AgentState) - str: return state[category]这个路由函数返回的是字符串它必须能匹配某个已注册的节点名或者是 END。我第一次踩坑就在这里返回了tech_support却忘记add_node注册这个名称或者大小写不一致运行时直接报 KeyError。这种问题本身不复杂但在不熟悉框架时很容易被当作“模型输出错误”去排查白白浪费大量时间。2.3 图引擎从 START 到 END 的一次执行LangGraph 的执行过程是每次从 START 开始根据当前状态往下走。图引擎找到 START 连出去的第一个节点把它返回的结果合并进 State再找它的下一跳继续执行直到抵达 END。一次invoke就代表跑完一条完整路径状态的具体合并规则由字段的 reducer 决定。我把图引擎类比成快递分拣流水线每个节点是一个分拣工只负责自己关心的一两栏工单信息改完放回传送带传送带本身不关心工单长什么样只负责按路线图把它送到下一个工位。如果某个环节出错你只需要回看它经过每个工位时的状态快照问题大概率一眼定位。这也是 Graph 编排方式最大的工程优势流程可追溯、可暂停、可恢复。3. 从零搭建第一个有状态的 Agent 工作流下面用一个常见场景做完整示例某工单分诊助手。用户提交一个问题描述系统先判断类别再分流到不同处理节点最后生成工单完结标记。这个例子规模不大但覆盖了状态定义、节点注册、条件边和完整调用流程。3.1 环境准备与依赖安装使用 Python 3.10 以上版本安装pip install langgraph langchain-openai模型调用方面我建议选一个 OpenAI 兼容接口的服务商只要接口风格一致就能通过 base_url 指过去。示例代码如下from langchain_openai import ChatOpenAI model ChatOpenAI( model你的模型名称, base_url你的模型服务地址, api_key你的密钥, temperature0, )如果你本机有本地模型资源也可以换成本地部署方案LangGraph 不关心底层模型是哪一家它只认 state 里返回了什么字段。练习阶段最重要的原则是别纠结选型先跑通流程。另外我特别提醒API 密钥务必放到环境变量里不要写死在代码中。我们团队曾出过密钥被提交进仓库然后被自动扫描工具抓到的安全事故这种低级问题一旦发生代价远比你想象得大。3.2 定义状态和四个核心节点先定义状态结构from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] category: str resolved: bool然后定义节点函数。这里省略部分提示词细节重点看结构def classify_node(state: AgentState) - dict: user_text state[messages][-1].content prompt f 判断下面用户问题属于哪一类工单只输出一个词 tech_support / billing / after_sales 用户问题{user_text} resp model.invoke(prompt) return {category: resp.content.strip().lower()} def tech_support_node(state: AgentState) - dict: return {messages: [{role: assistant, content: 技术问题已登记请提供设备型号我会跟进排障。}]} def billing_node(state: AgentState) - dict: return {messages: [{role: assistant, content: 账单问题已定位系统将在24小时内下发调账记录。}]} def after_sales_node(state: AgentState) - dict: return {messages: [{role: assistant, content: 售后工单已创建专员会在1个工作日内联系你。}]} def finalize_node(state: AgentState) - dict: return {resolved: True, messages: [{role: assistant, content: 工单已处理完毕如需补充请直接回复。}]} def route_after_classify(state: AgentState) - str: return state[category]这里有一个小点需要注意classify_node 读取state[messages][-1]时依赖的是 add_messages 已经合并好的完整消息列表。所以你先要保证上游确实有一条 user 消息否则取最后一个元素可能拿到 assistant 消息或者直接越界。后面我会在排查清单中专门提到这类问题。3.3 组装图并完成一次调用组装图和调用流程如下graph StateGraph(AgentState) graph.add_node(classify, classify_node) graph.add_node(tech_support, tech_support_node) graph.add_node(billing, billing_node) graph.add_node(after_sales, after_sales_node) graph.add_node(finalize, finalize_node) graph.add_edge(START, classify) graph.add_conditional_edge( classify, route_after_classify, { tech_support: tech_support, billing: billing, after_sales: after_sales, }, ) graph.add_edge(tech_support, finalize) graph.add_edge(billing, finalize) graph.add_edge(after_sales, finalize) graph.add_edge(finalize, END) app graph.compile() result app.invoke({messages: [{role: user, content: 我收到了错误的扣款通知}]}) print(result[category]) print(result[messages])运行结果里category 会变成billingmessages 会按顺序包含用户的原始消息、billing 节点的回复、finalize 节点的回复。因为 add_messages 的存在每个节点返回的 assistant 消息都被追加而不是覆盖。对照传统 if else 写法这个图版本的优势很直观新增一个售后分支时只需要新增节点和边不需要改动其它分支的代码。每个节点也能独立测试传入构造好的 state断言返回字段即可。顺便说一下这段代码里我没往状态里放太多字段只有对话消息和两个业务标记。这是我在踩过几次“状态过度设计”的坑之后总结出来的原则State 里的字段越少出问题的概率越低。别把这里当成万能杂物箱能不放的字段就不放。4. 状态更新与持久化让 Agent 记住上一轮4.1 add_messages 是怎么合并消息的理解状态更新关键在于理解 reducer。默认情况下如果一个节点返回{category: billing}图引擎会用这个新值整体覆盖旧的 category。对于 messages 字段如果不用 reducer后面节点返回的新消息会把整个历史列表替换掉多轮上下文瞬间丢失。加了Annotated[list, add_messages]之后图引擎看到新消息会调用 add_messages 把新旧消息做合并并把相同 id 的消息去重后追加。实际开发里很多“为什么 messages 丢了一半”的疑问最后都归结到同一个原因某个节点没有返回完整消息列表或者某个状态字段漏写了 reducer。所以我的习惯是所有需要跨节点累积的数据字段从定义 State 那一刻就配上合适的 reducer而不是等出了问题再补。4.2 Checkpointer 与 thread_id多轮会话记忆的钥匙仅仅定义 State 只能保证单次 invoke 内部状态不丢。要让 Agent 跨轮记住上下文需要引入 Checkpointer。最简单的做法是用内存检查点from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: ticket-001}} app.invoke( {messages: [{role: user, content: 我收到了错误的扣款通知}]}, configconfig, ) app.invoke( {messages: [{role: user, content: 进度如何了}]}, configconfig, )第二次 invoke 时因为没有传完整历史只传了新的用户消息但 checkpointer 会从 thread_id 对应的存储里恢复上一次的完整状态。图引擎读到 messages 里已经有了上一轮对话就能正确理解“进度如何了”指的是刚才那张扣款工单。如果你每次调用都传一个新的 thread_id那么每次都是全新会话历史状态对当前调用不可见。很多团队排查“多轮对话不生效”时第一反应是模型配置出了问题实际上最常见的原因就是 thread_id 不一致或者压根没有 compile 时传入 checkpointer。4.3 检查点方案怎么选内存检查点适合开发和单机测试服务一重启状态就没了。生产环境按场景选择持久化方案我整理了一张选型表方案存储位置重启是否保留适用场景MemorySaver进程内存不保留本地开发、单元测试SqliteSaver本地文件保留单机单 worker 的小型服务PostgresSaver数据库保留多实例部署需要共享会话状态选择的核心逻辑是只要服务有多副本就不要用内存方案否则用户请求落到不同实例时各自看到的会话状态完全不同。数据库方案虽然多一次网络开销但状态一致性更好也方便做人工审计。4.4 持久化之后输入方式要跟着变加了 checkpointer 之后invoke 的输入就不再是“完整历史”而是“增量输入”。无状态模式下我们习惯把整段对话拼好再传进去有状态模式只需要传用户新增的那一条消息。我初期用 LangGraph 时犯过一个很蠢的错误手动把历史消息拼回去再 invoke导致同一条用户消息被 add_messages 重复追加上下文里出现大量冗余内容。后来 AI 回答质量很短时间下降排查半天才发现是这个问题。这个心智模型的转变很关键有状态工作流和普通 Prompt 工程完全不同你不再负责把记忆拼进上下文而是把记忆交给运行框架。你要做的只是确认 thread_id 在用户的整个会话生命周期内保持不变以及明确哪些字段应该跨轮保存、哪些字段只属于当前轮次。5. 条件路由、循环与并行工作流的三种转弯5.1 条件边让流程自己决定走向条件边在示例里已经出现过它返回一个节点名图引擎根据这个字符串决定下一步去哪个节点。除了直接返回 state 字段你也可以在路由函数里调用模型或查数据库再用结果做分支。需要注意两点路由函数必须是同步返回字符串的普通函数不要在它里面做耗时很久的 I/O 操作否则整个图执行会被卡住返回前记得做映射表校验能落到预先定义的节点集合里。在实际项目里我通常不会让路由函数直接返回模型原始输出因为模型大概率会返回Billing、billing。、billing\n这类带大小写和标点的内容。比较稳妥的做法是在分类节点里就对输出做清洗和映射让路由函数只消费标准化的字符串。5.2 循环在信息不足时反复追问有状态最有价值的一个体现是工作流可以停下来向用户要信息下一轮再接着走。给前面的分诊助手加一个“信息不足”分支def ask_user_node(state: AgentState) - dict: return {messages: [{role: assistant, content: 信息不够请补充你的设备型号或订单编号。}]} graph.add_node(ask_user, ask_user_node) def route_after_classify(state: AgentState) - str: category state.get(category, ) if category not in (tech_support, billing, after_sales): return need_info return category graph.add_conditional_edge( classify, route_after_classify, { tech_support: tech_support, billing: billing, after_sales: after_sales, need_info: ask_user, }, )当分类结果无法匹配任何已知分支时图会走到 ask_user 节点让模型回复一句“请补充信息”并结束本轮。下一轮用户补充信息后再调用checkpointer 恢复状态classify 重新执行这次 category 落到正确分支流程就继续往下走了。这就是一个典型的跨轮次循环。如果你希望在同一轮内部做循环比如反复调用检索工具直到结果满足条件那需要给 ask_user 或对应节点加一条条件边回到 classify。不过这类设计存在死循环风险我会加一个 loop_count 字段每循环一次加一达到上限后强制走向兜底节点。没有上限的循环生产环境里迟早会出事故。5.3 并行分支一次调用同时做多件事并行分支常用于“分诊完成后既要技术回复又要写工单日志”这类场景。实现上只需要给前驱节点添加多条出边再让多个分支汇聚到同一个终点graph.add_edge(classify, tech_support) graph.add_edge(classify, log_issue) graph.add_edge(tech_support, finalize) graph.add_edge(log_issue, finalize)这样一次 invoke 会同时启动 tech_support 和 log_issue 两个节点。这里最容易踩的坑是并行节点返回了相同普通字段的更新。两个节点同时尝试写同一个 status 字段时谁先完成谁后完成取决于调度最终值完全无法预测。正确做法是并行分支只写有 reducer 的累积字段比如 messages如果确实需要写不同业务字段则这些字段在两个并行分支之间不要重叠。说到 log_issue 这类日志节点我建议返回字段设计成“追加型”的列表定义时用 Annotated 配合 add_messages 或者自定义 reducer。如果不这样做两个并行分支只有一个日志能被记录另一个会被悄悄覆盖而且不会报错排查起来非常隐蔽。6. 常见问题与排查技巧6.1 高频问题速查表我整理了实际使用中频率最高的一批问题直接做成表格遇到对应症状时对号入座即可。症状常见原因解决思路invoke 之后 messages 只剩最后一条messages 字段没用 add_messages reducer改为Annotated[list, add_messages]多轮调用不记得上一轮没传 checkpointer或 thread_id 每次不同compile 时传入 checkpointer固定 thread_id条件路由报 KeyError路由函数返回的节点没有注册检查 add_node 名称与路由返回值是否一致并行分支状态被覆盖多个节点同时写同一个普通字段让并行分支写不重叠字段或者改用 reducer工作流死循环不停条件边形成了一个无出口的反向环且无循环上限加 loop_count 计数超限强制跳转兜底节点消息重复出现手动把历史拼进 invoke又用了 checkpointerinvoke 只传增量输入分类节点读 messages[-1] 得到 assistant 消息上游刚追加过 assistant 消息最后一条不是用户输入取出 last user message 再传给模型拼接 prompt这些症状我在不同项目里都遇到过尤其前三条几乎是新人上手必踩的。好消息是它们都有一个共同特征报错信息并不直接指向问题根因而是会伪装成“模型回答质量差”或“结果不稳定”。所以排查时先别怀疑模型把状态快照打出来看一遍。6.2 调试状态的实用技巧LangGraph 提供了查看状态快照的能力这是我排查问题的第一工具config {configurable: {thread_id: ticket-001}} snapshot app.get_state(config) print(snapshot.values)它会把当前线程最新的状态完整打出来哪些字段有值、哪些字段缺失一目了然。如果你想回看历史步骤可以用app.get_state_history(config)拿到不同时间点的快照列表。这个功能相当于给你的工作流装了一台黑匣子每一站的状态都被录了下来。我在团队里推行过一个很朴素的习惯每个节点开头打印一行日志包含节点名和一个状态关键字段的摘要。比如print(f[classify] resolved{state.get(resolved)} category{state.get(category)})节点多了之后这些日志拼接在一起基本就是一条可读的工作流 trace。指望靠断点逐个去查多节点状态效率太低日志反而是最可靠的现场还原方式。另外写测试时尽量把节点函数设计成纯函数不依赖模块级全局变量只用传入的 state 和外部依赖。测试可以直接构造一个最小 state 字典调用节点函数断言返回字典里的字段。这样你不需要启动一次完整图就能验证每个节点的行为回归成本会低很多。7. 一些实战体会7.1 什么场景才真正值得用图上手 LangGraph 之后有一段时间我恨不得把所有流程都改成图后来发现过度设计不值。如果业务流程只有两三个固定步骤没有分支、没有循环、不需要多轮记忆直接用简单业务代码或者 Chain 反而更省心。图的优势在于控制复杂代价是把流程声明成图状结构本身也需要额外的学习和维护成本。我目前的选择标准很简单只要满足“会根据中间结果改变后继步骤”或者“需要跨轮保留并恢复状态”任一条件就上 LangGraph否则保持轻量。7.2 我给新项目定的三条规则做多了之后我给使用 LangGraph 的新项目定了几条小规则在这里分享给同样在摸索的朋友。第一状态字段精简化。只放必须跨节点、跨轮次的数据其余临时变量尽量放在节点内部。这能显著减少 reducer 冲突和意外覆盖。第二reducer 先想清楚再写。并行分支要更新什么字段、普通字段由哪个节点独占写入这些在设计阶段就要定下来不要在写完代码后再返工。第三全程用 thread_id 隔离会话。即使只是开发阶段也要把 thread_id 当作业务字段来看待否则很容易出现“测试时会话串台”的诡异问题。串台问题比模型幻觉更难定位因为状态本身看起来完全正常只是内容属于另一个用户。我最近在做一个需要人工确认后退款的工单场景LangGraph 的暂停和恢复机制帮了大忙它允许图在一个节点上停下来等待外部人工结果再通过更新状态继续执行。这种“人在回路”的工作流如果用手写状态机去维护成本会非常高。后面如果有更完整的实践我会再整理出来分享。回到最核心的一条建议不管用什么框架把你节点里的关键状态记成结构化日志排查问题时它比任何优雅的抽象都管用。