046、LangGraph构建有状态Agent流程那天早上生产环境的告警把我从咖啡因的醉意里拽了出来。用户反馈说对话机器人偶尔会“失忆”——上一轮说过的话下一轮就忘得干干净净。我拉出日志一看LangGraph 的 state 里明明有messages可到了第二个节点历史消息就只剩一个 system prompt 了。我盯着终端里那条node_2 returned None的日志突然明白了我根本没把节点返回的状态 merge 回去而是让每个节点都返回了一个全新的 state 字典把之前的东西给覆盖了。这就是有状态 Agent 最初级的坑也是我今天想跟你聊的 LangGraph 最核心的设计——StateGraph。先说那个 bug。我当时的图长这样fromlanggraph.graphimportStateGraphdefnode_a(state):return{messages:state[messages][(human,hi)]}defnode_b(state):return{messages:state[messages][(ai,hello)]}builderStateGraph(...)# 大概这个意思看这代码好像每次返回都带上state[messages]逻辑对。但问题出在 StateGraph 的 state 合并机制上。默认情况下每个节点返回的 dict 会直接替换 state 里对应键的值而不是追加。如果某个节点漏写了state[messages]比如它只想改一个counter字段返回{counter: 1}那么整个 state 就变成只有counter了。旧的消息全没了。当时生产环境里那个节点是从别处复制来的忘了带上前面的状态字段于是历史消息就在每次调用时被悄悄清空。后来我改用add_reducer或者add_messages来修饰 state 的字段才把问题摁住。LangGraph 的 StateGraph 知道怎么合并状态不是靠我们手动拼字典而是要告诉它每个键的合并策略。比如messages字段用operator.add它就会把新消息追加到旧列表后面用Annotated[list, add_messages]它还会做消息去重和角色合并。但如果你啥也不声明默认就是“覆盖”——新值直接替换旧值。这是第一个大坑。LangGraph 用“有状态”这个词含义比我们在普通函数里说的全局变量要扎实得多。它本质是一个图执行引擎每个节点接收当前 state执行逻辑返回一部分 state 的更新。引擎负责把这部分更新合并进全局 state然后传递给下一个节点。这个过程是函数式的节点本身不允许修改外部 state只能通过返回值来声明变更。这样设计的好处是图可回放、可中断、可恢复每个节点都是纯函数状态是显式的数据流。坏处是如果你不理解合并规则就会像我一样被覆盖问题折腾一宿。有状态 Agent 的真正价值不在“记住变量”而在于跨节点、跨轮次的上下文保持。我后来把对话管理切成三段接收用户输入、调用 LLM、执行工具。每个节点都要访问同一个messages列表还要能写入新的消息。没有 state 的框架你得自己写一堆全局队列或数据库表有了 StateGraph你只需要在初始化 state 里放一个messages键然后让每个节点返回追加后的消息。LangGraph 会在图运行结束后把最终 state 挂在内存里。但内存是脆弱的一旦服务重启所有会话上下文全飞了。于是有了 Checkpointer。Checkpointer 是 LangGraph 里真正体现“有状态”实力的东西。它允许你把每次节点执行的 snapshot 持久化到磁盘、数据库或 Redis 里。我试过在本地用SqliteSaver一行代码挂上去图立刻具备了“断点续跑”和“时间旅行”能力。所谓时间旅行就是你可以在一次图执行完成后把 state 恢复到中间某个节点执行后的状态然后从那里岔开再跑。这对调试 Agent 特别有用。以前我调试 LLM 调用得把整个 prompt 打出来看是不是参数传错了。现在直接在 checkpoint 里找到那个节点执行前的 state改了输出重新从那里跑完全不用重放前面的网络请求。但 Checkpointer 不是银弹。我第一次用的时候遇到另一个恶心的问题state 里的某个字段是自定义类实例比如数据库连接对象或者 pandas DataFrame。Checkpointer 序列化它的时候直接炸了。LangGraph 默认用 JSON 序列化自定义类不实现to_json就报错。解决办法要么把这类对象排除在 checkpoint 之外用InMemorySaver只做内存快照要么给类加序列化方法。我现在的经验是state 里只放可序列化的数据连接这种东西放全局别塞进 state。再聊聊条件边。有状态 Agent 最典型的流程是先判断要不要调用工具再决定下一步走哪个节点。LangGraph 的add_conditional_edges让你根据 state 的内容动态选择下一跳。比如你有个route函数接收 state返回一个字符串表示去哪个节点。我当时写了一个工具调用判断因为用了state[messages][-1][tool_calls]在 LLM 没返回 tool_calls 时 KeyError导致整个图直接崩溃。后来我在route里加了个try-except或者用.get()才算稳定下来。条件边本身就是有状态决策它依赖 state 里的最新信息。如果你在之前某个节点忘记更新 state 的needs_tool字段那么条件边就会按旧值走逻辑就歪了。所以我的经验是条件边读的字段一定要在路径上显式地写过而且最好在注释里标明“这里必须由 XX 节点更新”。代码示例来一个实际点的。假设我们要做一个带工具调用的 Agent图结构如此fromtypingimportAnnotated,TypedDictfromlanggraph.graphimportStateGraph,ENDfromlanggraph.graph.messageimportadd_messagesclassAgentState(TypedDict):messages:Annotated[list,add_messages]current_tool:strretries:intdefcall_llm(state):# 这里我踩过坑直接用state[messages]传给LLM结果把历史消息全塞进去了token爆掉# 正确做法是只取最近几轮或者做摘要压缩truncatedstate[messages][-4:]responsellm.invoke(truncated)return{messages:[response]}defcall_tool(state):tool_namestate[current_tool]resultexecute_tool(tool_name,state[messages][-1].content)return{messages:[{role:tool,content:result}]}defdecide(state):# 别这样写state[messages][-1].tool_calls 直接索引可能KeyErrorlaststate[messages][-1]ifhasattr(last,tool_calls)andlast.tool_calls:returncall_tool# 去执行工具returnEND# 结束graphStateGraph(AgentState)graph.add_node(llm,call_llm)graph.add_node(tool,call_tool)graph.add_edge(llm,tool,conditiondecide)# 条件边graph.add_edge(tool,llm)# 工具执行完再回LLMgraph.set_entry_point(llm)这里有个细节add_edge带 condition 参数和add_conditional_edges其实是等价的。我更喜欢前者少写一点。但如果你有多个分支用add_conditional_edges映射字典更清晰。再说add_messages这个 reducer它不只是简单的追加列表还会根据消息 ID 做合并。同一 ID 的消息如果内容更新它会替换旧消息而不是重复添加。这个特性在重放 checkpoint 或者重新生成 LLM 响应时特别有用不会让你的消息列表里出现两条同一角色同一内容的残影。聊到有状态不得不提END节点。每次图执行完state 就冻结在最后一次 checkpoint。下次执行同一个图对象时你可以选择从零开始清空 state 重跑或者继续上一次的 state接续会话。LangGraph 的invoke方法支持传入一个config里面带上thread_idCheckpointer 就能根据这个 thread_id 找到对应会话的 state。这就是多轮对话的根基。我之前的项目里每个用户 ID 对应一个 thread_id服务端一启动就把所有会话的 state 都放在一个 Sqlite 文件里用户说话直接graph.invoke({messages: [new_msg]}, config{configurable: {thread_id: user_id}})简单粗暴且有效。但注意如果你的 Agent 有状态字段随业务变化比如retries计数器你得想清楚它是应该跟着会话走还是每次执行都重置。我一般把这种临时变量放在 state 里但在图的入口节点加一个“清洗”步骤每次新用户输入时把临时字段清掉避免上次遗留的状态干扰本次决策。另一个经验是不要在节点里直接修改传入的 state 字典。虽然 Python 字典是引用传递你改了 state[“messages”] 就等于改了全局 state但 LangGraph 的并行执行或 checkpoint 机制可能因此出现诡异行为。我试过为了省事直接在节点里执行state[messages].append(...)结果有一次图里两个节点同时执行并行分支互相覆盖了对方 append 的内容。LangGraph 文档里明确说了节点应该返回部分 state 的更新而不是原地修改。尽管实际运行时很多例子是原地修改也work但这属于“侥幸”不是“正确”。我后来强制自己写纯返回值任何对 state 的变更都通过 return 表达调试时反而更清楚。说到并行执行StateGraph 的add_node也是支持异步的你可以定义 async 函数作为节点。有状态在这种场景下更关键——多个并行分支都读取同一个 state但写入时合并策略决定了最终结果。比如一个分支读counter加了 1另一个分支加了 2如果counter没有 reducer后面执行的那个分支的结果会覆盖前面的。要得到 3你得给counter配上自定义 reducer。我踩过这个坑当时用operator.add对整数做合并结果发现它把新旧值相加而不是覆盖正好符合需求。LangGraph 的 reducer 本质是“如何把新的返回值合并到旧 state 里”的函数你可以自己写任意逻辑。现在回头看构建有状态 Agent 流程核心其实就两件事一是定义好 state 的形状和每个字段的合并规则二是选好 Checkpointer 的持久化策略。这两件事搞定了图本身反而变得简单节点就是普通函数边就是决策逻辑。我见过很多新手把逻辑全部塞进一个节点用一个大 while 循环模拟图美其名曰“简单”。但一旦需要重试、断点续跑或人工介入这种单体结构就寸步难行。LangGraph 的图结构天生的可恢复性才是它值得用的原因。比如你可以设置interrupt_before[tool]图跑到工具节点前自动暂停把控制权还给人类审批。审批通过后从暂停处继续跑state 完全保留。这在生产环境里做人工复核非常有用。我最后想给的实操建议有三条。第一条定义 state 的 TypedDict 时把每个字段的合并规则写在旁边注释里比如messages: Annotated[list, add_messages] # 追加别覆盖。过一个月你再读代码你会感谢自己的注释。第二条Checkpointer 早点接上不要先跑通再挂。从第一天起就把 thread_id 设计进 API 签名否则后面加会话隔离会改得你怀疑人生。第三条写一个调试用的可视化函数把每次节点执行后的 state 快照打出来或者用 LangGraph 自带的 debug 模式。我看到很多同事用print(state)打出一大串混乱的东西毫无重点。我习惯只打印关键字段的变更比如print(f[{node}] msgs{len(state[messages])}, retries{state[retries]})。这样能在不污染日志的情况下快速定位问题。有状态的 Agent 流程本质上是在管理时间维度的数据。LangGraph 把状态变成了一等公民我们这些工程师要做的不是把代码写得花哨而是想清楚状态从哪里来、到哪里去、怎么合并、怎么持久化。那个生产事故教会我的不是加一个 reducer而是意识到“有状态”这个形容词的分量——它意味着责任。你写下的每一个 state 字段都会在图的某个角落影响流程行为。今天偷懒用覆盖明天出 bug 就只能熬夜。希望这篇笔记能给你省几个通宵就这样。
