# LangGraph 并行与 reducer：菜单为什么变成了 24 个菜

配套代码：

- `04_langgraph_paramlizm_demo.py` —— 错误写法的现场，就是本文要解释的那 24 个菜
- `05_langgraph_paramlizm_demo2.py` —— 追踪版，把 menu 的每一次变化打出来（见第六节）

本文所有输出都在 langgraph 1.2.1 上实测过。

## 一句话结论

`menu` 字段挂了 reducer（`Annotated[List[str], add]`），**有 reducer 的字段，节点返回的值是"增量"，会被追加到通道上，而不是覆盖它**。原代码既 `append` 改了列表本身，又 `return state` 把改完的整份列表当增量交回去，等于每个节点都写了两遍，逐层翻倍到 24。

## 现象

三个厨师各做一道菜，期望 3 条，实际：

```
🍽️ 所有菜都准备好了！菜单如下：   ← serve 里打印出 12 条
 - 鱼香肉丝
 - 番茄炒蛋
 ...（共 12 条）

最终结果: {'menu': [...24 条...], 'all_ready': True}
```

注意 `serve` 打印的是 12 条，而 `invoke` 返回的是 24 条——**这是两级不同的翻倍**，后面会拆开看。

## 一、reducer 是什么

先说 LangGraph 里 state 的真实模型。你写的 `MyState` 不是一个普通字典，它的每个字段都是一个 **通道（channel）**，通道有自己保存的值，也有自己的"合并规则"。

节点函数拿到的 `state` 是通道当前值的一份快照，节点的返回值是**要对通道做的更新**。至于这份更新怎么落到通道上，就由这个字段的合并规则决定——这个规则就是 **reducer**。

reducer 就是一个两参数的函数：

```python
def reducer(通道现在的值, 节点返回的值) -> 通道的新值
```

字段声明成 `Annotated[类型, 函数]` 时，那个函数就是 reducer：

```python
class MyState(TypedDict):
    menu: Annotated[List[str], add]   # add 就是 reducer，来自 operator
    all_ready: bool                   # 没写 Annotated，就没有 reducer
```

`operator.add` 作用在两个 list 上就是拼接，所以 `add(["番茄炒蛋"], ["宫保鸡丁"])` 得到 `["番茄炒蛋", "宫保鸡丁"]`。

**没有 reducer 的字段，默认行为是覆盖**，节点返回什么，通道就变成什么：

```python
class S(TypedDict):
    menu: List[str]          # 无 reducer

def a(s): return {"menu": ["番茄炒蛋"]}
def b(s): return {"menu": ["宫保鸡丁"]}   # a 之后串行执行

# 实测输出: {'menu': ['宫保鸡丁']}     ← 番茄炒蛋被覆盖掉了
```

这就是两类字段的根本差别，也是后面所有坑的来源：

| 字段 | 声明 | 节点返回值的含义 |
|---|---|---|
| `all_ready` | `bool` | 新值，**覆盖** |
| `menu` | `Annotated[List[str], add]` | 增量，**追加** |

## 二、为什么并行必须有 reducer

顺手回答一个问题：既然 reducer 这么容易踩坑，能不能不写？并行的时候不行。

把 `menu` 的 `Annotated` 去掉，让 a、b 两个节点并行写它：

```python
class S1(TypedDict):
    menu: List[str]          # 故意不加 reducer

def a(s): return {"menu": ["A"]}
def b(s): return {"menu": ["B"]}   # 与 a 并行
```

实测直接报错：

```
InvalidUpdateError At key 'menu': Can receive only one value per step.
Use an Annotated key to handle multiple values.
```

道理很直白：同一步里 a 和 b 各交回一个值，覆盖语义下 LangGraph 无法决定该听谁的——它不猜，直接报错让你给个规则。**reducer 就是你给的那个规则**，告诉它"两份都要，拼起来"。

所以 `menu` 上那个 `add` 不是可选的装饰，是并行汇聚的必要条件。它同时也意味着你必须按增量的方式写节点。

## 三、reducer 是被逐个调用的

这一点最影响直觉，所以单独验证。把 `add` 换成一个会打印的版本：

```python
def traced_add(old, new):
    print(f"[reducer] old={old}  new={new}")
    return add(old, new)
```

正确写法（各节点只返回自己的增量）下的实测输出：

```
start: 收到 state = {'menu': []}
a: 收到 state = {'menu': []}
b: 收到 state = {'menu': []}
   [reducer] old=[]              new=['番茄炒蛋']
   [reducer] old=['番茄炒蛋']     new=['宫保鸡丁']
serve: 收到 state = {'menu': ['番茄炒蛋', '宫保鸡丁']}
```

