看到“状态代码”这四个字,先别误会成国家代码。在 LangGraph 里聊状态代码,聊的是围绕state本身写出来的那部分代码:状态字段怎么定义、节点怎么更新状态、条件边怎么用状态做路由、检查点怎么把状态存下来。我给自己定的练习主题,就是把 LangGraph 的状态机制一条条拆开,全部用最小可运行的代码跑一遍。这篇文章就是这一轮练习的记录,适合刚开始接触图状态、被“状态到底在哪更新”这个问题卡住的人。

先说为什么要单独练状态。LangGraph 里图结构其实很直白,节点就是普通函数,边就是箭头。可一旦跑起来,真正决定行为的是每个节点传进传出的状态:节点能不能继续走、走到哪条分支、失败了能不能重试、应用重启后能不能恢复,全部依赖状态。如果只在脑子里想“状态是一个字典”,写代码时一定会踩坑。所以我建议,一开始不写复杂业务,就写状态代码,把状态的更新、流转、持久化全部练熟,再回去写真正的 Agent 流程,会轻松很多。

1. 开局思路:先把状态当成一张不断变化的“协议表”

1.1 状态到底是什么

我在练习里把 LangGraph 的 State 理解为一张贯穿整个图运行的协议表。这张表由一组键值对组成,键就是状态字段,值就是当前这一刻的数据。每次调用图,都会初始化这张表,然后图里的节点依次处理它:

• 节点从状态表中读取自己关心的字段;
• 节点处理完业务后,返回一个字典;
• 这张字典不会整体替换状态,而是作为“部分更新”合并回状态表;
• 下一个节点拿到的,就是更新后的完整状态表。

这个设计和普通函数调用有本质区别。普通函数栈是私有的,参数传什么就是什么,返回值也只给调用方。StateGraph 的状态表是共享的,任何节点都能读,有了条件边之后,任何节点也都能根据状态决定别人会不会执行。因此,状态不仅仅是一个“参数集合”,它同时承担了通信、决策、审计这三件事。

1.2 这套练习要拿下的三个核心能力

我把状态代码练习拆成三个递进目标,后面的章节基本就是这个路数:

• 第一层,能定义清楚状态字段,并能设计节点返回值。这个阶段只要求把状态从一个节点传到另一个节点,搞清楚哪些键被覆盖、哪些键被保留。
• 第二层,能用状态字段驱动路由。给状态加一个类似status_code的字段,条件边读取这个字段决定下一个节点是继续、重试还是结束。
• 第三层,能用检查点保存和恢复状态。让图跑一半被中断后,可以从上一次状态继续,而不是每次从零开始。

当这三个能力都通过了,再回头写真实的工具调用、多步思考、人工审批,你会发现那些复杂场景本质上都只是这三件事的组合。

2. 从最小状态图开始,先跑通状态的流动

2.1 用 TypedDict 定义状态字段

最简单的状态代码练习,是定义一个只有count字段的状态,然后写一个节点把count加一。用 TypedDict 定义状态有几个好处:一是字段名字清晰,二是在 IDE 里能获得类型提示,三是后续加字段、加约束都很方便。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class SimpleState(TypedDict): count: int def increment(state: SimpleState) -> dict: return {"count": state["count"] + 1} builder = StateGraph(SimpleState) builder.add_node("increment", increment) builder.add_edge(START, "increment") builder.add_edge("increment", END) graph = builder.compile() print(graph.invoke({"count": 0}))

这段代码跑完会输出{'count': 1}。注意我调用graph.invoke({"count": 0})时,初始状态里手动传了count这个字段。如果传空字典{},increment里直接写state["count"]就会因为缺少count键直接抛异常。

注意:TypedDict 只是给编辑器看的类型声明,不会在运行时自动帮你补默认值。状态字段缺失时,需要自己在节点里用state.get("count", 0)来做兜底,或者在每次 invoke 时把完整初始状态传进去。

2.2 节点返回的是“部分状态”,不是完整状态

这是 LangGraph 新手最容易搞错的地方。节点返回的字典并不要求包含所有字段,只要返回你需要更新或新增的字段就够了。举个例子,如果状态表里有count和name两个字段,一个节点只返回{"count": 1},那name字段仍然保留原样,不会被清空。

