LangGraph05：状态管理

LangGraph 状态管理源码拆解TypedDict 还是 PydanticStateManager 三板斧的不可变设计前面四篇都在用 LangGraph 写功能。从本篇开始下沉一层状态State为什么是 LangGraph 的根基、两种状态定义方式怎么选以及状态更新三板斧update/assign/delete背后的不可变设计——这是整个框架安全可回滚的源代码级答案。一、状态失控所有 Agent 框架的共病传统的链式调用里数据只沿单向前进A 的输出喂给 BB 的输出喂给 C传完即弃。C 想回头看看 A 干了什么做不到——A 的数据在 B 那里就已经被加工过、丢弃了。这带来两个系统性风险状态漂移同一个字段被谁改过、改成什么没人说得清副作用失控一个节点顺手改了个全局变量千里之外的另一个节点行为就变了排查全靠运气。LangGraph 的解法是把状态摆到明面上所有节点都从一个显式声明的状态对象里读、往里面写。一份状态贯穿全图每个节点的输入输出都围绕它展开。这样做的三个直接收益可测试任意中间状态都能构造出来喂给单个节点、可追溯每一步的状态变化有记录、可扩展新节点只能动声明过的字段破坏面可控。二、两种定义方式TypedDict 与 PydanticLangGraph 允许用两种方式声明状态结构fromtypingimportTypedDict,AnnotatedfrompydanticimportBaseModel# 方式一TypedDict轻量classMyState(TypedDict):user_input:strsteps:Annotated[list,append]result:str# 方式二Pydantic 模型带校验classMyStateModel(BaseModel):user_input:strsteps:listresult:strgraph_builderStateGraph(MyStateModel)# 初始化图时指定怎么选看一个对比表维度TypedDictPydantic 模型运行时校验无只是类型注解有实例化即校验非法数据直接报错默认值与验证器无支持默认值、field_validator自定义规则序列化无内置能力model_dump()/model_validate()一键互转额外字段处理无约束默认拒绝未声明字段防脏数据流入开销与复杂度极低几乎无感略高多一个依赖适用阶段快速原型、小团队小项目生产环境、需要数据契约的场合实践建议原型期用 TypedDict 快速迭代进入生产前切到 Pydantic。关键不是选哪个而是所有节点共享同一份状态契约——字段名、类型、语义全图统一杜绝我说的和同事理解的不是一回事。三、Annotated给字段挂上合并语义状态里最精妙的设计是字段级合并规则通过Annotated标注classState(TypedDict):messages:Annotated[list,append]# 列表追加而不是覆盖count:Annotated[int,sum]# 数值累加config:dict# 普通字段覆盖写入含义是节点返回{messages: [新消息]}时框架按字段上挂的规则合并——messages是追加进旧列表count是加在旧值上config才直接替换。没有这个机制每个节点都得手写读旧值 → 拼接 → 写回样板代码成灾还容易在并发下丢数据。四、StateManager 三板斧update / assign / delete状态更新的入口被收敛为三个方法统一由StateManager提供。先看简化后的源码classStateManager:defupdate(self,current_state,new_values):合并更新新值覆盖旧值其他字段保留ifnotnew_values:returncurrent_state updatedcurrent_state.copy()# ① 先复制原状态不动forkey,valueinnew_values.items():ifkeyinself._reserved_fields:# ② 预留字段禁止业务修改raiseValueError(fField{key}is reserved)ifisinstance(value,dict)andkeyinupdated:updated[key]self._deep_merge(updated[key],value)# ③ 字典深度合并else:updated[key]value# ④ 普通字段直接覆盖returnupdateddefassign(self,current_state,updates):赋值update 的语义别名returnself.update(current_state,updates)defdelete(self,current_state,keys):删除清理不再需要的字段updatedcurrent_state.copy()forkeyinkeys:ifkeyinupdated:# 字段存在才删避免 KeyErrordelupdated[key]returnupdatedupdate核心中的核心update的精髓在深度合并。假设状态里有一个嵌套字典字段# 原状态current_state{query:查询订单服务 CPU,tool_calls:{prometheus:{service:order-service,time_range:1h}}}# 只想改 time_range、再加一个指标new_values{tool_calls:{prometheus:{time_range:2h,metrics:cpu_usage}}}# update 之后# {prometheus: {service: order-service, time_range: 2h, metrics: cpu_usage}}如果简单覆盖原有的service字段就丢了深度合并只动你指定的子字段其余原样保留——精细化更新的关键一步。是否是否是否传入 new_values是否为空?直接返回原状态copy 原状态为 updated字段是预留字段?抛 ValueError新值是 dict且原状态已有该字段?深度合并到 updated直接覆盖 updated返回新状态assign语义别名assign内部就是return self.update(...)功能完全一致。存在它的理由是可读性给普通字段赋新值用assign“我就是在设置这个值”修改嵌套结构用update“我在合并更新这部分”。同一操作、两种语义表达让代码意图一目了然。delete安全清理delete负责状态瘦身临时字段工具原始返回、中间计算值用完即删避免状态越滚越大、存档越来越肥。实现上先判存在再删除杜绝KeyError。三个方法的取舍方法核心行为何时用update字典深度合并普通字段覆盖增量更新、修改嵌套结构默认首选assign与 update 相同普通字段赋值追求语义清晰delete删除字段返回新副本清理临时字段、状态瘦身五、不可变设计为什么总是返回新对象三个方法有一个共同纪律永远不修改传入的状态而是复制一份、在副本上改、返回新对象。这是分布式和并发场景下的数据安全基石。它带来三层价值可回滚每一步的状态都是独立快照出错可以精确退回到上一步并发安全多个执行路径同时读写时不会互相踩踏同一份可变对象可预测节点拿到的是稳定的输入执行中途不会被别的代码偷偷改掉。这个设计有一个直接的编程要求——必须接收返回值# 错误改了空气原状态纹丝不动state_manager.update(state,{final_answer:CPU 使用率 80%})# 正确用新对象承接结果statestate_manager.update(state,{final_answer:CPU 使用率 80%})不接返回值是新手最高频的我明明改了为什么没生效的元凶。小结本篇的三个要点显式状态是 LangGraph 对抗状态漂移与副作用失控的根本手段TypedDict 原型、Pydantic 生产——Annotated的合并语义是状态设计的点睛之笔update / assign / delete 不可变设计——状态修改永远复制后改、返回新对象这是安全更新与可回滚的源代码级保证。状态怎么改才安全解决了下一篇解决改好的状态怎么存才不丢检查点Checkpointer的三种存储后端怎么选。