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": [f"add_one: {state['total']} -> {new_total}"]} def double(state: SimpleState) -> dict: new_total = state["total"] * 2 return {"total": new_total, "logs": [f"double: {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 判断是否需要调用工具，如果需要则执行工具查询，否则直接回复。
我先定义一个带工具查询的模拟工具，比如查天气。为了演示清晰，这里用一个假函数代替真实 API：
def 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(model="gpt-4o-mini", temperature=0) 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(checkpointer=memory_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 元给用户"}, config=thread_config) print(result) # 只会执行到 interrupt 为止，不会继续 # 用户点“批准”后，继续执行 result2 = app.invoke(Command(resume="yes"), config=thread_config) print(result2["final_result"])这里有个非常重要的经验教训：thread_id是唯一的任务标识，两次调用必须使用同一个 thread_id，图才知道要从哪个中断点恢复。如果第二次调用换了 thread_id，LangGraph 会认为这是一个全新任务，重新走一遍流程。这个坑我掉进去过不止一次。
为什么 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. 实战：跨平台自动化客服 Agent
4.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(model="gpt-4o-mini", temperature=0)定义状态。这个状态比前面几个例子复杂一些，因为要跨节点传递订单信息、审批结果、回复内容：
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(checkpointer=checkpointer)编译这一步没什么额外开销，但它会帮你检查图结构是否合法，比如有没有节点没有连接、有没有多边冲突。编译出错总比执行到一半出错好。
测试一下物流查询流程：
config = {"configurable": {"thread_id": "cs-1001"}} result = app.invoke({ "messages": [{"role": "user", "content": "帮我查一下订单 100 号的物流"}] }, config=config) print(result["final_reply"])退换货流程测试：
config2 = {"configurable": {"thread_id": "cs-1002"}} # 第一次调用，停在审批节点 app.invoke({ "messages": [{"role": "user", "content": "我要退款，订单号 200 号"}] }, config=config2) # 模拟人工点击“同意” final = app.invoke(Command(resume="yes"), config=config2) 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": "上海天气怎么样？"}]}, config=config ): print(event)每次 event 是一个 dict，key 是节点名称，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 应用的最好视角之一。
