Featured image of post 第六话|状态管理:从 dict 到 StateGraph

第六话|状态管理:从 dict 到 StateGraph

从手写 dict 到 StateGraph,理解 Agent 的状态管理。

系列 GYA(Get Your Agent) 第 6 / 12 篇
主题 AgentLLM教程StateGraphLangGraph

第五话我们写出了一个会多步决策的 Agent。回头看它的执行轨迹,你注意到一个细节吗——每一步之前,模型都要"想"一次

=== Step 1 ===
[Thought] 用户想要三件事:查天气、算数、总结...
[Action] 调用 get_weather(...)  →  [Observation] 北京: 多云,25℃
[Action] 调用 calculator(...)   →  [Observation] 56088

=== Step 2 ===
[Thought] 我有了两个结果,现在总结...
[Final Answer] 北京今天多云... 123×456=56088

也就是说,这个循环里**“下一步做什么"是由模型推理决定的**:模型说"我要调工具”,代码就去调;模型说"我该总结了",循环就结束。这很自由——但代价是,每一步的路由决策都花掉一次模型推理

现在问一个问题:如果这个流程是固定的——先理解意图、再调工具、最后总结——每次让模型重新"想"一遍下一步去哪,是不是有点浪费?更麻烦的是,如果任务是"查完笔记,有冲突就提醒,没冲突就生成报告",“去哪"的判断写在模型脑子里,它偶尔会想歪,而且每一步都在消耗 token 做路由而不是做任务。

我先说结论:

复杂任务的 Agent,需要一个显式的状态管理:

  • State:共享的状态对象(所有节点读写同一份)
  • Node:一个执行单元(函数:读 State → 做事 → 返回更新)
  • Edge:节点间的路由(确定性代码,不走 LLM) 这三个东西组在一起,就是 StateGraph——图管流程(确定性的传送带),LLM 管内容(只在需要语义的节点被调用)

这一篇,我们把"流程控制"从模型手里收回来,交给确定性的代码——先写一个手写状态机(零依赖),再用 LangGraph 的 StateGraph 跑同一个任务。你会发现:框架不是黑盒,它只是把你手写的东西声明式化了。

一、旧世界:循环里的状态为什么撑不住

第五话的 while 循环,本质上是个"自由形式"的流程:每次迭代,模型决定下一步做什么,代码照做。它的问题在第四话笔记里已经埋下伏笔——ReAct 让 LLM 包办一切:推理、路由、调用、总结,全在循环里。

规模化后,四个问题暴露:

问题表现
条件分支不可靠“查完笔记该去哪"写在模型推理文本里,模型偶尔跑偏
只能串行需要"同时查 3 个工具再聚合"时,ReAct 只能一个个来
无法优雅暂停想"等用户确认再继续”,循环很难挂起
路由成本高每一步都让 LLM 想"下一步去哪”,token 费在路由上而不是任务上

关键洞察:ReAct 循环的问题不是"循环"本身,而是路由决策也交给了 LLM。让模型每次都想"下一步去哪",既贵又不稳——路由这种事,代码做得又快又准。

二、转折:显式状态机——State / Node / Edge 三件套

解决思路来自一个古老而成熟的概念:状态机(State Machine)。把它用在 Agent 上,就是三个核心概念:

class AgentState(TypedDict):        # State:共享状态
    messages: list                  #   对话历史
    intent: list                    #   用户意图(要哪些工具)
    results: dict                   #   工具结果
    next_step: str                  #   路由指针

def parse_intent(state) -> dict:    # Node:执行单元
    result = llm.parse_intent(...)
    return {"intent": result}       #   只更新需要的字段

# Edge:路由表(确定性代码)
if state["next_step"] == "parse_intent": parse_intent(state)
elif state["next_step"] == "execute_tools": execute_tools(state)

对照你熟悉的 Java 概念:

Java 状态机StateGraph
State 枚举Node(可执行代码)
transition(event)Edge(自动流转)
Context 上下文State(共享字典)
Guard 条件条件边的路由函数(确定性 Python)

关键设计原则:图管流程,LLM 管内容。

图的节点 = 一个单元操作
├── 纯代码节点   → 调工具、做计算(零 token)
├── 代码+LLM 节点 → 需要语义推理时调 LLM
└── 纯 LLM 节点   → 理解意图、生成文本

图的边 = 节点间路由
├── 普通边     → 永远从 A 到 B
└── 条件边     → 根据 State 决定(确定性代码,不走 LLM)

关键洞察:图是骨架(确定性的传送带),LLM 是肌肉(只坐到需要语义推理的工位上)。图的职责是控制 LLM 被调用的时机和次数——该省的路由 token 一分不花。

三、原理讲透:为什么状态要"显式"?

你可能觉得:第五话的循环里也有 state 啊,为什么要"显式"?区别在于谁在管理状态的流转

第五话手写循环状态机
状态存在哪零散在循环代码里一个显式 State 对象
下一步去哪模型推理决定(或 if-else 散落)next_step 字段 + 路由表
节点边界没有,循环体是"一大坨"每个 Node 独立函数
调试靠打印看 State 快照即知全貌

