LangGraph 这个库这两年讨论度很高但我在社区里看过不少人的练习路径有个共同问题一上来就追着多代理架构跑结果图是搭出来了State 却改得乱七八糟一加分支就崩。我自己练 LangGraph 的顺序和别人不太一样最先啃的就是状态代码——State 的定义、节点返回值怎么合并、Reducer 怎么控制更新、条件分支里状态怎么传。这套练扎实之后再看那些多节点、带分支、带记忆的流程基本就通了。这篇文章就把我完整的状态代码练习过程记录下来环境基于我常用的较新版 LangGraph适合刚接触这个库、以及已经会搭图但对状态更新机制一知半解的同学。1. 先把状态模型这层窗户纸捅破1.1 状态到底是什么、为什么值得专门练LangGraph 的核心是一个状态图节点Node负责干活边Edge决定顺序而贯穿所有节点之间、让每个节点都能读到上下文的东西就是 State。State 本身就是一个字典结构的对象你用 TypedDict 或 dataclass 提前定义好它长什么样然后每次节点执行完后把“想改的字段”做成一个字典返回图的运行时负责把这些改动合并回总状态里。我说的“练状态代码”练的就是这三件事第一怎么定义一份合理的状态结构第二节点返回什么样的数据、以什么方式合并进总状态第三在分支、循环、并行节点里状态怎么传递不丢不串。这三件事任何一个没搞明白图都能跑但跑出来的结果就是不对而且特别难排查。为什么说这个值得单独练因为很多传统开发者的直觉是“函数输入输出”。一个函数处理完数据把结果返回给下一个函数参数一颗颗传下去。这套思路在写普通后端接口时没问题但一旦涉及多步骤的智能体流程——比如先规划、再检索、再生成、中途还要根据结果决定走哪条分支——函数式的传参会让代码变成一团乱麻。State 本质上是一块“公共白板”所有节点都能往上写、都能往里读你不需要在调用时手动把每一步的临时变量传来传去。这个模型想通了LangGraph 就算入门了。1.2 从“接力棒”到“公共白板”的思维切换先看一个典型的旧思路代码def step1(raw_text: str) - str: cleaned raw_text.strip().lower() return cleaned def step2(cleaned: str) - list[str]: tokens cleaned.split() return tokens result step2(step1( Hello LangGraph ))每一步函数都只接收上一个函数的返回值如果你想在最后一步同时拿到 raw_text、cleaned、tokens就得用元组或者类把这些都包着往下传。代码一多参数列表就失控了。换成 LangGraph 的状态模型写法变成了这样from typing import TypedDict from langgraph.graph import StateGraph, START, END class MyState(TypedDict): raw_text: str cleaned: str tokens: list[str] def step1(state: MyState) - dict: return {cleaned: state[raw_text].strip().lower()} def step2(state: MyState) - dict: return {tokens: state[cleaned].split()} builder StateGraph(MyState) builder.add_node(step1, step1) builder.add_node(step2, step2) builder.add_edge(START, step1) builder.add_edge(step1, step2) builder.add_edge(step2, END) graph builder.compile() print(graph.invoke({raw_text: Hello LangGraph }))注意 step2 里读的state[cleaned]并不是 step1 直接传给你的参数而是 step1 写好、图运行时合并进 State 之后的公共数据。这就像一个团队共用的项目白板每个人干完活把自己负责的那部分更新上去下一个同事需要什么直接看白板不需要等某个人把结果递到手里。你可能觉得这个例子太简单看不出差别。但你要是在这个图里再加一个“日志收集节点”、加一个“异常处理分支”State 的优势马上就出来了——日志往logs字段里累积分支判断读state[status]每个节点只需要关心自己读写的那几个字段不要碰自己不关心的东西。这就是状态驱动设计的核心思想。2. 第一个状态练习最小可运行图2.1 把环境准备好LangGraph 是纯 Python 库安装很简单pip install langgraph建议 Python 版本 3.9 以上另外如果你计划跑带持久化的例子还需要确认langgraph-checkpoint已经作为依赖装好。我练习时习惯在普通 Python 脚本里写因为脚本跑完能直接看到 print 输出比在交互式环境里更容易保留完整记录。注意LangGraph 的 API 在 0.1 到 0.2 之间有过调整比较明显的变化是入口图用START而不是__start__。我这篇文章的代码基于 0.2.x 及以上版本如果你还在用很老的版本建议先升级到最新稳定版再照着练。2.2 用 TypedDict 把状态结构先钉死状态结构是一张图的“数据库表结构”先想清楚再写节点。我第一次练习时直接上来写节点函数结果反复改字段名图一跑就 KeyError。后来我学乖了任何练习的第一步永远是定义状态。from typing import TypedDict class CounterState(TypedDict): count: int logs: list[str]这个状态有两个字段count用来记录数值logs用来记录每一步的处理日志。就这么简单但注意一个细节字段的类型要写具体int、list[str]都明确标注因为后面写节点函数时 IDE 会帮你做提示字段名拼错时也能更快发现。为什么用 TypedDict 而不是普通字典因为普通字典太自由了节点里写state[cnt]和state[count]都不会报错运行时才会炸TypedDict 至少能在编辑器里给你一层提示和约束。LangGraph 本身也支持 pydantic 模型或 dataclass 作为状态定义但初学者从 TypedDict 起步最合适它足够轻量心智负担最小。2.3 用三个节点跑通完整链路定义好状态之后我加三个节点一个加一、一个翻倍、一个收尾记录。重点看每个节点怎么返回“更新补丁”。from typing import TypedDict from langgraph.graph import StateGraph, START, END class CounterState(TypedDict): count: int logs: list[str] def add_one(state: CounterState) - dict: new_count state[count] 1 return {count: new_count, logs: state[logs] [fadd_one - {new_count}]} def double(state: CounterState) - dict: new_count state[count] * 2 return {count: new_count, logs: state[logs] [fdouble - {new_count}]} def final_note(state: CounterState) - dict: return {logs: state[logs] [ffinal count is {state[count]}]} builder StateGraph(CounterState) builder.add_node(add_one, add_one) builder.add_node(double, double) builder.add_node(final, final_note) builder.add_edge(START, add_one) builder.add_edge(add_one, double) builder.add_edge(double, final) builder.add_edge(final, END) graph builder.compile() result graph.invoke({count: 1, logs: []}) print(result)运行结果{count: 4, logs: [add_one - 2, double - 4, final count is 4]}我建议你亲手跑一遍这个例子重点观察两件事。第一invoke的入参是“初始状态”返回值是“最终状态”中间每个节点产生的新状态你是看不见的除非你在节点里打印。第二节点返回的字典不需要包含所有字段比如double节点完全可以不返回logs原样保留也可以像我这样手动拼接后返回。这里有个初学者最容易踩的坑以为节点返回什么状态就“变成”什么实际上对于没有配置特殊合并逻辑的字段返回的 value 会直接覆盖旧值。2.4 在节点里埋点观察状态流转光看最终结果你可能还是不知道状态到底怎么变的。我的练习习惯是在每个节点开头打印当前状态def double(state: CounterState) - dict: print(f[double] 进入时状态: {state}) new_count state[count] * 2 return {count: new_count, logs: state[logs] [fdouble - {new_count}]}这样跑一次就能看到每条边触发时状态里已经累积了什么、当前节点准备改什么。等你以后排查“哪个节点把状态改坏了”时这个习惯能帮你省下一大半时间。提示compile()之后得到的是一个可调用的编译对象。你可以反复调用invoke每次传入新的初始状态互不干扰。这跟“同一个图的多次运行之间默认没有记忆”这一特性有关后面讲 Checkpointer 时再展开。3. 状态更新的核心Reducer 机制一定要亲手试3.1 默认覆盖行为先知道“什么都不写”时是什么效果在 LangGraph 里如果你在状态定义里对某个字段没做任何标注那它的默认更新方式就是“整体覆盖”。两个节点不管谁先谁后只要都返回同一个字段名后执行的节点就会把前一个节点的值整个替换掉。这个行为本身没问题count这种标量字段用覆盖就很合理——最后一步的值就是我想要的值。但一旦字段是列表、字典这种“需要累积”的数据覆盖行为就会让你抓狂。比如上面例子里如果我不手动把state[logs] [新日志]拼好再返回而是直接返回{logs: [新日志一]}那跑完后logs里只剩最后一个节点的记录前面的全丢了。所以你需要理解 Reducer它决定了“多个更新同时作用到同一个字段上时怎么合并”。没有 Reducer就是后写覆盖有 Reducer就能做到追加、拼接、去重等自定义合并。3.2 用 add_messages 维护对话消息列表聊到 LangGraph 的红区最常用的就是add_messages。它原本是给聊天消息设计的 Reducer但理解它的行为能帮你更好地理解所有 Reducer 的运作方式。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] intent: str def user_turn(state: ChatState) - dict: return {messages: [{role: user, content: 我想退款}]} def assistant_turn(state: ChatState) - dict: return {messages: [{role: assistant, content: 好的我来帮您处理}]}这里的关键是Annotated[list, add_messages]。Annotated是 Python 自带的类型标注工具LangGraph 会读取它的第二个参数作为这个字段的 Reducer。add_messages的作用是如果新旧值都是消息列表就把它们合并如果存在带id的消息对象还能按id去重更新。你不需要手动把历史消息拼进返回值只要返回“这一次新增的消息”运行时就会自动追加到状态里的messages上。这就是正确用法节点只返回增量合并交给 Reducer。3.3 自定义 Reducer写一个属于自己的合并规则add_messages 好用但它解决不了所有问题。比如你希望某个字段做“取最大值”“数值累加”“字典合并”那就得自己写 Reducer。Reducer 的签名很固定接收两个参数第一个是当前状态里已有的值第二个是节点返回的更新值返回合并后的新值。from typing import TypedDict, Annotated, Annotated def sum_reducer(current: int, update: int) - int: return current update class ScoreState(TypedDict): total: Annotated[int, sum_reducer] name: str def node_a(state: ScoreState) - dict: return {total: 10} def node_b(state: ScoreState) - dict: return {total: 5}这两个节点无论怎么排序执行最终total都是 15 而不是“后写的 5”。因为每个节点返回的数值都会通过sum_reducer与当前值合并顺序是先加 10 再加 5或者反过来。这里有一个特别重要的细节Reducer 是一对一合并的也就是说它每次只处理“一个更新值”和“当前累积值”。所以在并行分支场景下你不能在节点里返回一个“已经包含旧值的完整快照”否则会把已有内容再算一遍导致重复累加。正确的做法永远是返回增量让 Reducer 去做累积。# 错误示范节点里自己拼了历史 def wrong_node(state: ScoreState) - dict: return {total: state[total] 100} # 如果配了 sum_reducer这里最终会多加一次 # 正确示范只返回增量 def right_node(state: ScoreState) - dict: return {total: 100} # sum_reducer 自己处理累加我把这个坑单独拿出来讲是因为它在真实项目里特别常见。凡是配了 Reducer 的字段节点里的返回值含义就变了不再是“最终值”而是“这次贡献的更新量”。这个思维转变是练状态代码最核心的一关。4. 状态驱动的条件分支与多路汇合4.1 条件边的路由约定LangGraph 最吸引人的能力就是条件分支根据当前状态的内容决定下一步走哪个节点。实现方式是用add_conditional_edges挂一个路由函数。路由函数接收当前状态返回一个节点名字符串或者返回一个能被映射表翻译的 key。我推荐直接返回节点名字符串少一层映射少踩一个坑。def route(state: ChatState) - str: if state[intent] refund: return refund if state[intent] tech: return tech return human然后这样挂边builder.add_conditional_edges(classify, route)运行时会在classify节点执行完毕后根据返回的字符串决定下一步进哪个节点。返回的名字必须在图里已经注册过否则会报错说找不到节点。4.2 一个带分支的完整练习意图路由客服系统把前面的内容串起来我练过一个完整的小例子模拟一个客服入口按意图分流到不同处理节点from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class SupportState(TypedDict): messages: Annotated[list, add_messages] intent: str handled: bool def classify(state: SupportState) - dict: text state[messages][-1][content] if 退 in text: intent refund elif 卡 in text or 慢 in text: intent tech else: intent human return {intent: intent} def refund_handler(state: SupportState) - dict: return {messages: [{role: assistant, content: 退款流程已启动}], handled: True} def tech_handler(state: SupportState) - dict: return {messages: [{role: assistant, content: 已创建技术支持工单}], handled: True} def human_handler(state: SupportState) - dict: return {messages: [{role: assistant, content: 转接人工客服}], handled: True} def route(state: SupportState) - str: return state[intent] # 直接返回节点名 builder StateGraph(SupportState) builder.add_node(classify, classify) builder.add_node(refund, refund_handler) builder.add_node(tech, tech_handler) builder.add_node(human, human_handler) builder.add_edge(START, classify) builder.add_conditional_edges(classify, route) builder.add_edge(refund, END) builder.add_edge(tech, END) builder.add_edge(human, END) graph builder.compile() result graph.invoke({messages: [{role: user, content: 我想退款}], intent: , handled: False}) print(result)注意这里route直接返回state[intent]而intent是classify节点写进状态里的。所以路由的决策完全依赖状态里的字段这就是“状态驱动”四个字的含义。我建议你把输入改成“网络很卡”和“我要投诉”各跑一遍观察它走的分支完全不同但代码里没有写死任何顺序。4.3 分支汇合时的状态同步细节很多时候分支结束后还要汇合到一个公共节点继续处理。比如客服所有分支最后都要进入一个“记录日志”节点。汇合本身不复杂只要在每条分支末尾都连到同一个节点就行builder.add_edge(refund, record_log) builder.add_edge(tech, record_log) builder.add_edge(human, record_log)但这里有个非常隐蔽的状态问题如果多个分支是并行执行的比如用add_edge让两个节点同时指向同一个下游它们会同时写同一个状态字段。对于配了 Reducer 的字段比如messages更新会按顺序合并不会丢但对于没配 Reducer 的字段比如普通int或者str后写完的覆盖先写完的最终结果取决于执行顺序而这个顺序在并行场景里是不确定的。所以我的经验法则是并行分支里尽量让每个分支写自己的独立字段refund_result、tech_result不要在多个分支里改同一个无 Reducer 的字段。如果确实要汇总就定义一个results字段配上 Reducer让各方把结果塞进去再统一处理。这个设计思路我在真实项目里用得很频繁能避免掉 80% 的“状态被莫名覆盖”问题。5. 状态练习中的常见坑与排查实录5.1 状态没更新先按这三个方向查我在练习状态代码时最常碰到的问题就是“我明明 return 了为什么状态没变”。排查顺序基本固定第一检查返回字典的 key 拼写有没有跟状态定义里完全一致包括大小写和下划线。TypedDict 在编辑器里会提示但你如果用动态字符串构造字典提示就失效了必须肉眼核对。第二检查节点函数是否真的执行了 return。很多人写节点时先写了一段打印逻辑最后忘了 return函数默认返回 NoneLangGraph 会把它当成“没有更新”状态自然原封不动。第三检查这个字段是否配了 Reducer。如果配了累加型 Reducer而你想“设置新值”返回单个值的结果可能跟你预期不一样。比如total配了sum_reducer某个节点想重置为 0简单返回{total: 0}并不会变成 0而是继续累加。排查技巧也很直接在每个节点开头打印当前状态、return 前打印准备返回的更新值跑一遍输出就能定位是哪一层的问题。别靠猜状态图一旦节点多了靠猜是猜不出来的。5.2 Reducer 的“双刃剑”效应给字段加 Reducer 能解决累积问题但也会带来意想不到的副作用。最典型的是你在开发初期给messages加上了add_messages后期某个节点想“清空消息列表重新开始”结果发现清不掉因为任何返回的列表都会跟现有列表合并。解决这类问题我常用的办法有三个一是换一个字段名来承载“重置后的新列表”让旧列表留在状态里但不再被读取二是在 Reducer 里加入特殊标记值比如检测到更新值是某个特定字符串时直接替换而不是合并三是干脆把“重置”也做成一个明确的业务节点它直接返回一个全新的消息列表并且你通过设计让它的上游节点不再往同一个字段写增量。这里面没有银弹关键是想清楚一旦某个字段配置了 Reducer全图所有节点对这个字段的更新语义都已经变了写节点时必须时刻记住这一点。我在团队里跟人协作时经常在状态定义旁边写清楚每个字段的合并语义避免别人把普通字段当成覆盖字段来写。5.3 TypedDict 的类型边界与运行时校验TypedDict 虽然在定义阶段给了你类型提示但它的约束主要停留在编辑器层面。LangGraph 在运行时对状态的校验深度有限尤其是嵌套结构比如state[profile]是一个 dict里面少了一个 key 时TypedDict 不一定能拦住你等到节点里访问时才报 KeyError。我的建议是状态结构尽量保持扁平。字段就是一层每个字段的值要么是标量要么是列表要么是明确类型的数据对象。如果确实有复杂的结构化数据用 dataclass 或 pydantic 模型作为字段类型在节点边界做显式校验。不要试图在状态里塞三层以上的嵌套字典那会让整个图的可排查性变得非常差。这里还有一个相关细节LangGraph 对“节点返回了状态定义里不存在的 key”这件事在不同版本里处理方式不完全一致。我建议永远不要依赖旧版本的宽松行为返回的 key 必须都定义在状态结构里否则直接视为实现错误尽早暴露。5.4 Checkpointer 相关thread_id 不配就白改状态练习做到后面你一定会碰到持久化和多轮对话的需求。LangGraph 的记忆机制依赖 Checkpointer当你把图编译成带记忆的版本后调用方式会发生变化from langgraph.checkpoint.memory import MemorySaver saver MemorySaver() graph builder.compile(checkpointersaver) config {configurable: {thread_id: thread-1}} graph.invoke({messages: [{role: user, content: 第一轮}]}, config) graph.invoke({messages: [{role: user, content: 第二轮}]}, config)没有传config的话每次invoke仍然是从干净的初始状态开始跑之前写入状态的内容不会被保留。我见过不少人配了 Checkpointer 却忘了传thread_id然后来问“为什么我的图没有记忆”问题就出在这。另外同一个thread_id的多次调用之间状态是累积的。你想在调试时重新开始就换一个新的thread_id或者用graph.get_state(config)查看当前状态、用graph.update_state(config, {...})手动修改某一步的状态。这些 API 都是练状态代码时迟早要碰的我建议你在基础练习通过后专门花半天时间只练 Checkpointer 相关的操作。6. 我的练习心法与最后几条建议6.1 按什么顺序练效率最高如果让我给一个建议的练习顺序我会把它分成四步第一步只练三个节点的直线图把invoke的输入输出彻底搞清楚重点理解“节点返回的是更新补丁”这件事第二步练习各种 Reducer自己写一个累加器、一个add_messages的聊天累积、一个取最大值的合并函数把Annotated的用法吃透第三步练条件分支从最简单的二分路由开始慢慢加分支、加汇合第四步练带 Checkpointer 的多轮对话理解thread_id和状态持久化的关系。每完成一步就写一个独立的小脚本保存下来脚本里加几个断言判断输出的状态是否符合预期。比如跑完CounterState后断言result[count] 4。这样你后面改动状态结构时一跑就能知道有没有破坏旧行为非常值得。6.2 最值得带进真实项目的一个习惯最后分享一个我个人非常受用的习惯每个状态字段的定义旁边用注释写清楚它的更新语义。比如class OrderState(TypedDict): step: str # 覆盖后写为准 logs: Annotated[list, operator.add] # 累加各节点只追加增量 messages: Annotated[list, add_messages] # 聊天消息自动合并这短短几行注释能避免掉大量的“跟我预期不一样”的 bug。状态代码本来就是 LangGraph 项目里最容易被忽视、却最影响全局稳定性的部分前期多花一个下午把状态模型吃透后面搭真正的多代理应用时你会感谢自己当初没有跳过这一步。