2.4 全局状态 (Graph State) 管理机制

2.4 全局状态 (Graph State) 管理机制

状态如何跨节点存活

本节在全册位置：状态是图的共享内存。每个节点看到的是同一份状态快照，返回的是增量；执行器用归约器把增量并回全局状态。所谓「全局」不是全局变量，而是每次运行（thread）独立的、由检查点托管的对象。我们把这部分单列一节，因为状态设计错，后面每一章都会踩坑。

机制：通道、归约与快照

状态被拆成若干「通道」（每个字段一条）。节点对通道的写入经归约器合并：operator.add 累加、add_messages 按 id 合并、自定义函数做任意变换。每次节点执行后会生成一个新状态快照，供下一步读取。这保证节点是纯函数，副作用被收口到归约器。类比建筑：通道像楼板里的管线，归约器像管件接头，快照像每层竣工图。快照机制也是第四章检查点能落盘的前提。

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    seen: Annotated[list, operator.add]   # 累加式通道
    last: str                              # 覆盖式通道

def n1(s: S):
    return {"seen": ["x"], "last": "x"}
def n2(s: S):
    return {"seen": ["y"], "last": "y"}

b = StateGraph(S)
b.add_node("n1", n1)
b.add_node("n2", n2)
b.add_edge(START, "n1")
b.add_edge("n1", "n2")
b.add_edge("n2", END)
g = b.compile()
print(g.invoke({"seen": [], "last": ""}))
​

# 两种通道并存是常态：历史用累加，焦点用覆盖
​

案例：累加日志通道

背景：希望保留每个节点产生的审计条目，又不丢前序，方便事后追责任务走了哪些节点。

操作：把日志字段设为 operator.add 通道，各节点只追加自己的条目。

结果：运行结束拿到完整日志，且天然按执行顺序，不依赖节点间传参。

解读：用累加通道代替「读旧值再拼新值」，避免竞态与样板代码，节点间解耦。

变式：覆盖式通道适合「当前焦点」，累加通道适合「历史」，两者常并存，设计状态先列清每条通道语义。

工程清单

• 状态被拆成若干通道，每条字段一条，写入经归约器合并，节点保持纯函数。

• 累加通道留历史，覆盖通道存当前焦点，两种并存是常态，设计状态先列清每条通道语义。

• 每次节点执行后生成新快照，快照机制是检查点能落盘、能回放的前提。

常见误区

把大对象塞进状态，检查点体积爆炸、序列化变慢；只放跨节点存活的数据，临时变量留在节点局部，这是状态设计最容易犯的错。

自定义归约器与滑动窗口

内置归约器覆盖不了所有场景时，写一个纯函数即可。归约器的签名是 (current_value, new_value) -> merged，框架保证它被顺序调用，你不用关心并发。

def keep_last_n(current: list, new, n=5):
    """只保留最近 n 条，先拼再截断"""
    merged = list(current) + (list(new) if isinstance(new, list) else [new])
    return merged[-n:]

class S(TypedDict):
    history: Annotated[list, keep_last_n]
​

自定义归约器最容易犯的错是改动了入参本身：current 是状态里的真实对象，原地 append 会让旧快照跟着变，破坏回放。先复制再合并是铁律。滑动窗口这个例子在长对话里很有用——上下文只留最近几轮，既控制 token 成本又避免检查点膨胀。

并行写冲突：报错是设计

两个节点在同一超级步进内写同一条「覆盖式」通道时，LangGraph 直接抛 InvalidUpdateError。这不是 bug，而是逼你声明合并语义：

class S(TypedDict):
    tag: str                    # 覆盖式：并行写会冲突
    notes: Annotated[list, operator.add]   # 累加式：并行写安全
​

遇到冲突有三个解法：给通道配归约器（累加、消息合并或自定义）；把两个节点改成串行（加边）；或把各分支的写操作收敛到一个节点再合并。选哪个取决于「这两个写是不是真的该合并」——语义上该合并就配归约器，不该合并就串行，别为了省事硬配一个语义错误的归约器。

快照是引用还是拷贝

执行器传给节点的 state 是当前快照。这里的细节决定你会不会踩「改了输入污染历史」的坑：归约器收到的是真实对象引用，而节点入参是快照的映射。因此——

1. 在节点里直接修改 state 的某个字段（如 s["list"].append(x)），不经过归约器，不会被记录到新快照。
2. 想「就地更新」必须返回增量，由归约器合并生成新快照。
3. 自定义归约器里复制后再改，避免污染上一份快照。

记住一句话：节点声明增量，归约器生成快照，执行器只负责调度。谁改了谁，关系一旦颠倒，回放与断点续跑都会失真。

add_messages 的合并细节

消息通道是全册用得最多的通道，add_messages 的行为值得单独过一遍：它按消息 id 去重，同 id 保留后到的那条，同时把新消息追加到列表末尾。id 相同通常意味着「同一条消息被重新生成」，去重避免重复。

from langchain_core.messages import AIMessage, HumanMessage

# id 相同：后者覆盖前者，列表不增长
m1 = AIMessage("旧版", id="m1")
m2 = AIMessage("新版", id="m1")
merged = add_messages([m1], m2)   # 结果为 [新版]，只留一条

# id 不同：直接追加
m3 = AIMessage("另一条", id="m2")
merged = add_messages([m2], m3)   # 两条都在
​

这个语义解释了「消息翻倍」类 bug 的根源：调用方每次构造新消息时不带 id，或 id 每次都随机生成，add_messages 就退化成普通 append。要控制上下文体积，与其依赖去重，不如在节点里显式裁剪旧消息再写回，两个机制配合使用。