class ProfileState(TypedDict): name: str count: int def first_node(state: ProfileState) -> dict: return {"count": 1} def second_node(state: ProfileState) -> dict: # 这里依然能读到 name return {"name": state["name"] + "-processed", "count": state["count"] + 1}

这种“部分更新”的设计非常实用。因为真实业务里,不同节点负责不同字段的更新,有的节点只改状态码,有的节点只追加消息。如果要求每个节点都返回完整状态,代码就会变得又啰嗦又容易出错。

2.3 串起来以后,状态是怎样流走的

把两个节点串在一起看,整个过程会更清晰:

builder = StateGraph(ProfileState) builder.add_node("first", first_node) builder.add_node("second", second_node) builder.add_edge(START, "first") builder.add_edge("first", "second") builder.add_edge("second", END) graph = builder.compile() print(graph.invoke({"name": "demo", "count": 0}))

执行顺序是:START → first → second → END。第一步,状态表是{"name": "demo", "count": 0};second 读到的状态是{"name": "demo", "count": 1};最后输出{"name": "demo-processed", "count": 2}。

我在这个阶段练习里最深刻的体会是:不用急着把图画得很复杂,只要反复做“节点返回部分状态、下一个节点读取状态”的小实验,把“谁写、谁读、什么时候写”的直觉建立起来,后面学条件边和检查点就会顺畅很多。

3. 把业务状态码做成路由信号

3.1 为什么状态码要放在状态里

状态代码练习里,我特别推荐给自己的状态表加一个status_code字段。这里的status_code不是 HTTP 状态码,而是业务状态码,常见值有pending、running、retry、completed、failed等。它的作用就是标记当前流程处于哪个阶段。

为什么要专门放一个这样的字段?因为它能让条件边的逻辑变得非常清晰。条件边需要做判断,判断依据只有两个来源:一个是输入参数,另一个是状态字段。输入参数只代表这一轮请求,而状态字段代表整个流程到目前为止的累积结果。把状态码放进状态里,意味着决策不仅依赖当前输入,还依赖之前所有节点留下的痕迹,这才是图状态最强大的地方。

3.2 写一个带状态码的重试流程

下面这个练习模拟一个最多执行三次的任务流程。第一次进入work节点时,状态码是running,路由到work继续执行;第二次、第三次只要没有完成,就继续回到work;累计执行三次后,把状态码改成completed,路由到结束。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class TaskState(TypedDict): task_id: str status_code: str attempts: int def start_process(state: TaskState) -> dict: # 初始标记为 running,表示任务开始执行 return {"status_code": "running"} def work(state: TaskState) -> dict: attempt = state.get("attempts", 0) + 1 # 这里故意用次数控制:前两次继续跑,第三次完成 if attempt < 3: status_code = "running" else: status_code = "completed" return {"attempts": attempt, "status_code": status_code} def route_by_status(state: TaskState) -> str: if state["status_code"] == "running": return "work" return "end" builder = StateGraph(TaskState) builder.add_node("start", start_process) builder.add_node("work", work) builder.add_edge(START, "start") builder.add_conditional_edges( "start", route_by_status, {"work": "work", "end": END}, ) builder.add_conditional_edges( "work", route_by_status, {"work": "work", "end": END}, ) graph = builder.compile() result = graph.invoke({"task_id": "task-001", "status_code": "pending", "attempts": 0}) print(result)

输出结果是:

{'task_id': 'task-001', 'status_code': 'completed', 'attempts': 3}

这个过程中,节点并没有通过返回值直接指定“下个节点是谁”,它只是更新了status_code。真正决定下一跳的,是条件边读取状态码后返回的路由键。这就是状态代码的核心价值:节点只负责改写状态,路由只负责读状态,两者解耦。

3.3 条件边的映射规则一定要写清楚

写条件边时,最忌讳的是只写“判断函数”忘了写“映射表”。看下面这个误用:

builder.add_conditional_edges("work", route_by_status) # 容易出问题

如果不提供映射字典,LangGraph 会默认把路由函数返回的字符串当成节点名。比如route_by_status返回"end",图会去找一个名叫end的节点,而不是结束。大部分情况下,我们需要的是返回"work"这种节点名,但要结束时就该返回END。所以更稳妥的写法是显式提供映射:

builder.add_conditional_edges( "work", route_by_status, {"work": "work", "end": END}, )