三个要点：

1. **reducer 不是一次收到所有并行结果**，而是每个节点的更新来一次，`old` 是累积到目前的通道值，`new` 只是那一个节点返回的东西。
2. **并行节点看到的 state 是同一份起始快照**（a 和 b 都看到 `{'menu': []}`），a 的结果对 b 不可见——所以别指望在并行节点里读到兄弟节点写的东西。
3. `menu` 有 reducer，所以通道初始值就是 `[]`，不需要 `start` 去初始化。原代码里的 `state["menu"] = []` 是多余的。

## 四、24 是怎么算出来的

原代码每个厨师节点干了两件事：

```python
def chef_a(state):
    state["menu"].append("番茄炒蛋")   # ① 就地改了列表
    return state                      # ② 又把整份列表当增量交回去
```

**错误 ①：三个并行节点改的是同一个列表对象。** 打印 `id(state["menu"])` 实测：

```
a   拿到 menu 的 id = 2635165391232
b   拿到 menu 的 id = 2635165391232
c   拿到 menu 的 id = 2635165391232      ← 同一个对象
```

快照是浅的，`state` 字典是各自的，但里面的 list 是同一个。三个 `append` 全落在这一个列表上，于是它变成 3 条。这一步绕过了 reducer，是直接篡改通道里的数据。

**错误 ②：`return state` 让这 3 条又被当作增量追加。** 节点结束时，LangGraph 拿返回值里的 `menu` 交给 reducer。此时那个列表已经是 3 条了，于是每个节点都贡献一次"3 条"的追加。

两个错误叠在一起，用带打印的 reducer 实测原代码：

```
   [reducer] len(old)=0  len(new)=0  -> len=0      ← start 返回空 menu
   [reducer] len(old)=3  len(new)=3  -> len=6      ← 三个 append 已让列表变 3 条
   [reducer] len(old)=6  len(new)=3  -> len=9
   [reducer] len(old)=9  len(new)=3  -> len=12
serve 看到的菜单条数: 12
   [reducer] len(old)=12 len(new)=12 -> len=24     ← serve 的 return state
final len: 24
```

逐行读这段就能把两级翻倍分清：

- 第一级，**3 → 12**：三个 `append` 把共享列表变成 3 条（这时 `old` 已经是 3 了，因为 append 改的就是通道里那个对象），再被三次 `return state` 各追加 3 条。这是 `serve` 打印 12 条的原因。
- 第二级，**12 → 24**：`serve` 自己也 `return state`，把它刚读到的 12 条整份又追加一遍。这是最终结果 24 条的原因。

第二级值得单独强调，因为它跟 `append` 无关。只犯 `return state` 这一个错就够翻倍了：

```python
def start(s): return {"menu": ["番茄炒蛋", "宫保鸡丁", "鱼香肉丝"]}
def serve(s):
    s["all_ready"] = True
    return s          # ← 唯一的错

# 实测: {'menu': [3 个菜 + 同样的 3 个菜], 'all_ready': True}   ← 6 条
```

**汇聚节点（fan-in）是这个坑的高发地。** 它只想改 `all_ready`，但 `return state` 顺带把读到的完整 `menu` 又交了一遍。

## 五、正确写法

规则一条：**节点只返回自己要写的那几个 key 组成的新 dict，不 `return state`，不改 `state` 里的可变对象。**

```python
def start_cooking(state: MyState):
    print("👨‍🍳 厨房准备开始！")
    return {}                        # 什么都不写，就返回空 dict

def chef_a(state: MyState):
    time.sleep(2)
    return {"menu": ["番茄炒蛋"]}     # 只交自己新增的那一份

def serve_dish(state: MyState):
    for dish in state["menu"]:       # 读是随便读的
        print(" -", dish)
    return {"all_ready": True}       # 千万不要带上 menu
```

实测输出：

```
🍽️ 所有菜都准备好了！菜单如下：
 - 番茄炒蛋
 - 宫保鸡丁
 - 鱼香肉丝

最终结果: {'menu': ['番茄炒蛋', '宫保鸡丁', '鱼香肉丝'], 'all_ready': True}
```

菜品顺序是 a、b、c 的声明顺序，不是完成时间顺序（c 只睡 1 秒但排最后）。`add` 按 LangGraph 处理节点更新的顺序拼接，跟谁先跑完无关——**并行汇聚的结果顺序是稳定的，但别理解成"先完成先入列"**。

## 六、自己动手看：`05_langgraph_paramlizm_demo2.py`

上面的日志都是这个脚本打出来的。它把 `add` 换成会打印的 `traced_add`，两种写法共用同一张图，只换节点函数：

