上上个月我负责的客服机器人第一次被用户骂了。原因很简单——用户两天前在对话框里报过一次产品型号再次咨询时机器人完全不记得又让对方从头说一遍。我当时的方案是 LangChain 的 Chain 式链路每个请求独立执行天然不带记忆。挨骂之后我去翻了 LangGraph发现它从设计上就是为有状态服务的AI Agent 工作流被建模成一张图状态在节点之间流动分支和循环由条件边控制对话记忆通过 Checkpointer 持久化。这段迁移过程不算轻松但确实解决了之前无状态方案完全没法解决的问题。这篇文章不打算给你堆概念而是把我从入门到实战的完整路径拆给你看先讲清楚为什么需要它再拆解 StateGraph 的四个核心角色然后用一个带记忆的客服 Agent 走通完整代码最后聊聊多轮对话、人工介入和上线后踩过的一堆坑。适合已经写过几个 LangChain 应用、但发现一遇到循环调用工具、记住上下文、暂停等人确认就无从下手的开发者。1. 为什么 Chain 式 Agent 不够用一个客服机器人的翻车现场1.1 从一次失败上线说起老读者可能记得我之前分享过用 LangChain 搭客服机器人的文章。当时用的是 LCEL 风格把识别意图→查订单→生成回复串成一条固定管道。说实话演示阶段跑得很顺畅但一上线就被真实流量揍了。问题出在哪Chain 本质是线性的。数据从入口流到出口中间经过的每个环节都是预先定死的。写固定流程的问答、信息查询没问题可一旦出现根据当前结果决定下一步做什么代码就立刻开始变难看。比如客服场景里最常见的路径用户报订单号Agent 调用查询工具查询结果发现订单已发货就继续查物流轨迹物流轨迹显示异常还要再调一次售后登记工具。这套逻辑在 Chain 里怎么写只能用一堆 if-else 把工具调用塞进一个巨型函数。我当时的代码到了第二周就已经没人愿意动了——你自己都分不清哪个分支在哪个条件下触发更别说让产品经理理解。等到记住用户上次说过什么这个需求出现LCEL 直接崩了。每次请求进来我都得手动把 chat_history 从数据库捞出来拼进 prompt再去调用模型。上下文一长prompt 拼接逻辑、token 控制、历史消息裁剪全挤在一起维护成本涨得飞快。1.2 有状态和无状态的本质差异这里先厘清一个关键概念无状态的 Agent 每次调用都是失忆的。你说一句话我处理一句话返回结果后这轮对话就结束了。你带了 chat_history 参数进来我才能假装记得你你忘了带我就真忘了。有状态的工作流则完全不同。Agent 本身维护一份贯穿整个运行的全局状态——用户历史消息、中间计算结果、工具返回数据、人工审批结果全都放在同一个状态对象里。节点函数可以从状态里读任意数据也可以向状态写入自己的结果。这样一来记住上下文就变成框架层面的能力而不是业务代码里手搓的脏活。我画过一张表对比三种常见工程方案现在分享给你方案流程控制分支与循环对话记忆人工介入恢复执行普通 Chain / LCEL线性管道靠代码硬写手动注入不支持不支持ReAct AgentExecutor内置循环固定策略手动维护不支持不支持LangGraph StateGraph图结构条件边自由路由Checkpointer 持久化interrupt 原生支持断点续跑结论很直白如果你的 Agent 只需要问一句答一句LangChain 的老方案完全够用但一旦涉及多轮对话记忆、条件分支、循环调用工具、暂停等人确认这四个特征里任何一个LangGraph 就是更合理的选择。这不是炫技是架构匹配问题。2. StateGraph 的四个核心角色State、Node、Edge、Checkpointer 到底在干什么LangGraph 的入门门槛其实不高它一共就四个核心概念。把这四样东西吃透你已经能看懂九成以上的开源项目了。2.1 State一张全局共享的便签本State 是整个工作流的共享内存通常用一个 TypedDict 来定义。每个节点函数接收当前 State经过自己的逻辑处理后返回一个部分更新。框架会自动把这部分更新合并回全局 State然后传给下一个节点。举一个最经典的例子from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str注意messages这个字段上的Annotated[list, add_messages]。普通字段的默认合并策略是覆盖——新写入的值直接替换旧值。但add_messages是一个 reducer它告诉框架消息列表不是覆盖而是追加合并。这对多轮对话至关重要因为每轮生成的 HumanMessage、AIMessage、ToolMessage 都应该被保留下来而不是被最后一轮覆盖掉。你可以把 State 理解为公司群里那份共享文档谁都能读谁都能改但每个人只提交自己负责的那部分更新互不干扰。这种设计让节点之间天然解耦——订单节点不需要关心物流节点怎么实现它只需要读取 State 里的订单号字段。2.2 Node 与 Edge执行和流转的分工Node 是工作流里真正干活的函数。它接收整个 State返回部分更新的 dict。前面提到的 assistant 节点、工具执行节点本质上都是普通 Python 函数。Edge 则负责定义节点的流转顺序分两种普通边固定流转。比如graph.add_edge(tools, assistant)表示工具执行完就回到 assistant形成一个循环。条件边根据 State 的内容决定跳转方向。这是 LangGraph 最灵活的地方。条件边的写法是函数返回一个路由 keydef route_after_assistant(state: AgentState): last_msg state[messages][-1] if getattr(last_msg, tool_calls, None): return tools return end graph.add_conditional_edges( assistant, route_after_assistant, {tools: tools, end: END} )这个函数也很简单拿到最新一条 AI 消息如果它带有工具调用请求就跳去 tools 节点执行否则说明模型已经给出最终答复直接走 END 结束。条件边是让 Agent 具备自己决定下一步能力的核心机制谁掌握了它谁就掌握了 AI Agent 工作流的自由度。每个 StateGraph 都有两个特殊节点START入口和END出口。图的构建方式就是声明节点 → 声明边 → 编译非常直观。2.3 Checkpointer让记忆落盘的关键角色如果你只创建了 StateGraph 和节点编译后直接调用那运行结束后 State 就丢了——这跟普通无状态方案没区别。要真正实现有状态必须给图挂上一个Checkpointer检查点。Checkpointer 的作用是在每一步节点执行后把当前 State 完整保存下来。这样至少带来两个能力多轮对话记忆同一个会话 ID 再次调用时自动从检查点恢复历史 State模型就能想起上一轮的内容断点续跑工作流中途中断后可以从最近的检查点继续执行而不是重新跑一遍前面的所有节点。LangGraph 内置的MemorySaver把 State 存在进程内存里适合开发和测试。生产环境通常换成基于 SQLite、Redis 或 Postgres 的持久化实现。后面第四章我会专门展开讲。到这里一个 LangGraph 工作流的基本骨架已经很清晰了State 承载数据Node 执行逻辑Edge 控制流向Checkpointer 负责记忆。接下来我们用它写一个真正能用的客服 Agent。3. 三步搭出一个带记忆的客服 Agent核心代码逐段拆解这节直接给你一套可运行的完整代码。我特意把代码保持得极简——去掉了复杂的 prompt 工程和业务逻辑只保留最核心的结构方便你把注意力放在图本身。3.1 环境准备与 State 定义先装依赖pip install langgraph langchain-openai然后定义 State。这里我加了user_id用于标识当前用户。messages用add_messages合并确保历史消息逐轮累积from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str为什么 State 里不直接放一个order_id这样的业务字段因为客服 Agent 的流程是动态的用户可能先问订单、再问发票、又问退货这些话题之间没有固定顺序。把全部信息都存在messages里让模型从历史消息中提取需要的信息比硬编码业务字段更灵活。user_id则单独拎出来因为它在整个生命周期里是稳定不变的。3.2 节点逻辑assistant 与 tools先说选择这种节点划分的理由。很多新手会把一个功能一个节点画成一堆节点结果图变得极其臃肿。我的经验是最少节点清晰流转。客服场景里只有两类职责一是思考并决定下一步assistant二是执行工具并返回结果tools。两个节点加一条循环边就足够表达完整的 Agent 推理过程。assistant 节点负责调用模型并把回复写入 Statellm ChatOpenAI(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools([get_order_status, get_refund_status]) def assistant(state: AgentState): ai_msg llm_with_tools.invoke(state[messages]) return {messages: [ai_msg]}这里最关键的是bind_tools。它把工具的函数签名注入给模型模型在回答中需要查数据时会生成一个tool_calls字段而不是直接编造答案。AI Agent 的可靠性首先来自该调用工具时就调用工具而不是自己瞎编。tools 节点读取 AI 消息里的tool_calls逐个执行真实函数然后把执行结果包装成ToolMessage塞回 Statedef tools(state: AgentState): last_ai_msg state[messages][-1] tool_msgs [] for tool_call in getattr(last_ai_msg, tool_calls, []): if tool_call[name] get_order_status: result get_order_status.invoke(tool_call[args]) elif tool_call[name] get_refund_status: result get_refund_status.invoke(tool_call[args]) else: result 未知工具 tool_msgs.append( ToolMessage(contentstr(result), tool_call_idtool_call[id]) ) return {messages: tool_msgs}注意tool_call_id一定要匹配上否则模型无法把工具结果和对应调用对应起来。这个字段是我最开始最容易漏掉的坑之一。3.3 建图、编译与运行从第一轮到失忆的第二轮建图的过程就是声明节点、声明边、编译def route_after_assistant(state: AgentState): last_msg state[messages][-1] return tools if getattr(last_msg, tool_calls, []) else end graph StateGraph(AgentState) graph.add_node(assistant, assistant) graph.add_node(tools, tools) graph.add_edge(START, assistant) graph.add_conditional_edges(assistant, route_after_assistant, {tools: tools, end: END}) graph.add_edge(tools, assistant) saver MemorySaver() app graph.compile(checkpointersaver)流转路径是这样的START → assistantassistant 判断是否需要调工具需要则assistant → tools → assistant循环直到模型不再产生工具调用最后assistant → END。整个循环天然支持多轮工具调用比如先查订单状态再根据结果查退款进度。现在跑第一轮对话result app.invoke( {messages: [HumanMessage(content订单 10086 现在到哪了)], user_id: 233}, config{configurable: {thread_id: user-233}} )模型会先调用get_order_status拿到结果后生成回复您的订单正在运输途中预计明天送达。由于我们传入了thread_id这一步结束时完整的 State 已经被 Checkpointer 存下来了。第二轮对话我故意不把历史消息传进去result2 app.invoke( {messages: [HumanMessage(content那退款呢处理到哪一步了)]}, config{configurable: {thread_id: user-233}} )奇迹发生了模型记得您刚才问的订单 10086不需要我手动拼接任何历史消息。这就是 Checkpointer 的威力——记忆是工作流自动维护的而不是业务代码手工搬运的。你对比一下原来用 Chain 时手写 chat_history 的酸爽就知道这个设计有多值。4. 让有状态真正落进生产MemorySaver、thread_id 与上下文管理跑通 Demo 只是开始。真要把这套东西部署到线上有三个问题躲不开记忆存在哪、会话怎么隔离、历史消息无限增长怎么办。4.1 MemorySaver 只适合开发和测试MemorySaver把检查点存在进程内存里有两个致命缺陷进程重启所有记忆全部丢失多实例部署时不同实例各自保存各自的记忆用户请求被负载均衡打到不同实例历史会话就错乱了。生产环境必须用共享存储。LangGraph 提供了多种持久化实现我实测过的方案对比如下持久化方案适用规模优点缺点MemorySaver单机测试零配置、快重启丢数据不支持多实例SqliteSaver中小项目单文件、易部署单机写入存在锁竞争PostgresSaver / Redis 等生产环境支持多实例、并发好需要额外基础设施我目前用的方案是团队自己实现的基于 Redis 的 Checkpointer方式是在compile时传入实现了 Savepoint 接口的对象。接口本身不复杂核心就是把每个检查点序列化后按 thread_id 保存。注意一点改成持久化实现后原来接口调用方式完全不变该传thread_id还是传thread_id对业务代码几乎零侵入。提示从这个角度看LangGraph 把 Checkpointer 设计成一个可替换的独立组件是它比很多自研框架高明的关键原因——记忆机制和业务逻辑彻底解耦切换存储只需换一个实现类。4.2 thread_id 是会话隔离的钥匙thread_id不是什么深奥机制它就是一个会话主键。同一个thread_id共享同一份记忆不同的thread_id之间完全隔离。我在工程里的做法是把用户登录态的会话 ID 或用户 ID 直接作为thread_idconfig {configurable: {thread_id: fuser_{user_id}}}这样做的好处是天然支持多用户并发。用户 A 和用户 B 的请求同时进来各自的 State 互不干扰。如果你在做了隔离之后发现 A 能看到 B 的记忆那一定是thread_id拼错了而不是框架的问题——这是我排查过的真实案例。值得警惕的反模式是把thread_id写死成一个固定值。之前我接手过一个遗留项目为了省事全局用一个 thread_id结果所有用户的对话全串在一起客服系统差点变成公开聊天室。排查了半天才发现是这种低级错误。4.3 状态会疯长消息修剪与摘要有状态带来的副作用是State 里的messages会随着对话轮数持续膨胀。几十轮对话后每次请求都要把全部历史发给模型token 成本飙升响应延迟也跟着涨。我目前用的两种策略你可以按需组合方案一滑动窗口修剪。只保留最近的 N 条消息。通过在assistant节点入口对state[messages]做截断实现def trim_messages(state: AgentState): max_turns 20 if len(state[messages]) max_turns: state[messages] state[messages][-max_turns:] return state这种方案简单粗暴适合客服问答场景——最近的上下文足够模型理解当前意图。方案二早期消息摘要压缩。把较早的对话抽出来让模型生成一段摘要替换掉原始消息。这个方案能保留更长时间维度的信息代价是需要额外的模型调用成本。我的经验是以下单、查物流、退货为主流程的 Agent滑动窗口就够用了如果你做的是那种需要长期记忆用户偏好的推荐型 Agent摘要方案才更有价值。千万不要一上来就追求无限记忆记忆越多噪声越大模型反而更容易被无关历史误导。5. 中断与恢复让真人插进 Agent 工作流如果 Agent 只能单机跑完全程那很多事情是不敢交给它的——比如让模型直接操作退款、删除数据库、发送营销消息。这类高风险动作必须有一个暂停等真人确认的环节。LangGraph 原生支持这个能力这对我而言是它最香的地方。5.1 interrupt() 到底做了什么interrupt()是 LangGraph 提供的一个特殊函数在节点内部调用后工作流会立即暂停并把传入的参数作为待人工处理的任务暴露给外部。此时整个 State 被 Checkpointer 完整保存进程可以继续干别的甚至可以重启。外部程序可以通过检查点查询到这个暂停状态拿到需要人工确认的内容。确认完成后用一个Command(resume...)把结果塞回去工作流从暂停处继续执行仿佛什么都没发生过。这个机制本质上就是把人机协作变成一个普通的后台任务模型负责提出方案人负责做决定代码负责执行。这句话我用了一年多越用越觉得精辟。5.2 一个退款审批的完整链路给你看一个我实际做过的退款审批场景。用户要求退款Agent 生成退款方案先把方案提交给管理员审批审批通过后才真正执行退款调用。节点的核心逻辑from langgraph.types import interrupt, Command def refund_agent(state: AgentState): plan generate_refund_plan(state) approved interrupt( { type: refund_approval, amount: plan[amount], order_id: plan[order_id], reason: plan[reason], } ) if approved: result execute_refund(plan) else: result 退款未获批准 return {messages: [{role: assistant, content: str(result)}]}当interrupt被调用时app.invoke不会继续执行后面的代码而是返回当前暂停信息。人类管理员看到用户 233 申请退还 399 元点击通过后外部程序调用app.invoke( Command(resumeTrue), config{configurable: {thread_id: user-233}} )interrupt调用得到True节点继续执行execute_refund(plan)。整个过程跨了三个系统——Agent、审批前端、退款后端但用户感知到的只是进度暂停了一下。提示Command(resume...)传的值会直接成为interrupt()的返回值。传 True/False 可以传审批备注字符串也行完全按你的业务需求来。5.3 中断状态恢复的工程注意点这块的坑不少踩一遍才长记性。我提炼三个比较容易踩的状态恢复后节点幂等性。工作流恢复后确实是从暂停处继续但如果后续节点调用了外部 API要注意 API 本身的幂等性。退款这种操作尤其危险——万一网络抖动导致恢复执行了两遍就会退两次款。我通常会在退款工具里加一层去重逻辑用订单号做幂等键保证同一张订单不可重复退款。暂停超时处理。人工审批不可能永远等下去。我在生产环境加了定时扫描超过 24 小时未审批的暂停任务自动用Command(resumeFalse)关闭并给用户回一句审批超时请重新提交。暂停说明要对人不对话。interrupt传入的数据是给外部系统和处理人员看的应该包含完整上下文比如订单号、金额、用户备注而不是有一笔待审批。前期我犯过这个错审批同事在系统里看到一团含糊的提示还得去翻用户聊天记录效率极低。6. 上线两个月后我踩过的坑条件边路由、reducer 冲突与调试技巧最后这部分我挑三个真实踩过的坑每一个都让线上环境抖过几秒钟。写出来帮你绕开。6.1 条件边返回的 key 一定要在路由映射表里条件边的机制是路由函数返回一个字符串 key框架根据映射表把它转换成目标节点。我第一次写的时候映射表里只有tools: tools漏了end: END结果模型正常结束对话时路由函数返回end框架直接抛 KeyError。报错信息还特别隐晦指向的是引擎内部代码不熟悉的话根本想不到是自己映射表漏了。检查有两点路由函数所有可能返回的 key 都必须出现在映射表里映射表里的 value 必须是图里已存在的节点名或者是END。一个小建议路由函数的返回值可以固定为end而不是直接返回END对象。因为END是一个特殊对象如果你不小心把END直接作为 key 返回而映射表里又没定义排查起来会更绕。6.2 reducer 用错状态会被静默覆盖前面讲过Annotated[list, add_messages]的作用。如果哪天你把某个关键字段定义为普通 dict 而没有加合并策略那么每个节点写入时都会整体覆盖旧值。我有一个真实案例State 里存了scratchpad字段用来记录中间推理结果。因为忘了配 reducerassistant 节点写入新内容时把之前的草稿覆盖了后面工具节点读到的永远是最后一个节点的输出导致整个推理链断裂。而那段时间我没看 State 日志一直在怀疑是模型问题白白浪费了两天。经验总结任何需要累积、追加、汇总语义的字段都必须显式声明 reducer。除了add_messagesLangGraph 也支持自定义 reducer 函数比如求和、去重、保留最大值。把 State 字段语义想清楚再动手能省掉大量半夜排查的时间。6.3 调试 LangGraph 工作流的三种手段Graph 抽象的调试比 Chain 难一些因为你面对的不再是一个线性列表。我常用的调试手段有三个第一种单独测试节点函数。节点就是普通函数直接构造一个假 State 调用它验证输入输出。这能最快定位是节点逻辑坏了还是图结构错了。第二种检查点日志输入。在编译后每次 invoke 拿到完整结果打印 State 里每个关键字段的变化轨迹。看看模型生成的tool_calls、工具返回的ToolMessage、路由函数的返回 key——这条链路追一遍绝大多数问题都能暴露。第三种可视化图谱。需要可视化支持的话可以用配套开发者工具在浏览器里观察每一步的执行顺序也可以看中断点、回放和手动修改 State 继续跑。这是排查复杂分支循环场景最顺手的工具。我现在每改一版图都会先可视化跑一遍流程再上生产。调试这件事我最大的心得是别把图想得太玄。它再复杂也是一个个普通函数和一条条跳转规则拼出来的。把每个函数拆出来测一遍把每个边的走向打出来看一眼问题很少能藏得住。最后再分享一个我这几个月的习惯每次给图新增节点或边之前先写一版最小复现脚本用 3 条以内的消息把新流程跑通再搬到真实场景里去丰富 prompt 和业务细节。LangGraph 的抽象决定了它适合先小后大的开发节奏细节放在后期打磨完全来得及——但图的主干结构一旦定错返工成本会高得让你怀疑人生。