这里的键是条件函数可能返回的值,值是真正要去的目标节点或END。只要条件函数返回"end",就映射到END,图才会结束。

我在做这个练习之前,一度很困惑为什么图会莫名“卡住”或“多跑一轮”。后来排查发现,就是忘了加映射字典,导致返回的字符串被当成节点名,而图里根本没有那个名字的节点。所以现在只要写条件边,我都会先写映射表,再回头看路由函数。

4. 加入检查点,让状态能存、能读、能回放

4.1 没有持久化时,状态一次 invoke 就消失

默认情况下,graph.invoke()跑完调用后,状态就只存在于本次调用范围内。下一次再 invoke,状态表不会保留上一次运行的痕迹。这在单次流程里没问题,但真实的 Agent 场景经常需要跨轮次保持记忆。比如用户第一轮问“我要查订单”,第二轮提供订单号,系统需要记住第一轮聊到了哪一步。

LangGraph 解决这个问题的机制是检查点。检查点会把每一步执行后的状态快照保存下来。当我给编译后的图传入一个检查点时,图的每次执行都会先尝试从对应线程恢复历史状态,执行完再把新状态写回去。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "thread-task-001"}} result = graph.invoke( {"task_id": "task-001", "status_code": "pending", "attempts": 0}, config, ) print(result)

用上检查点之后,同一个thread_id相当于一条独立的“对话线”或“任务线”。不同线程之间的状态互不干扰,相同线程再次执行时,状态就会延续上次的进度。

4.2 thread_id 是状态存档的钥匙

thread_id本质上就是存档槽位。它放在config.configurable里,传给每次 invoke。我第一次跑的时候,总是忘记传config,导致检查点根本没生效;或者换了thread_id,发现状态对不上,白白排查了很久。

下面是一个刻意演示的读取场景,第一次跑完后,我用get_state拿出线程里保存的状态:

current = graph.get_state(config) print(current.values)

current.values就是该线程当前最新的状态表。如果流程执行到了某个中间步骤,我可以继续往同一个线程里传新的输入,节点会基于历史状态继续跑,而不是重新初始化。

next_result = graph.invoke( {"task_id": "task-001", "status_code": "running", "attempts": 1}, config, )

相同线程的状态会和历史状态合并:我在输入里只传了要更新的字段,其他字段依然保留。

4.3 用 get_state_history 做状态回放

状态练习中最有意思的部分,是回放历史快照。get_state_history()会返回这个线程所有历史检查点,从最新到最旧:

history = list(graph.get_state_history(config)) for snapshot in history: print(snapshot.values)

这个能力用在调试上非常香。比如某个节点把状态改错了,直接通过历史快照就能看到是哪个步骤引入的脏数据。我曾经遇到一个“状态码正确但内容丢了”的问题,靠历史快照逐条比对,最后发现是某个节点返回了空字典,把原本要追加的消息覆盖掉了。

在本地练习阶段,MemorySaver就够用。但它只在进程内存里保存状态,进程一重启就消失。如果要跨进程、跨机器保存,需要换用专门的检查点实现,把状态写到数据库或对象存储里。练习时不需要纠结,先把thread_id和get_state_history玩熟。

5. 组合练习:用消息列表模拟 Agent 的对话状态

5.1 用 reducer 把多条消息合并进状态

前面用的状态字段几乎都是“覆盖式更新”:节点返回什么,状态里就替换成什么。但真实 Agent 的对话消息列表需要的是“追加式更新”:每一轮新增的消息不能覆盖历史,而应该拼接到已有列表后面。

LangGraph 里通过Annotated类型指定 reducer。最常见的 reducer 是operator.add,它能让列表字段在每次更新时执行拼接。

from typing import Annotated, TypedDict from operator import add from langgraph.graph import StateGraph, START, END class ChatState(TypedDict): messages: Annotated[list[str], add] turn: int status_code: str def agent_reply(state: ChatState) -> dict: turn = state.get("turn", 0) + 1 message = f"第 {turn} 次回复" return { "messages": [message], "turn": turn, "status_code": "running", } def route_by_turn(state: ChatState) -> str: if state.get("turn", 0) < 3: return "agent" return "end" builder = StateGraph(ChatState) builder.add_node("agent", agent_reply) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", route_by_turn, {"agent": "agent", "end": END}) graph = builder.compile() result = graph.invoke({"messages": [], "turn": 0, "status_code": "pending"}) print(result["messages"])