在展开之前,先精确理解 State 的生命周期——这是状态管理和第五话随手传 list 的根本区别:

1. 初始 State:{"question": "...", "messages": [...], "results": {}, "intent": "", "next_step": "parse_intent"}
2. parse_intent 执行 → 返回 {"intent": [...]}        ← 只更新 intent 字段
3. 引擎把返回合并进 State → next_step 指向 execute_tools
4. execute_tools 执行 → 返回 {"results": {...}, "messages": [...]}  ← 只更新自己的字段
5. ...循环...
6. summarize 返回 {"final_answer": "..."} → next_step=END → 输出最终 State

每个 Node 是一个纯函数:输入完整 State,返回"要更新的字段子集"。Node 不直接改 State——它返回更新,由引擎(手写循环或框架)合并。这个设计保证了:任何节点都能拿到完整上下文(读 State),但只负责自己那部分(写自己的字段),不会互相踩踏。这也是"可并行"和"可恢复"能成立的原因——状态的读写是受控的,不是随处可改的全局变量。

“显式"的三个直接收益

  1. 可审计:任何时刻,State 里的 intent/results/next_step 告诉你 Agent 走到哪了、已经干了什么。挂掉时一眼看出问题。
  2. 可恢复:State 是数据,可以序列化存盘。长任务挂了,从上次 State 恢复,不用从头跑(检查点 checkpoint)。
  3. 可并行:State 明确后,互不依赖的分支可以并行执行再聚合(StateGraph 支持 fan-out/fan-in)。

这三件事正是生产 Agent 和 demo Agent 的分水岭——demo 靠调试,生产靠状态管理

这里值得多说两句"可并行"和"可恢复”,因为它们是把状态管理从"概念正确"推向"工程价值"的关键。

并行(fan-out / fan-in):假设一个任务要同时查三个数据源(天气、股票、新闻),ReAct 循环只能串行——先查天气、等结果、再查股票、再等结果。显式状态机可以把三个查询节点做成并行分支,全部完成后在聚合节点合并。LangGraph 的 Send API 就是干这个的。对延迟敏感的生产系统,这可能是 3 倍和 1 倍的差别。代价是状态必须可合并——每个并行分支只更新自己的字段,不能互相覆盖,这正是 State(共享 dict)设计时要小心的。

检查点(checkpoint):State 既然是数据,就可以序列化。LangGraph 有内置的 checkpointer——每步执行完把 State 存下来。长任务(比如批量处理 1000 条数据)跑到第 700 条挂了,恢复时从第 700 条的 State 继续,而不是从头跑。这对 token 成本和工程效率是量级的差别。手写版要做到这点,就是在 run() 里每步 json.dumps(state) 存盘——原理一样,框架替你做了。

还有一点值得注意:State 的字段设计直接影响系统复杂度。原则是"每个节点只更新自己关心的字段"——parse_intent 只写 intent,execute_tools 只写 resultsmessages,互不干扰。如果所有节点都直接改一个"大杂烩" dict,并行和检查点都会变得困难。这跟 Java 里"类职责单一"是同一个道理。

四、动手:同一个任务,两种实现

代码https://github.com/renxin2024/GYA/tree/main/c06-state-management

  • state_machine.py:手写状态机(零依赖)
  • langgraph_demo.py:LangGraph StateGraph(pip install langgraph
export DEEPSEEK_API_KEY=sk-你的key
python3 state_machine.py
uv run --with langgraph python3 langgraph_demo.py   # 或 pip install langgraph

任务:“北京天气怎么样?顺便算一下 123*456,最后把两个答案整理成一句话。”

手写版:三节点状态机

class StateMachine:
    def __init__(self, question):
        self.state = {
            "question": question,
            "messages": [...],
            "results": {},       # 工具结果
            "intent": "",        # 语义理解结果
            "next_step": "parse_intent",   # 路由指针
        }

    def parse_intent(self):      # Node 1:LLM 节点(唯一需要语义的)
        intent = llm.parse(...)
        self.state["intent"] = intent
        self.state["next_step"] = "execute_tools"

    def execute_tools(self):     # Node 2:代码节点(路由确定性)
        for tc in tool_calls: ...执行...
        self.state["next_step"] = "summarize"

    def summarize(self):         # Node 3:LLM 节点
        self.state["final_answer"] = llm.summarize(...)
        self.state["next_step"] = "END"

    def run(self):               # 状态机引擎:while + next_step 路由
        while self.state["next_step"] != "END":
            node = self.state["next_step"]
            getattr(self, node)()

实测输出(2026-08-20,deepseek-v4-flash):

=== Node: parse_intent ===
=== Node: execute_tools ===
=== Node: summarize ===
最终答案: 北京天气多云、25℃、东北风3级;另外,123×456的计算结果是56088。
执行轨迹: intent=['get_weather', 'calculator']
          results={'get_weather': '多云,25℃...', 'calculator': '56088'}

注意三件事:

  • 路由是 next_step 字段跳转,不是模型推理——parse_intent → execute_tools → summarize,每一步去哪由代码决定
  • LLM 只被调了 3 次:parse_intent(理解意图)、execute_tools 里(决定工具参数)、summarize(总结)——路由本身零 token
  • State 一眼可查intentresults 告诉你"Agent 认为要做什么、已经得到了什么"

LangGraph 版:同一三节点,声明式表达

g = StateGraph(AgentState)
g.add_node("parse_intent", parse_intent)
g.add_node("execute_tools", execute_tools)
g.add_node("summarize", summarize)
g.set_entry_point("parse_intent")
g.add_edge("parse_intent", "execute_tools")
g.add_conditional_edges("execute_tools", route_after_tools, {"summarize": "summarize"})
g.add_edge("summarize", END)
graph = g.compile()
result = graph.invoke({...})

节点函数和手写版一模一样parse_intent 的代码是复制的)——区别只在:

  • 手写版:while 循环 + getattr(self, next_step) 路由
  • LangGraph:add_node + add_edge 声明 + 框架内部替你跑循环

