写 Agent 流程时最容易撞上的怪事上一个节点刚写入的数据下一个节点读到时却只剩自己的内容日志里也查不出是谁把字段改没了。只盯着节点函数看很难定位因为状态合并发生在框架层节点返回的字典会被悄悄合并进全局状态一个字段究竟是被覆盖还是被追加取决于它背后的 reducer而这层逻辑在业务代码里完全看不见。下面分享一套可直接落地的做法先用TypedDict 规约状态 再用reducer 对照表定位合并语义 最后用条件边与 checkpoint 把流程跑通每一步都配可运行的最短代码。一、三要素分工State 存数据、Node 做加工把 LangGraph 想成一张会自己跑起来的流程图先认清三样东西各自的职责State整张图共享的数据包节点从这里读、往这里写字段在编译前就声明完毕。Node一个普通 Python 函数接收当前状态只返回一份局部更新字典不必回传全部字段。Edge连接节点的线段分普通边与条件边决定执行流下一步落在哪个节点。概念职责记忆口诀State保存跨节点数据数据的唯一事实源Node加工数据并返回增量只交回改过的字段Edge决定下一个执行者箭头指向谁就跑谁核心结论状态是共享黑板节点是贴便签的人边是指示下一位的箭头。二、状态规约先行用 TypedDict 约定字段动手写节点之前先把状态字段一次声明清楚TypedDict用标准类型注解描述字段图编译期就能发现拼写错误与类型不匹配。totalFalse表示字段允许逐步产生入口可以只传usernamemessages由节点在运行中补齐。默认即覆盖没有显式 reducer 的字段后来的值直接替换旧值列表同样逃不过这条规则。核心结论先规约再编码状态结构就是节点之间的契约。三、覆盖陷阱复现普通列表会被新值顶掉下面这段代码建一个最普通的 TypedDict 状态两个节点先后写messages运行后直接打印结果fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassCommonState(TypedDict,totalFalse):username:strmessages:list[str]defcommon_a(state:CommonState)-dict:return{messages:[你好我是 state[username]]}defcommon_b(state:CommonState)-dict:return{messages:[AI你好state[username]]}builderStateGraph(CommonState)builder.add_node(common_a,common_a)builder.add_node(common_b,common_b)builder.add_edge(START,common_a)builder.add_edge(common_a,common_b)builder.add_edge(common_b,END)graphbuilder.compile()print(graph.invoke({username:小黄})[messages])输出只剩一条[AI你好小黄]而username依然健在。被顶掉的只是 messagescommon_b返回的新列表整体替换了common_a的列表。未涉及的字段不受影响框架按字段粒度合并没返回的字段保持旧值。核心结论普通列表既不会追加也不会去重它只知道整体替换。四、换上 MessagesState消息自动追加保留把状态基类换成官方的MessagesState同样的两个节点就会得到完全不同的结果fromlanggraph.graph.messageimportMessagesStateclassChatState(MessagesState):username:strdefmessage_a(state:ChatState)-dict:return{messages:[(user,f你好我是{state[username]})]}defmessage_b(state:ChatState)-dict:return{messages:[(ai,fAI你好{state[username]})]}builderStateGraph(ChatState)builder.add_node(message_a,message_a)builder.add_node(message_b,message_b)builder.add_edge(START,message_a)builder.add_edge(message_a,message_b)builder.add_edge(message_b,END)graphbuilder.compile()resgraph.invoke({username:小黄})print([m.contentforminres[messages]])差别只在基类MessagesState内部给messages挂了add_messagesreducer新消息被追加成AnyMessage对象。自定义字段照常共存username没有 reducer依旧是覆盖语义两种规则互不干扰。核心结论想让消息累积就把字段交给带 reducer 的状态基类。五、reducer 对照表看清每类字段的合并语义遇到「字段莫名其妙丢了」先按下表判断它属于哪一类字段类型合并规则典型场景普通 str / dict新值直接覆盖旧值用户名、配置快照普通 list新列表整体替换中间结果、临时缓存messagesMessagesState按 id 追加或更新对话历史Annotated[list, operator.add]拼接两个列表步骤记录、工具调用流水覆盖型字段适合「只要最新值」的场景例如当前路由目标、重试次数。追加型字段适合「要留痕」的场景例如思考链、执行轨迹、审计日志。核心结论先给字段选对合并语义再写节点逻辑顺序反了必返工。六、自定义合并规则Annotated 接上 operator.add下面这段代码演示如何在 TypedDict 上手动挂 reducer让列表字段从覆盖变成追加importoperatorfromtypingimportAnnotated,TypedDictclassStepState(TypedDict):steps:Annotated[list[str],operator.add]# 追加合并answer:str# 普通覆盖defplan(state:StepState)-dict:return{steps:[规划拆解问题],answer:}defact(state:StepState)-dict:return{steps:[执行调用工具],answer:完成}builderStateGraph(StepState)builder.add_node(plan,plan)builder.add_node(act,act)builder.add_edge(__start__,plan)builder.add_edge(plan,act)graphbuilder.compile()print(graph.invoke({steps:[],answer:}))# {steps: [规划拆解问题, 执行调用工具], answer: 完成}Annotated 是挂载点list[str]仍描述类型第二个参数才是框架实际调用的合并函数。同字段可混用不同语义steps拼接、answer覆盖两种规则在同一状态里并行生效。核心结论自定义 reducer 只需一行 Annotated就能把覆盖型字段改造成留痕型字段。七、局部更新合并节点只交回自己改的部分理解框架如何把返回值写回状态就能预测每一步的最终结果节点返回增量 → 框架按字段查 reducer → 执行覆盖或追加 → 写回共享 State → 下游节点读到最新值可以少返回节点只需带回自己改动的字段其余字段原样保留这让节点函数保持单一职责。不能返回 None返回None或空字典以外的非法结构会直接报错返回None则等价于不更新。核心结论节点写的是差量合并的是框架读懂这条链路就不再怕字段丢失。八、条件边做路由让流程按状态走不同分支下面这段代码用条件边按消息条数决定下一步走搜索还是直接回答fromtypingimportLiteralfromlanggraph.graphimportStateGraph,START,ENDdefrouter(state:ChatState)-Literal[search,answer]:returnsearchiflen(state[messages])3elseanswerdefsearch(state:ChatState)-dict:return{messages:[(ai,检索结果……)]}defanswer(state:ChatState)-dict:return{messages:[(ai,最终回答……)]}builderStateGraph(ChatState)forname,fnin{router:router,search:search,answer:answer}.items():builder.add_node(name,fn)builder.add_edge(START,router)builder.add_conditional_edges(router,router,{search:search,answer:answer})builder.add_edge(search,END)builder.add_edge(answer,END)graphbuilder.compile()路由函数只读状态它不修改任何字段返回一个字符串作为分支键映射表把键翻译成节点名。分支必须收敛每条路径都要能走到END否则图会在运行期报「到达不了结束节点」。核心结论普通边定顺序条件边定策略两者合起来才是完整的控制流。九、并行与断点用 checkpoint 追踪状态变化当多个节点可能同时写状态时靠打印已经不够需要可回放的检查点fromlanggraph.checkpoint.memoryimportMemorySaver graphbuilder.compile(checkpointerMemorySaver())cfg{configurable:{thread_id:demo-1}}snap1graph.get_state(cfg)# 当前快照print(snap1.values.get(messages))# 看每一步之后的状态forsingraph.get_state_history(cfg):# 回放历史节点print(s.next,list(s.values.keys()))thread_id 隔离会话同一张图给不同用户各开一条线程状态互不串台。快照按节点粒度留存get_state_history能看到每个超步之后的字段值覆盖发生在哪一步一目了然。核心结论并行写入先用 checkpoint 定位再决定给字段换哪种 reducer。十、排错速查表常见状态异常的定位路径把高频问题整理成一张可以直接照着查的表现象可能原因处理动作字段只剩最后一条普通 list 被覆盖改用Annotated[..., operator.add]对话历史顺序错乱多节点并发写消息加条件边串行化或按 id 追加入口字段读不到未标totalFalse放宽必填或补全入口参数图跑不到 END条件分支缺边检查映射表是否覆盖所有返回值老数据莫名消失换了状态基类对照第五节表格重新选语义先看类型再看代码九成的丢数据问题出在字段合并语义选错而不是节点逻辑写错。再看图结构确认边是否齐全、分支键与映射表是否一致最后才去加日志。核心结论状态异常按「合并语义 → 图结构 → 节点逻辑」的顺序排查最快收敛。结语状态与节点是 LangGraph 最容易被低估的一对概念节点写起来只是普通函数真正决定程序行为的是每个字段背后的合并语义。把 TypedDict 规约放在最前面用对照表给字段选对 reducer再用条件边和 checkpoint 把控制流与可观测性补齐绝大多数「数据莫名丢失」的问题都能在编译或第一次运行时暴露。这套写法可以直接搬进自己的项目新流程先声明状态、再挑合并规则、最后接路由与断点团队协作时状态结构本身就是一份活文档。把合并语义想清楚节点代码自然就简单了。