输出结果大概是:

['第 1 次回复', '第 2 次回复', '第 3 次回复']

这就是 reducer 的作用。如果删掉Annotated[list[str], add],每次agent_reply返回{"messages": [message]}时,旧消息就会被整体覆盖,最后只剩下"第 3 次回复"。覆盖和追加,完全取决于字段类型里有没有声明 reducer。

5.2 把对话阶段写进状态码

在真实的对话流里,状态字段不应该只有消息列表,还应该有一个表示当前对话阶段的字段。比如awaiting_order_id表示等用户给订单号,querying_api表示正在查数据,completed表示结束。把这个阶段字段放进状态码后,条件边就成了对话管理器:

def route_by_phase(state: ChatState) -> str: if state["status_code"] == "awaiting_order_id": return "ask_user" if state["status_code"] == "querying_api": return "call_api" return "end"

这段逻辑的优点是,节点的职责被拆得很干净。节点只更新status_code,不管下一步去哪;路由函数只读status_code,不处理业务数据。所以我可以随时调整路由策略,比如加了人工审批,只要把approval节点的状态码改成pending_approval,让路由函数返回对应的处理节点即可。

5.3 配合检查点做多轮恢复

消息列表和检查点结合起来,就能模拟真实的多轮对话。第一轮调用后,状态里保存了这轮的消息;第二轮用同一个thread_id再次调用时,历史消息会被恢复出来。

config = {"configurable": {"thread_id": "chat-demo-001"}} graph.invoke( {"messages": ["你好，我要查订单"], "turn": 0, "status_code": "pending"}, config, ) graph.invoke( {"messages": ["订单号是 ABC-123"], "turn": 1, "status_code": "awaiting_order_id"}, config, ) state = graph.get_state(config) print(state.values["messages"])

输出里会同时包含第一轮和第二轮的消息,而不是只保留第二轮。这个机制对 Agent 的重要性怎么强调都不为过:记忆不是额外功能,它就是状态持久化的直接结果。

注意:配合 reducer 的字段,在每次 invoke 传输入时也要注意类型。比如messages是列表,你传的输入也必须是列表格式。如果传成字符串,运行时会因为类型不匹配报错。

6. 常见翻车点与排查思路

6.1 典型异常速查表

我练状态代码这段时间,把高频翻车场景整理成了表格,基本每一条自己都踩过一遍:

6.2 排查状态问题的三个动作

遇到状态不对劲,我习惯按三个步骤来,基本能定位九成问题。

第一步,打印本轮的初始状态。在节点函数开头加一行print(f"node={node_name}, state={state}"),看起来笨,但比任何调试工具都直观。只要每个节点都打印,状态从哪一步开始偏离就能一眼看出来。

第二步,调用get_state和get_state_history。如果用了检查点,这两条命令能告诉你当前线程的真实状态和历史快照。很多“状态为什么没更新”的疑惑,其实是程序在某个分支上用了完全不同的 thread,或者检查点根本没挂上。

第三步,把条件边单独提出来测试。条件函数本质上是普通函数,可以直接用一个模拟状态调用:

test_state = {"status_code": "running", "attempts": 2} print(route_by_status(test_state))

如果返回值和预期不符,就不用去图里猜了,问题就在路由函数或状态字段本身。

6.3 独家避坑心得

最后分享一个我自己很有用的习惯:在设计状态 schema 时,给每个字段的用途加上注释。比如status_code写“当前业务阶段,可选值: pending, running, completed, failed”;messages写“历史消息列表,使用 add reducer”。这种注释成本很低,但能极大避免后续自己改代码时“忘了字段是覆盖还是追加”的问题。

另一个心得是,练习状态代码时不要一步到位写大图。我见过很多人一上来就画十几个节点,结果状态更新乱成一团。我的做法是先写一个三节点小图,确认状态流、路由、检查点都正确,再逐步扩展。每加一个节点,就观察一次状态变化。这样即使出了问题,也能把范围缩小到最近一次改动。

如果你也在学 LangGraph,我建议从今天开始,把练代码的目的改成“练状态”,而不是“练功能”。每次写完图,多问自己一句:如果状态变成某个异常值,图会怎么走?如果你能准确回答这个问题,说明状态代码这关就算真正过了。