写多轮对话 Agent 最常见的翻车现场:跑了几轮,模型突然"失忆",前面的对话全部消失。十有八九不是模型的问题,而是状态合并规则写错了——消息列表用了默认的覆盖语义,每轮回复都把历史冲掉。这篇从 LangGraph 的 StateGraph 入手,把"状态怎么流、怎么合并"讲透。
四个概念一个比喻
StateGraph 的世界只有四样东西:节点是处理函数,边决定执行顺序,状态是全图共享的一份数据,Schema 声明这份数据长什么样。可以用流水线来类比——节点是工位,边是传送带,状态是传送带上流动的工件,每个工位只声明"我改了工件的哪几个属性",不需要知道全貌。
分层关系值得记住:StateGraph 在 langchain-core 之上、create_agent 之下。很多人直接从高层封装入门,出了问题不知道往哪层找——先学会底层状态图,上层的封装才透明。
十行代码跑通最小图
从最简单的整数状态开始,避免一上来就碰消息列表:
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class CounterState(TypedDict):
count: int
def increment(state: CounterState) -> dict:
return {"count": state["count"] + 1}
graph = StateGraph(CounterState)
graph.add_node("increment", increment)
graph.add_edge(START, "increment")
graph.add_edge("increment", END)
print(graph.compile().invoke({"count": 0})) # {'count': 1}
三个细节:节点返回的是部分更新而不是完整状态;START/END 是虚拟节点;同一个键被两个节点先后写入时,默认后写覆盖前写。
Reducer:合并规则才是灵魂
覆盖语义在消息场景是灾难。多轮对话里每条新回复都要"追加"进历史,覆盖等于每轮清空记忆。解法是给键显式指定 Reducer——一个决定"旧值和新值怎么合成结果"的函数:
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class ChatState(TypedDict):
messages: Annotated[list, add_messages]
Annotated 上的第二个参数就是合并规则,add_messages 负责按消息 ID 追加与更新。自定义 Reducer 也简单,写一个 (old, new) -> merged 的函数挂上去即可。
调试合并逻辑有个必杀技:app.invoke(state, config={"print": True}) 之类的方式开启逐步输出(或用 app.stream()),能清楚看到每个节点写入前后的状态差异。合并 bug 99% 靠 stream 输出定位,不用猜。
并发写冲突与两条实战铁律
条件边之外还有一个新手盲区:并行节点同时写同一个键。两个节点同时向 messages 追加没有问题(追加可交换),但两个节点同时覆盖同一个字符串就会报 InvalidUpdateError。设计状态时先问一句:这个键会不会被并行写入?会的话要么加 Reducer,要么串行化。
另外两条工程经验:
- Schema 定义集中放一处。多个模块各自定义同名状态类,类型检查形同虚设,改字段时必然漏改。
- 节点函数保持纯函数。在节点里做 IO、改全局变量,会让图的重放与恢复机制(checkpoint)失效——状态图的价值一半在可恢复性,别亲手废掉它。
什么场景不该用图
单轮问答、纯顺序的三步管道,直接写函数调用链更清晰;状态图的收益在条件路由、循环(工具调用重试)、并行分支这些控制流复杂度上。控制流不复杂就别上图,图本身的调试成本不低,用简单抽象解决简单问题才是正路。

评论0