实测输出与手写版一致(intent 相同、results 相同、最终答案相同):

[Node: parse_intent] intent=['get_weather', 'calculator']
[Node: execute_tools] get_weather → 多云,25℃,东北风 3 级
[Node: execute_tools] calculator → 56088
[Node: summarize] 北京天气多云、25℃;123×456的结果是56088。

关键洞察:跑同一个任务,手写版和 LangGraph 版输出完全一致——框架没有魔法,它只是把 while 循环和路由表声明式化了。你手写过一遍,用任何框架都是"换个写法"。

五、工程化延伸:什么时候该上 StateGraph?

状态机不是万能的。什么时候该用,什么时候 ReAct/手写循环就够?

该用 StateGraphReAct/手写循环就够
流程固定、高频重复低频、一次性、探索性任务
需要并行 + 聚合(fan-out/fan-in)单步骤顺序执行就够
需要人机交互中断(等确认)不需要中间暂停
需要审计追溯、断点恢复不需要追踪状态
规模大、token 成本敏感一天跑一两次

一个直觉判断法:如果流程图能画出来(固定节点+固定边),就用 StateGraph;如果流程本身是探索性的(不知道下一步是什么),用 ReAct。

作为系列的双语言惯例,这里说一句 Java 版的事:LangGraph 是 Python 生态库(基于 asyncio 和 TypedDict 设计),没有官方 Java 等价实现。所以第六话的 Java 版(GYA-Java c06-state-management/)实现的是手写状态机等价版——State 用 Map,Node 用 Function,图声明就是一张注册表:

static final Map<String, Function<Map<String, Object>, Void>> NODES = new LinkedHashMap<>();
NODES.put("parse_intent",  state -> parseIntent(state));
NODES.put("execute_tools", state -> executeTools(state));
NODES.put("summarize",     state -> summarize(state));
// 引擎:while + next_step 路由(与 Python 版同构)

用 Java 跑同一个任务,输出和 Python 版完全一致(intent/results/最终答案相同)。这说明:状态机的思想是语言无关的——不管用 Python 的 LangGraph、Java 的手写注册表,还是你未来可能遇到的任何 Agent 框架,State/Node/Edge 三件套都是同一套骨架。

另外澄清一个常见误区:ReAct 不是 StateGraph 的对立面,而是 StateGraph 的一个特例——三节点循环图(Think → Act → Observe → 回环)。类比:ArrayListList 的实现,ReAct 是 StateGraph 的一种工作流模式。用 StateGraph 也能画出 ReAct 图(第五话的循环就是),只是 StateGraph 能表达的远不止 ReAct。

下一步

State 管住了"流程怎么走"。但还有一个大问题没解决:模型的记忆

第五话里我们把所有历史都塞进 messages——对话 20 轮后,prompt 越来越长,直到超出上下文窗口。这时会发生什么?模型"忘记"最早说的话。上下文窗口 ≠ 记忆

下一篇第七话,我们讲三层记忆架构:上下文(窗口内)、短期(会话内)、长期(跨会话)——以及向量检索怎么让 Agent"记住"重要的事。


演示代码https://github.com/renxin2024/GYA/tree/main/c06-state-management(state_machine.py 零依赖 + langgraph_demo.py 需 langgraph;DeepSeek 官方 API,默认模型 deepseek-v4-flash)。Java 21 手写状态机等价实现见独立仓库 https://github.com/renxin2024/GYA-Javac06-state-management/,Gradle 工程,gradle run)。 参考:StateGraph 三概念(State/Node/Edge)、图管流程 LLM 管内容、ReAct 是 StateGraph 特例,来自 LangGraph 官方文档与 Agent 工程实践(hello-agents 讲义延伸);demo 输出为 2026-05-12 本机实测(deepseek-v4-flash,Python 手写版/LangGraph 版/Java 版三跑一致)。

留言

欢迎分享你的想法。评论提交后会在审核通过后显示。