LangGraph 是近两年 LLM 应用开发里绕不开的一个名字。我最初是被它的“有状态”这三个字吸引过去的因为之前用常见的 Agent 框架做多步骤任务时最头疼的就是状态管理——每个节点各自为政对话历史要手动拼接分支逻辑要写在业务代码里一旦流程复杂起来代码很快就变成一团乱麻。LangGraph 把 Agent 工作流抽象成了一张图节点是处理逻辑边是流转规则而状态则是贯穿全图的“共享变量”这套思路很对我这种后端出身的人的胃口像在写一个有向无环图只是它额外支持了循环、分支、中断和恢复真正把一个 AI Agent 变成一个可以控制的程序。这篇文章我会围绕“有状态的 AI Agent 工作流”这个核心从 LangGraph 的状态设计讲起手写一个带条件路由的 Agent再做一个带人工审批的实用案例最后把我和团队在项目里踩过的坑、总结的调试方法一并分享出来。无论你是刚开始接触 LangGraph还是已经用它搭过简单流程想进阶这篇应该都能给你一些可复用的东西。1. 先搞清楚 LangGraph 解决的是什么问题1.1 为什么普通 Agent 框架不够用先说说我为什么从其他框架转过来。用常见的 ReAct Agent 框架时整体是一个黑盒循环模型决定调用什么工具工具返回结果模型再决定下一步。这个循环在简单场景下没什么问题但一旦遇到下面这些需求就非常别扭中间步骤需要人工确认比如转账、发邮件、删除数据这些操作不能全权交给模型。不同任务需要走完全不同的处理流程比如一个客服机器人要根据用户意图分别走退货、催单、故障排查三条链路。需要持久化、断点续跑比如一个耗时任务跑到一半崩溃了下次要能接着跑。需要控制模型访问状态的粒度有些中间变量不能给模型看但需要在节点间传递。这些需求本质上都指向同一个关键词可控性。LangGraph 的主张很干脆把流程控制权从模型手里拿回来模型只负责“决策”而“图”负责“流程”。你可以把 LangGraph 想象成一张地铁线路图模型在站点之间选择怎么走而轨道、换乘站、终点都是预先定义好的。1.2 有状态与无状态的本质区别要理解 LangGraph 的有状态先要理解它对比无状态的优势。无状态调用比如直接调大模型 API每次请求都是独立的服务端不记住你之前说过什么。你需要自己把聊天记录拼进 Prompt自己维护上下文变量任何一个节点重启前面的信息就全丢了。有状态则是在整张图的执行过程中维护一个 State 对象这个对象会随着图的运行不断更新每个节点都能读取它、修改它、基于它做决策。关键点有三个第一状态是全局共享的但不一定是全局可见的。节点可以只读取自己关心的字段。 第二状态更新是有机制约束的每个节点都可以声明自己新增或修改哪些字段未声明的字段不会被变更。 第三状态可以持久化LangGraph 支持将状态保存到数据库或内存中图表执行到一半中断之后可以从检查点恢复。这个设计有点像 Web 开发里的共享 Session也有点像工作流引擎里的流程变量。但 LangGraph 的状态更灵活它支持 TypedDict、Pydantic 模型、Dataclass 等多种定义方式甚至可以存任意 Python 对象。1.3 LangGraph 的核心抽象图、节点、边、状态LangGraph 的核心抽象只有四种理解了这四种整个框架的骨架就搭起来了。节点Node是最小处理单元接收 State 作为输入返回更新后的字段。节点内部可以调用大模型、执行工具函数、查询数据库什么都能做框架不限制你。边Edge定义节点之间的流转方式。普通边是“执行完 A 一定执行 B”条件边则是“根据某个条件决定下一个去哪个节点”。这个条件逻辑是一段普通 Python 函数返回一个字符串作为目标节点名称非常灵活。状态State是贯穿全图的“共享数据中枢”用一个 TypeDict 结构定义。每个节点可以声明自己要更新哪些键框架会对你声明的键执行更新操作然后把合并后的新状态传给下一个节点。图Graph则是把节点和边组装起来执行时从一个入口开始按照边的定义在节点间穿梭直到到达终止节点。LangGraph 提供了两种图的定义方式低阶的 Graph API 和高阶的 StateGraph API。日常开发用 StateGraph 就足够了它能自动帮你管理状态合并逻辑。2. 环境准备与核心概念上手2.1 安装与版本选择LangGraph 是 Python 生态的库安装非常直接。pip install langgraph它会自动拉取 langchain-core 作为底层依赖。如果你想在 Jupyter 或者脚本里直观看到图的运行过程建议再装两个辅助库pip install langgraph-checkpoint-sqlite这个库的作用是提供 SQLite 后端的持久化检查点没有它的话状态默认只存在内存里进程一结束就没了。日常学习阶段可以先不装但要做真正的 Agent 应用我建议从一开始就加上。版本方面我现在的建议是尽量用 0.2 及以上版本早期 0.0.x 的 API 变化非常大网上很多教程现在已经跑不通了就是因为版本太老。装完之后可以执行下面这段代码验证一下import langgraph print(langgraph.__version__)顺带说一句LangGraph 是一个纯 Python 库不依赖 LangChain 的完整生态你完全可以只用 langchain-core 的基础模型封装再配合自己的普通 Python 工具函数来构建工作流。它的核心图执行引擎是独立于模型调用层的这一点很多人容易误解以为必须全家桶。实际上不是你甚至可以直接用一个把字符串变成大写字母的普通函数作为节点体验整个机制。2.2 状态对象State的设计状态设计是整个 LangGraph 项目里最重要的决策。为什么我敢这么说因为状态的 schema 一旦定下来后面每个节点、每条边、每个持久化的检查点都围绕它展开改一次状态结构的成本非常高。用 TypedDict 定义一个状态这是官方文档最常用的方式也是我推荐新手起步的方式。from typing_extensions import TypedDict class AgentState(TypedDict): messages: list current_step: str result: dict三个字段各有讲究。messages 字段用来储备对话历史current_step 用来记录当前执行进度result 用来存最终输出。但这里有一个很容易被忽略的问题默认情况下节点返回的 dict 会直接覆盖对应键的旧值。比如第一个节点返回 {messages: [msg1]}第二个节点又返回 {messages: [msg2]}那消息列表会被后者整个替换掉而不是追加。要解决这个问题就要用到 Annotation。Annotation 负责定义状态的“更新策略”from typing_extensions import TypedDict from langgraph.graph import MessagesState class AgentState(TypedDict): messages: Annotated[list, operator.add] current_step: str result: dict加了Annotated[list, operator.add]之后messages 字段的更新行为就从覆盖变为合并。每次节点返回的新消息都会追加到原列表后面。这是 LangGraph 状态设计中最关键的一个机制我见过很多新手项目运行结果很怪异都是忽略了这层合并逻辑。这里需要停下来解释一个细节为什么官方推荐直接用MessagesState因为它内部已经帮你定义好了 messages 字段并配置了合适的合并策略还预设了一个add_messages函数来处理消息对象。如果你做的是对话类 Agent继承它比自己造轮子更省心from langgraph.graph import MessagesState class AgentState(MessagesState): current_step: str result: dict这样做的好处是 messages 字段自动支持消息列表追加合并你只需要专注扩展自己业务需要的字段。关于字段的可见性还有一个设计建议。有些中间变量是整个流程的“工作内存”比如工具返回的 JSON、上一步计算出的中间值这些不应该暴露给大模型但它们确实需要在节点间传递。我的做法是给状态里加一个hidden: dict字段专门存放内部数据并且不把它放进任何 Prompt 模板里。这样做既保证了信息流通又避免了 Prompt 被无关信息撑爆。2.3 从零构建一个最简单的有状态图纸上谈兵到这里直接跑一个最小的例子让整个执行机制在你脑子里立体起来。from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class SimpleState(TypedDict): total: int logs: list def add_one(state: SimpleState) - dict: new_total state[total] 1 return {total: new_total, logs: [fadd_one: {state[total]} - {new_total}]} def double(state: SimpleState) - dict: new_total state[total] * 2 return {total: new_total, logs: [fdouble: {state[total]} - {new_total}]} graph StateGraph(SimpleState) graph.add_node(add_one, add_one) graph.add_node(double, double) graph.add_edge(START, add_one) graph.add_edge(add_one, double) graph.add_edge(double, END) app graph.compile() result app.invoke({total: 1, logs: []}) print(result)执行结果如下{total: 4, logs: [add_one: 1 - 2, double: 2 - 4]}注意我刚才故意演示了一个“错误”的写法logs 字段我没有加合并策略第二个节点返回的 logs 会把第一个节点返回的 logs 覆盖。但打印结果里两行都在对吧这是怎么回事原因在于 StateGraph 对未声明 Annotation 的字段默认策略是“后写覆盖”但实际运行时每个节点返回的 updates 都会先临时聚集然后在状态图上一次性应用。节点 add_one 的返回键和节点 double 的返回键如果相同且没有合并策略理论上 double 会覆盖 add_one 的结果。这里 logs 键名一样实际执行时确实只会保留后一个节点的值。我打印出来两行都显示是因为我在两个节点里都把之前的 state[total] 拼接进了日志文本而真正的 logs 列表已经被覆盖了。为了得到真正的追加效果你应该把 logs 字段声明成合并策略from typing_extensions import TypedDict, Annotated class SimpleState(TypedDict): total: int logs: Annotated[list, operator.add]这个例子虽然简单但它把 LangGraph 最底层的两条规则装进了你的脑子节点修改状态是通过返回值声明式完成的字段合并策略决定数据用“覆盖”还是“追加”的语义。后面所有实战都会踩着这两条规则展开。3. 构建有状态的 AI Agent从简单到复杂3.1 完整示例带条件路由的 Agent 工作流现在进入实用阶段——构建一个真正与模型交互、并且带条件路由的 Agent。这个 Agent 的场景是用户发起一句话请求Agent 判断是否需要调用工具如果需要则执行工具查询否则直接回复。我先定义一个带工具查询的模拟工具比如查天气。为了演示清晰这里用一个假函数代替真实 APIdef get_weather(city: str) - str: weather_data { 北京: 晴气温 35 度, 上海: 多云气温 28 度, 广州: 阵雨气温 26 度, } return weather_data.get(city, 暂无数据)然后是 Agent 的两个处理节点一个负责“决策”一个负责“执行工具”。from typing_extensions import TypedDict, Annotated import operator from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: Annotated[list, operator.add] city: str weather: str llm ChatOpenAI(modelgpt-4o-mini, temperature0) def agent_node(state: AgentState) - dict: system_prompt 你是一个助手。如果用户询问天气请只输出城市名不要输出其他内容 如果不涉及天气直接回答用户问题。 response llm.invoke([ {role: system, content: system_prompt}, *state[messages] ]) content response.content.strip() # 如果输出恰好是城市名说明需要查天气 if content in [北京, 上海, 广州]: return {messages: [response], city: content} else: return {messages: [response]} def weather_node(state: AgentState) - dict: weather_info get_weather(state[city]) msg {role: assistant, content: f{state[city]}的天气是{weather_info}} return {messages: [msg]} def should_call_weather(state: AgentState) - str: if state.get(city): return weather return end graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(weather, weather_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, should_call_weather, { weather: weather, end: END }) graph.add_edge(weather, END) app graph.compile()这个工作流的逻辑链路是用户提问进入 agent 节点模型决定是否输出一个城市名条件边根据状态里的 city 字段决定下一跳是天气节点还是直接结束。天气节点查询后把结果写回消息列表。执行测试result app.invoke({messages: [{role: user, content: 北京天气怎么样}]}) for msg in result[messages]: print(msg.get(role) :, msg.get(content))运行结果类似于user: 北京天气怎么样 assistant: 北京 assistant: 北京的天气是晴气温 35 度这个例子看起来简单但它已经把 LangGraph 最核心的条件路由机制演示完整了。关键点在should_call_weather这个函数上它不是节点而是边上的“路由裁判”它读取当前状态返回一个字符串然后图根据映射表跳到对应的节点。这种模式是 LangGraph 区别于普通 chain 的核心能力——同一个状态由模型或者规则动态决定下一步走哪条路。在实际项目中路由裁判函数可以非常复杂它可以从状态里读取 sentiment、intent、tool_choice 等字段甚至调用另一个模型做意图分类。这种方式即便大模型偶尔决策失误路由逻辑仍然在代码的控制范围内不至于完全失控。3.2 模拟人工审批异步中断与 Human-in-the-loop聊完了条件路由接下来做一个更接近生产环境的功能人工审批。为什么 Agent 需要这个如果一个工作流自动读取了数据库里的订单、执行了退款操作、发送了营销邮件这三个动作有一个出错后果都很严重。我习惯的底线是凡是涉及资金、隐私、对外发布的动作必须插入人工确认环节。LangGraph 实现人工审批有几种方式我重点讲最适合入门的一种interrupt机制。它的思路很简单图执行到某个节点时暂停将控制权交还给调用方比如你的 Web 后端调用方在获得用户确认后再向同一个线程 ID 继续执行图。先看一个完整的例子from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import interrupt, Command class ApprovalState(TypedDict): request: str approved: bool final_result: str def human_approval(state: ApprovalState) - dict: # interrupt 会暂停图并向调用方展示这个值 user_decision interrupt({ prompt: f是否批准执行{state[request]}, allow_yes: 批准, allow_no: 拒绝 }) if user_decision.lower() yes: return {approved: True} else: return {approved: False} def execute_action(state: ApprovalState) - dict: if state[approved]: return {final_result: f已执行操作{state[request]}} return {final_result: 操作已被用户取消} graph StateGraph(ApprovalState) graph.add_node(approval, human_approval) graph.add_node(execute, execute_action) graph.add_edge(START, approval) graph.add_edge(approval, execute) graph.add_edge(execute, END) app graph.compile(checkpointermemory_checkpointer)这里需要注意要用 interrupt 就必须配置 checkpointer否则图执行到一半无法暂停、无法从暂停点恢复。Checkpointer 可以简单理解为一个“存档点管理器”它把图的每个执行片段保存下来。上面代码里的memory_checkpointer可以是from langgraph.checkpoint.memory import MemorySaver memory_checkpointer MemorySaver()在生产环境建议用from langgraph.checkpoint.sqlite import SqliteSaver sqlite_checkpointer SqliteSaver.from_conn_string(checkpoints.db)配置好之后第一次调用图时会挂在 approval 节点第二次调用时传入用户确认结果图会从暂停的地方继续。thread_config {configurable: {thread_id: task-001}} # 第一次调用启动任务停留在审批节点 result app.invoke({request: 退款 100 元给用户}, configthread_config) print(result) # 只会执行到 interrupt 为止不会继续 # 用户点“批准”后继续执行 result2 app.invoke(Command(resumeyes), configthread_config) print(result2[final_result])这里有个非常重要的经验教训thread_id是唯一的任务标识两次调用必须使用同一个 thread_id图才知道要从哪个中断点恢复。如果第二次调用换了 thread_idLangGraph 会认为这是一个全新任务重新走一遍流程。这个坑我掉进去过不止一次。为什么 interrupt 比手动写一个“保存待审批任务到数据库再轮询数据库状态”的方案要好因为 interrupt 不需要你自己维护暂停状态、续跑状态和上下文恢复逻辑整个图的调用栈、局部变量、节点参数都被 checkpointer 完整保留。你不需要去数据库中保存“当前执行到第几步”这种元信息架构上更简洁也不容易出现状态与实际流程错位的问题。3.3 循环与记忆多轮对话中的状态管理前面两个例子里图都是一轮执行结束就终止。但很多真实 Agent 场景是需要循环的比如多轮对话里用户反复查询不同城市天气、反复修改需求这时工作流要能在一张图内连续运转并且把历史消息一直带在状态里。LangGraph 实现循环靠的是“把终点接回起点”的边配置。看起来很简单但这里藏着状态合并策略能否正确生效的试金石。因为多轮对话中messages 字段必须使用追加合并策略否则每一轮都会把上一轮的 history 覆盖掉。我做一个多轮对话工作流的骨架from typing_extensions import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class ChatState(TypedDict): messages: Annotated[list, operator.add] current_task: str def chat_node(state: ChatState) - dict: # 这里会调用大模型见下面的完整代码 pass def has_tool_call(state: ChatState) - str: 判断当前用户的最后一条消息是否需要走工具流程 if state[current_task] weather: return tool return reply graph StateGraph(ChatState) graph.add_node(chat, chat_node) graph.add_node(tool, tool_node) graph.add_edge(START, chat) graph.add_conditional_edges(chat, has_tool_call, { tool: tool, reply: chat # 重点reply 条件回到 chat形成循环 }) graph.add_edge(tool, chat)这种写法在很多基于 LangGraph 的 Agent 演示项目里都能看到chat 节点负责生成回复或提取工具参数工具节点负责真实查询查询完成后把结果以 assistant 消息的形式写回 messages然后流程回到 chat模型看到工具结果后再组织最终回复。这里有一个“终止风险”需要特别重视循环必须有明确的终结条件否则图会无限循环下去。LangGraph 有recursion_limit这个参数默认是 25超过步数会抛异常。但靠异常终止不是负责任的做法我建议在条件路由函数里明确设置“完成条件”只有满足完成条件时才跳转到 END 节点。举个例子def route_after_tool(state: ChatState) - str: if state[current_task] completed: return end return chat当工具节点成功返回结果后它会把current_task标记为 “completed”这样下一轮 chat 计算出答案后就能正常跳转到 END工作流不会原地打转。再补充一点记忆管理。状态里存消息列表是直接简单的方式但它有明显的隐患如果对话足够长消息总量超过模型上下文窗口调用直接报错或者被截断。我在实际项目中常用的手法有两个一是滑动窗口只保留最近 N 轮消息把更早期的消息压缩成一个大模型生成的摘要。这个摘要可以作为一条 system 消息放在最前面占用的 token 远小于完整历史。二是“双层记忆”架构短期的对话历史存到 messages 里长期的结构化信息比如用户偏好、关键业务实体存到独立的 memory 字段只在需要时拼进 Prompt。这样即使对话了很久工作流也能把关键信息保留下来而不需要每次把整个历史重复灌给模型。循环和记忆这两个能力叠起来LangGraph 才真正开始像个“能干活”的 Agent 框架而不只是流程编排的玩具。4. 实战跨平台自动化客服 Agent4.1 场景拆解与图设计理论环节到这里用一个完整的实战案例把所有概念串起来。我一直认为只会在 Jupyter 里跑 demo 不算会能把一个真实业务拆成图、落地成代码、还能稳定上线维护才算真的上手了。这里我以一个跨平台客服 Agent 为案例全部代码可以照着跑业务逻辑做了简化但整体设计思路和我在企业项目里的做法是一致的。场景需求如下用户在网页、小程序等平台发消息。Agent 判断用户意图分为退换货、物流查询、商品咨询三类。物流查询直接调用后端 API 查询然后回复用户。退换货需要先查询订单状态再进行退款操作。退款操作需要人工审批。商品咨询直接由大模型回答。所有节点都要把消息记录写回状态。画成图大概是这样的START - 意图识别 意图识别 - 物流查询条件 意图识别 - 退换货条件 意图识别 - 商品咨询条件 物流查询 - 调用物流API - 回复 - END 退换货 - 调用订单API - 人工审批 - 执行退款 - 回复 - END 商品咨询 - 直接回复 - END4.2 环境准备与状态定义先把需要的库和服务连好。我默认你已经装好了 langgraph 和 langchain-openai。from typing_extensions import TypedDict, Annotated import operator from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver llm ChatOpenAI(modelgpt-4o-mini, temperature0)定义状态。这个状态比前面几个例子复杂一些因为要跨节点传递订单信息、审批结果、回复内容class CustomerServiceState(TypedDict): messages: Annotated[list, operator.add] intent: str order_id: str refund_approved: bool api_result: dict final_reply: str字段解释messages对话历史追加合并。intent意图识别的结果类型是 classify。order_id用户提到或者系统查到的订单号。refund_approved退款审批结果。api_result内部 API 查询返回的原始数据。final_reply最后要回复用户的内容。4.3 节点实现逐步拆解第一个节点意图识别。这个节点的工作是把用户输入分门别类。def intent_node(state: CustomerServiceState) - dict: prompt f 将用户的问题分类为以下三种意图之一 - logistics查询物流、快递、配送 - refund退货、退款、换货 - product商品咨询、使用问题 用户问题{state[messages][-1].content} 只输出一个单词不要输出解释。 result llm.invoke(prompt) intent result.content.strip().lower() return {intent: intent}第二个节点物流查询节点。这个节点连接后端订单系统的 API 查询物流信息。def logistics_node(state: CustomerServiceState) - dict: order_id state.get(order_id, unknown) # 这里是模拟接口调用实际项目里可以替换成你的后端 API import json api_result {order_id: order_id, status: 已签收} reply f您的订单 {order_id} 当前状态是{api_result[status]} return { api_result: api_result, final_reply: reply, messages: [{role: assistant, content: reply}] }第三个节点退换货节点。这里要分两步——先查订单状态再走人工审批和退款执行。为了演示清晰我把“查订单”和“执行优惠处理决策”分开。def refund_query_node(state: CustomerServiceState) - dict: order_id state.get(order_id, unknown) api_result {order_id: order_id, can_refund: True, price: 129.00} return {api_result: api_result}第四个节点人工审批节点。这里用前文讲过的 interrupt。from langgraph.types import interrupt, Command def human_refund_approval(state: CustomerServiceState) - dict: price state[api_result][price] decision interrupt({ type: refund_confirmation, order_id: state[order_id], amount: price, message: f用户申请退款 {price} 元请确认是否同意 }) return {refund_approved: decision yes}第五个节点执行退款。def execute_refund_node(state: CustomerServiceState) - dict: if state[refund_approved]: reply f退款 {state[api_result][price]} 元已执行预计 1-3 个工作日到账。 else: reply 您的退款申请已被拒绝如有疑问请联系人工客服。 return { final_reply: reply, messages: [{role: assistant, content: reply}] }第六个节点商品咨询节点。def product_node(state: CustomerServiceState) - dict: user_question state[messages][-1].content reply llm.invoke(f用简短、友好、专业的方式回答用户{user_question}).content return { final_reply: reply, messages: [{role: assistant, content: reply}] }节点实现要注意一个细节所有节点函数都返回一个 dict这个 dict 只会包含这个节点想要修改的键不用把整个 state 原样返回。LangGraph 会拿这个 dict 和旧 state 做合并生成新的完整状态。想改哪个键就返回哪个键非常简洁。千万不要为了“安全感”把所有字段都返回一遍那样很容易把别的节点写入的值覆盖掉而且靠覆盖策略运行时会发生难以排查的诡异问题。4.4 条件路由与图的组装现在把节点和边装起来重点看条件路由函数。def route_after_intent(state: CustomerServiceState) - str: intent state.get(intent, ) if intent logistics: return logistics elif intent refund: return refund_query else: return product def route_after_refund_query(state: CustomerServiceState) - str: if state[api_result][can_refund]: return approval return product # 不能退就直接当咨询回复 def route_after_refund(state: CustomerServiceState) - str: return END graph StateGraph(CustomerServiceState) graph.add_node(intent, intent_node) graph.add_node(logistics, logistics_node) graph.add_node(refund_query, refund_query_node) graph.add_node(approval, human_refund_approval) graph.add_node(execute_refund, execute_refund_node) graph.add_node(product, product_node) graph.add_edge(START, intent) graph.add_conditional_edges(intent, route_after_intent, { logistics: logistics, refund_query: refund_query, product: product }) graph.add_edge(logistics, END) graph.add_edge(refund_query, approval) graph.add_edge(approval, execute_refund) graph.add_edge(execute_refund, END) graph.add_edge(product, END) checkpointer SqliteSaver.from_conn_string(customer_service.db) app graph.compile(checkpointercheckpointer)编译这一步没什么额外开销但它会帮你检查图结构是否合法比如有没有节点没有连接、有没有多边冲突。编译出错总比执行到一半出错好。测试一下物流查询流程config {configurable: {thread_id: cs-1001}} result app.invoke({ messages: [{role: user, content: 帮我查一下订单 100 号的物流}] }, configconfig) print(result[final_reply])退换货流程测试config2 {configurable: {thread_id: cs-1002}} # 第一次调用停在审批节点 app.invoke({ messages: [{role: user, content: 我要退款订单号 200 号}] }, configconfig2) # 模拟人工点击“同意” final app.invoke(Command(resumeyes), configconfig2) print(final[final_reply])至此一个跨平台客服 Agent 的完整骨架就落在代码里了。它做的事情比最初的承诺更符合真实业务基于意图分流、查询后端 API、插入了人工审批流程。所有节点都可以在不改图结构的情况下单独替换比如把物流查询从模拟数据替换成真实的 HTTP 调用或者把大模型从一种换成另一种。4.5 图结构验证与状态可视化关于调试和验证这里多说一点。LangGraph 编译后的对象提供了一些方法可以帮助你确认图结构是否符合预期。# 打印所有节点 print(app.get_graph().nodes) # 打印所有边 print(app.get_graph().edges)如果安装了langgraph-cli或者langgraph-dev你还可以在可视化界面里看到整张图长什么样。我个人的习惯是每加一个节点就画一遍结构图让“节点-边”的演进过程保持在我脑子里而不是等画完再一次性连。因为条件边多的情况下节点之间谁跳谁必须在设计阶段理顺如果到调试阶段才发现跳错了排查成本比重新设计还高。从深入使用的角度再给一个建议一定要把节点函数写成“纯函数”。纯函数的意思是同样的输入一定得到同样的输出不依赖外部可变状态不做副作用或者把副作用封装在独立的节点里。这样做的直接好处是当 LangGraph 因为 checkpoint 恢复、重试、或并行调试时你不必担心同一个节点被调用两次会带来两次副作用。在实际项目里我见过有人把“发邮件”写进了查询节点结果重跑一次就给用户发了两封邮件非常尴尬。5. 常见问题、调试技巧与避坑指南5.1 常见问题速查表我用表格把我在实际项目里频率最高的问题整理出来每个问题都附上了核心原因和解决方案。这份速查表比长篇大论更有用建议直接存一份。问题现象核心原因解决方式消息列表被覆盖而不是追加messages 字段没有配置 Annotated 合并策略给字段加Annotated[list, operator.add]或直接继承 MessagesState图执行停在 interrupt 后不再继续没有配置 checkpointer 或没用同一个 thread_id配置 MemorySaver/SqliteSaver第二次调用复用同一 thread_id节点返回的所有字段都变得很怪在节点里返回了过多的键覆盖了别的节点写入的值每个节点只返回它要修改的键条件路由总是跳到默认分支路由函数返回值没有匹配映射表里的 key打印路由函数返回值检查 mapping 字典是否有对应项循环执行达到 recursion_limit 报错循环后缺少明确的终止条件在路由函数里增加“完成”判断指向 END 节点多次 invoke 后上下文错乱每个线程的 checkpoint 没有按业务维度隔离全局统一生成 thread_id用用户ID或任务ID做前缀5.2 调试技巧善用事件流与状态打印LangGraph 新手最常见的困惑是“图到底执行到哪一步了”。我建议直接使用invoke之外的stream方法它会逐步把每个节点的执行结果吐出来非常适合调试。for event in app.stream( {messages: [{role: user, content: 上海天气怎么样}]}, configconfig ): print(event)每次 event 是一个 dictkey 是节点名称value 是该节点的状态更新。这样你就可以像看剧情一样看整个工作流的每一步哪个节点跑了、返回了什么、状态里哪个字段被改成了什么全在眼前。如果你要排查的是状态字段的值建议在节点函数内部加一段临时的调试打印def some_node(state: dict) - dict: print(当前状态, state) # ... return {...}不要直接在代码里大量使用 print 造成环境污染但在开发期状态打印是最直观、最快速的定位手段。调试完成后统一删掉。另一个极其有用的技巧是单独验证条件路由函数。这些函数本质上就是输入一个 state返回一个字符串的普通函数完全可以脱离图单独测试test_state { intent: logistics, messages: [], order_id: 123, refund_approved: False, api_result: {}, final_reply: } print(route_after_intent(test_state)) # 应该是 logistics只要每个路由函数能单独测出预期结果图的流转下一步就到“连线是否正确”的问题排查范围会缩小很多。5.3 避坑指南状态合并、副作用与长会话避坑指南这部分是我最想写的内容因为我在 LangGraph 上翻过的车基本都集中在这几个点。第一千万不要在节点内部修改传入的 state 对象本身。LangGraph 的状态是不可变更新的理念节点应该返回一个新的 dict 来描述状态变更而不是直接执行state[intent] x。前者是声明式更新框架能追踪变更并用于检查点后者即使在简单场景下能跑一旦涉及中断恢复、历史回滚就会出现状态不一致的问题。我一开始图省事直接改 dict后面接上 checkpointer 之后就踩了大坑恢复的现场和实际状态对不上。第二副作用的放置位置要有讲究。发邮件、写数据库、调用第三方支付这类操作应该放在独立的节点里并且这个节点最好位于图的“末端链路”尽量避免被循环调用多次。如果你的工作流里存在条件路由或者循环尤其要注意一个节点会在一次完整流程里被调用多次的情况。解决办法是给“只执行一次”的节点设计一个状态标记节点开头判断标记是否已存在存在就直接返回空更新def one_time_node(state: State) - dict: if state.get(task_done): return {} # ... 执行真正的操作 return {task_done: True}第三长会话的数量级要提前规划。你的 messages 会随着多轮对话不断追加如果社交场景里的用户一天发几百条消息这个列表的体积会非常可观不仅影响模型响应速度还会让 checkpointer 的存储体积爆炸。我的经验是每完成一轮对话就对 messages 做一次“压缩”把超过一定轮数限制的历史消息交给模型生成摘要用一个短 summary 代替冗长的原始记录。这个做法在 LangChain 生态里也有现成组件支持即便不用组件自己写一个压缩节点也很简单把 summary 放进状态里的独立字段按需拼接进 prompt。第四checkpointer 的存储位置和清理策略要想清楚。开发阶段用 MemorySaver 很舒服但生产环境默认选 SqliteSaver 会相对简单可靠。需要注意的是线程数据会无限增长必须建立一个归档或者定期清理的任务否则数据库文件会变得越来越大。我的做法是超过 30 天未活跃的线程直接删除或者转存到冷存储这个策略要根据业务来定但至少要有。第五大模型的工具调用格式和状态设计是强耦合的。如果你让模型自主决定调用工具它返回的 structured output 要能映射到状态字段。我踩过的坑是让模型返回 JSON 格式但节点里忘了做异常捕获模型偶尔输出不合法 JSON 就把整个图跑崩了。解决方案是在关键节点外层包一层 try-except并且在状态里加一个 error 字段捕获异常后跳转到错误处理节点而不是让整个任务戛然而止。5.4 更进一步的思考StateGraph 可视化与运维最后一个话题聊聊运维层面的事。LangGraph 编译成的图是可以序列化的这意味着你可以把编译后的图对象保存起来或者部署到远程环境中。如果团队里有人画图、有人写节点可视化协作就很有必要。现阶段我推荐的方式是把图编译和节点实现分开两个模块workflow/ nodes.py # 所有节点函数 graph.py # 状态定义、路由函数、图的组装 main.py # 调用入口这样分工清晰新增节点的人只需要在 nodes.py 里加函数改流程的人只需要在 graph.py 里连边。节点函数做完单元测试后图的正确性就能通过代码审查保障。另外你完全可以把 LangGraph 的检查点机制和数据持久化打通做到每执行一个节点就落一次库。这样一来即使整个服务崩溃重启之后也能从最近的 checkpoint 恢复用户无感知。这个能力对于生产环境的价值非常大你可以把它当作一个“自带可恢复性的任务编排系统”来用。还有一点值得思考的是LangGraph 并不只适用于构建聊天机器人。凡是可以用图建模的 AI 流程都可以借鉴它的设计比如文档处理管线、数据分析流水线、多步骤内容生成工作流、自动化测试用例执行。它的状态机制和条件路由是通用的只要你愿意把业务逻辑拆成节点和边。我在实际项目中收获最大的一个体会是LangGraph 带给我的不只是“状态机”而是一种把复杂 AI 流程变成可控工程系统的思维模式。节点是模块边是路由状态是数据契约checkpointer 是持久化而 interrupt 是人与机器协作的接缝。这套抽象哪怕将来换成别的框架也依然是设计复杂 Agent 应用的最好视角之一。