```bash
python -X utf8 05_langgraph_paramlizm_demo2.py              # 错误/正确 对比跑
python -X utf8 05_langgraph_paramlizm_demo2.py --mode wrong # 只看错误写法
python -X utf8 05_langgraph_paramlizm_demo2.py --stream     # 再用 stream 看增量/快照
python -X utf8 05_langgraph_paramlizm_demo2.py --no-sleep   # 去掉 sleep，秒出
```

三个观察点：

**1. `🔧 [reducer #n]` 日志** —— 每行是一次合并，`old` 是通道现值，`new` 是某个节点交回来的东西。错误写法 5 次调用（`new` 全是 3 条或 12 条），正确写法 3 次（`new` 都是 1 条）。

**2. `👀` 行的 `id=`** —— 错误写法里三个厨师的 `id` 相同，正确写法里也相同（快照是共享的），差别在于正确写法**不去改它**。这解释了为什么"共享"本身不是问题，`append` 才是。

**3. `--stream` 的两种模式**，正好对应 reducer 的两个参数：

```
stream_mode="updates"  → 节点交回的增量，就是 reducer 的 new
stream_mode="values"   → 合并后的通道快照，就是 reducer 的 old
```

错误写法在 `updates` 下最扎眼——`chef_a` 交回来的不是 1 个菜而是一整份：

```
── 更新 2 ── chef_a 交回 menu(3) = ['番茄炒蛋', '宫保鸡丁', '鱼香肉丝']
── 更新 3 ── chef_b 交回 menu(3) = ['番茄炒蛋', '宫保鸡丁', '鱼香肉丝']
── 更新 4 ── chef_c 交回 menu(3) = ['番茄炒蛋', '宫保鸡丁', '鱼香肉丝']
── 更新 5 ── serve  交回 menu(12) = [...12 条...]
```

`values` 下只有 3 个快照（0 条 → 12 条 → 24 条），比节点数少——**并行的三个厨师合并完才 yield 一次**，这是它和 `updates` 的本质区别。

### 顺手看到的两个副作用

跑 `--mode wrong` 时，`append` 的执行顺序就是**完成时间顺序**：带 sleep 时 c 只睡 1 秒，菜单变成 `['鱼香肉丝', '番茄炒蛋', '宫保鸡丁']`；加 `--no-sleep` 又变回 a/b/c。而正确写法两种情况下都稳定是声明顺序 `['番茄炒蛋', '宫保鸡丁', '鱼香肉丝']`。**就地修改让结果依赖时序，返回增量不会。**

`--no-sleep` 跑错误写法还能看到更难查的一幕：

```
👀 chef_a 看到 menu(0) = []
👀 chef_b 看到 menu(1) = ['番茄炒蛋']      ← b 看见了 a 写的东西
👀 chef_c 看到 menu(2) = ['番茄炒蛋', '宫保鸡丁']
```

按 LangGraph 的语义，并行节点应该都看到同一份起始快照（正确写法里三个都是 `menu(0)=[]`）。这里之所以互相可见，正是因为 `append` 改的是共享对象。**并行节点之间出现了本不该存在的可见性，而且这种可见性随时序变化**——比数量翻倍更难排查。

## 七、检查清单

写并行图时按这几条自查：

- [ ] 会被多个并行节点写的字段，加了 `Annotated[..., reducer]` 吗？（不加会 `InvalidUpdateError`）
- [ ] 节点里有没有 `state[...].append(...)` / `.update(...)` / `+=` 这类就地修改？有就是 bug，并行节点共享同一个对象。
- [ ] 有没有 `return state`？带 reducer 的字段一旦这么写就翻倍。全部改成返回只含自己输出 key 的 dict。
- [ ] 汇聚节点单独看一遍：它读了带 reducer 的字段，返回值里绝不能再出现那个字段。
- [ ] 有没有在初始化节点里给带 reducer 的 list 字段赋 `[]`？多余，通道本来就是空的。

## 八、和本项目 `_004_` 的关系

`_004_langgraph_more_nodes/agent_state.py` 里的 `AgentState` **目前所有字段都没有 reducer**，节点也都是串行的，所以现在的 `def x_node(state) -> AgentState: ...; return state` 写法是安全的——覆盖语义下 `return state` 交回原值等于没改。

但如果以后要并行（比如六类实体的匹配同时跑，或者多路 Cypher 并发执行），给某个字段加上 reducer 的那一刻，**这个字段上所有 `return state` 的节点都会立刻开始翻倍**，而且不报错，只是结果变多——比报错难查。届时的改法就是本文第五节：那条链上的节点全部改成返回增量 dict。