第 2 章 · 状态、Reducer 与并发合并

当两个节点同时生成结果时，谁负责合并？本章用两位“检查员”向同一个列表写入结果，观察 Reducer 如何定义共享状态的更新语义。

工作机制与设计选择

未声明 Reducer 的字段默认覆盖；一个 superstep 内多个节点同时更新同一普通字段可能触发并发更新错误。用 Annotated[list[str], operator.add] 表示列表拼接。Reducer 是数据语义，不是互斥锁；列表拼接不会自动去重。并行结果的展示顺序不应成为业务假设，测试时可排序。汇聚节点用列表形式的起点等待两条分支完成。

示例

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

class State(TypedDict):
    findings: Annotated[list[str], operator.add]
    summary: str

def style(state: State):
    return {"findings": ["style: pass"]}

def content(state: State):
    return {"findings": ["content: pass"]}

def summarize(state: State):
    return {"summary": "; ".join(sorted(state["findings"]))}

builder = StateGraph(State)
builder.add_node("style", style)
builder.add_node("content", content)
builder.add_node("summarize", summarize)
builder.add_edge(START, "style")
builder.add_edge(START, "content")
builder.add_edge(["style", "content"], "summarize")
builder.add_edge("summarize", END)
result = builder.compile().invoke({"findings": [], "summary": ""})
assert len(result["findings"]) == 2
assert result["summary"] == "content: pass; style: pass"
print(result)

运行与验收

安装 langgraph，保存并运行本例。验收不仅检查列表长度，还检查汇聚结果包含两位检查员的输出。初始列表为空，避免重复把历史结果作为新输入追加。

容易踩的坑

不要为了消除错误给所有字段都加 operator.add。例如金额相加和替换余额是两种业务语义，混用会造成数据错误。重复运行也可能追加重复项；需要幂等时使用稳定 ID 及自定义合并规则。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 默认字段如何更新？

2. operator.add 会去重吗？

3. 汇聚前要等待谁？

4. Reducer 是什么？

动手实践

1. 增加第三位检查员并修改汇聚依赖，断言三条结果齐全。
2. 改写 Reducer，按稳定 ID 去重，并测试重复输入。
3. 解释为什么 summary 不需要列表拼接 Reducer。

官方参考

• graph-api

核验日期：2026-09-17。