<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>黑马</title>
</head>
<body>
黑马程序员。
<h1>黑马程序员</h1>
<h2>黑马程序员</h2>
<h3>黑马程序员</h3>
</body>
</html>

# 创建请求客户端
import requests

response = requests.get("http://127.0.0.1:8000/")
print(f"请求码：{response.status_code}")  # 200
print(f"获取到文本数据:{response.text}")
# print(f"获取到的json数据为：{response.json()}")  # {'message': 'Hello World'}

import time
from operator import add
from typing import TypedDict, List, Annotated

from langgraph.graph import StateGraph, START, END


# 定义状态结构
class MyState(TypedDict):
    menu: Annotated[List[str], add]  # ✅ 支持并行追加
    all_ready: bool


# --- 节点定义 ---
def start_cooking(state: MyState):
    print("👨‍🍳 厨房准备开始！")
    # 初始化共享菜单列表
    state["menu"] = []
    return state


def chef_a(state: MyState):
    print("👩‍🍳 A 厨师在做：番茄炒蛋...")
    time.sleep(2)
    # 向共享 list 添加菜品
    state["menu"].append("番茄炒蛋")
    return state


def chef_b(state: MyState):
    print("👨‍🍳 B 厨师在做：宫保鸡丁...")
    time.sleep(3)
    state["menu"].append("宫保鸡丁")
    return state


def chef_c(state: MyState):
    print("👩‍🍳 C 厨师在做：鱼香肉丝...")
    time.sleep(1)
    state["menu"].append("鱼香肉丝")
    return state


def serve_dish(state: MyState):
    print("\n🍽️ 所有菜都准备好了！菜单如下：")
    for dish in state["menu"]:
        print(" -", dish)
    state["all_ready"] = True
    return state


# --- 构建 LangGraph ---
def build_graph():
    graph_builder = StateGraph(MyState)
    graph_builder.add_node("start", start_cooking)
    graph_builder.add_node("chef_a", chef_a)
    graph_builder.add_node("chef_b", chef_b)
    graph_builder.add_node("chef_c", chef_c)
    graph_builder.add_node("serve", serve_dish)

    # 启动 → 并行三个厨师
    graph_builder.add_edge(START, "start")
    graph_builder.add_edge("start", "chef_a")
    graph_builder.add_edge("start", "chef_b")
    graph_builder.add_edge("start", "chef_c")

    # 三个厨师汇总到 serve
    graph_builder.add_edge(["chef_a", "chef_b", "chef_c"], "serve")

    graph_builder.add_edge("serve", END)

    return graph_builder.compile()


# --- 运行 ---
if __name__ == "__main__":
    graph = build_graph()
    result = graph.invoke({})
    print("\n最终结果:", result)

# 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。

"""
并行 + reducer 追踪版：把 menu 通道的每一次变化打出来。

04_langgraph_paramlizm_demo.py 是"错误写法"的现场，本文件是它的显微镜：
把 operator.add 换成一个会打印的 traced_add，就能看见 menu 每一步
从什么变成什么，从而搞清楚 3 个菜为什么会变成 24 个。

用法：
    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 --mode right
    python -X utf8 05_langgraph_paramlizm_demo2.py --stream    # 再用 stream 看增量/快照
    python -X utf8 05_langgraph_paramlizm_demo2.py --no-sleep  # 去掉 sleep，跑得快点

配套文档：04_并行与reducer说明.md
"""

import argparse
import time
from operator import add
from typing import Annotated, List, TypedDict

from langgraph.graph import END, START, StateGraph

# 是否真的 sleep（--no-sleep 关掉）
SLEEP = True
# reducer 被调用的次数，用来给日志编号
_reduce_calls = 0


def cook(seconds: float) -> None:
    """模拟做菜耗时。"""
    if SLEEP:
        time.sleep(seconds)


def traced_add(old: List[str], new: List[str]) -> List[str]:
    """带日志的 reducer。

    reducer 的签名永远是 (通道现在的值, 节点返回的值) -> 通道的新值。
    这里除了照常拼接，还把两边的内容和长度打出来 —— 这就是"追踪 menu"的关键。
    注意 reducer 是**每个节点的更新来一次**，不是一次收到所有并行结果。
    """
    global _reduce_calls
    _reduce_calls += 1
    result = add(old, new)
    print(
        f"   🔧 [reducer #{_reduce_calls}] "
        f"old({len(old)})={old}  +  new({len(new)})={new}  ->  长度 {len(result)}"
    )
    return result


class MyState(TypedDict):
    menu: Annotated[List[str], traced_add]  # 有 reducer：节点返回值是"增量"，会被追加
    all_ready: bool  # 无 reducer：节点返回值是"新值"，直接覆盖


def peek(node: str, state: MyState) -> None:
    """打印节点看到的 menu 快照，以及那个 list 对象的 id。

    并行节点打出来的 id 是同一个 —— 这说明 state 里的 list 是共享对象，
    所以 append 会直接改到通道里的数据上，绕过 reducer。
    """
    menu = state.get("menu", [])
    print(f"   👀 {node} 看到 menu({len(menu)}) = {menu}   id={id(menu)}")


# ======================================================================
# 写法一：错误写法（04 里的那份）—— append 改共享 list + return state
# ======================================================================
def wrong_start(state: MyState):
    print("👨‍🍳 [错误写法] 厨房准备开始！")
    peek("start", state)
    state["menu"] = []  # 多余：带 reducer 的通道本来就是 []
    return state


def wrong_chef_a(state: MyState):
    print("👩‍🍳 A 厨师在做：番茄炒蛋...")
    peek("chef_a", state)
    cook(2)
    state["menu"].append("番茄炒蛋")  # ① 就地改了三个节点共享的那个 list
    return state  # ② 又把改完的整份当增量交回去


def wrong_chef_b(state: MyState):
    print("👨‍🍳 B 厨师在做：宫保鸡丁...")
    peek("chef_b", state)
    cook(3)
    state["menu"].append("宫保鸡丁")
    return state


def wrong_chef_c(state: MyState):
    print("👩‍🍳 C 厨师在做：鱼香肉丝...")
    peek("chef_c", state)
    cook(1)
    state["menu"].append("鱼香肉丝")
    return state


def wrong_serve(state: MyState):
    print("\n🍽️ [错误写法] 上菜：")
    peek("serve", state)
    for dish in state["menu"]:
        print("   -", dish)
    state["all_ready"] = True
    return state  # 汇聚节点的 return state：把读到的整份 menu 再追加一遍


# ======================================================================
# 写法二：正确写法 —— 只返回自己新增的那一份，不碰 state 里的 list
# ======================================================================
def right_start(state: MyState):
    print("👨‍🍳 [正确写法] 厨房准备开始！")
    peek("start", state)
    return {}  # 什么都不写就返回空 dict


def right_chef_a(state: MyState):
    print("👩‍🍳 A 厨师在做：番茄炒蛋...")
    peek("chef_a", state)
    cook(2)
    return {"menu": ["番茄炒蛋"]}  # 只交增量，合并交给 reducer


def right_chef_b(state: MyState):
    print("👨‍🍳 B 厨师在做：宫保鸡丁...")
    peek("chef_b", state)
    cook(3)
    return {"menu": ["宫保鸡丁"]}


def right_chef_c(state: MyState):
    print("👩‍🍳 C 厨师在做：鱼香肉丝...")
    peek("chef_c", state)
    cook(1)
    return {"menu": ["鱼香肉丝"]}


def right_serve(state: MyState):
    print("\n🍽️ [正确写法] 上菜：")
    peek("serve", state)
    for dish in state["menu"]:
        print("   -", dish)
    return {"all_ready": True}  # 绝不带 menu，否则整份再追加一遍


# --- 构建 LangGraph（两种写法共用同一张图，只换节点函数）---
NODES = {
    "wrong": (wrong_start, wrong_chef_a, wrong_chef_b, wrong_chef_c, wrong_serve),
    "right": (right_start, right_chef_a, right_chef_b, right_chef_c, right_serve),
}


def build_graph(mode: str):
    start, a, b, c, serve = NODES[mode]

    graph_builder = StateGraph(MyState)
    graph_builder.add_node("start", start)
    graph_builder.add_node("chef_a", a)
    graph_builder.add_node("chef_b", b)
    graph_builder.add_node("chef_c", c)
    graph_builder.add_node("serve", serve)

    # 启动 → 并行三个厨师
    graph_builder.add_edge(START, "start")
    graph_builder.add_edge("start", "chef_a")
    graph_builder.add_edge("start", "chef_b")
    graph_builder.add_edge("start", "chef_c")

    # 三个厨师汇总到 serve（fan-in）
    graph_builder.add_edge(["chef_a", "chef_b", "chef_c"], "serve")
    graph_builder.add_edge("serve", END)

    return graph_builder.compile()


# --- 两种查看方式 ---
def run_invoke(mode: str) -> List[str]:
    """普通 invoke：靠 reducer 日志看 menu 的每一次合并。"""
    global _reduce_calls
    _reduce_calls = 0

    title = "错误写法：append + return state" if mode == "wrong" else "正确写法：只返回增量"
    print("\n" + "=" * 70)
    print(f"  {title}   (mode={mode})")
    print("=" * 70)

    result = build_graph(mode).invoke({})
    menu = result["menu"]
    print(f"\n📊 reducer 共被调用 {_reduce_calls} 次，最终 menu 长度 = {len(menu)}（期望 3）")
    print(f"   最终结果: {result}")
    return menu


def run_stream(mode: str) -> None:
    """stream 方式：把"节点交了什么"和"通道变成了什么"分开看。

    stream_mode="updates"（默认）：每个节点跑完 yield 一次 {节点名: 该节点的返回值}。
        看到的是**增量**，也就是交给 reducer 的那个 new。
        错误写法在这里最扎眼 —— chef_a 交回来的不是 1 个菜而是一整份。
    stream_mode="values"：通道值变化后 yield 一次完整快照。
        看到的是**合并后的结果**，也就是 reducer 返回的 old。
        并行的三个厨师合并完才给一份，所以快照数比节点数少。
    """
    global _reduce_calls

    print("\n" + "=" * 70)
    print(f"  stream_mode='updates'：每个节点交回的增量 (mode={mode})")
    print("=" * 70)
    _reduce_calls = 0
    for i, chunk in enumerate(build_graph(mode).stream({}), start=1):
        for node_name, update in chunk.items():
            menu = update.get("menu") if isinstance(update, dict) else None
            n = len(menu) if isinstance(menu, list) else "-"
            print(f"\n── 更新 {i} ── {node_name} 交回 menu({n}) = {menu}")

    print("\n" + "=" * 70)
    print(f"  stream_mode='values'：合并后的通道快照 (mode={mode})")
    print("=" * 70)
    _reduce_calls = 0
    for i, snap in enumerate(build_graph(mode).stream({}, stream_mode="values")):
        menu = snap.get("menu", [])
        print(f"\n── 快照 {i} ── menu({len(menu)}) = {menu}  all_ready={snap.get('all_ready')}")


# --- 运行 ---
def main() -> None:
    global SLEEP

    ap = argparse.ArgumentParser(description="LangGraph 并行 + reducer 追踪 demo")
    ap.add_argument(
        "--mode",
        choices=["wrong", "right", "both"],
        default="both",
        help="跑哪种写法，默认两种都跑做对比",
    )
    ap.add_argument("--stream", action="store_true", help="额外用 stream 看增量和快照")
    ap.add_argument("--no-sleep", action="store_true", help="去掉模拟耗时，跑得快点")
    args = ap.parse_args()

    SLEEP = not args.no_sleep
    modes = ["wrong", "right"] if args.mode == "both" else [args.mode]

    menus = {m: run_invoke(m) for m in modes}

    if args.stream:
        for m in modes:
            run_stream(m)

    if args.mode == "both":
        print("\n" + "=" * 70)
        print("  对比")
        print("=" * 70)
        print(f"  错误写法 menu 长度 = {len(menus['wrong'])}   (3 个菜被写了 8 遍)")
        print(f"  正确写法 menu 长度 = {len(menus['right'])}   ✅")
        print("\n  两级翻倍：")
        print("    3 → 12  三次 append 改共享 list，又被三次 return state 各追加 3 条")
        print("    12 → 24 serve 的 return state 把读到的 12 条整份再追加一遍")
        print("\n  正确写法的 menu 顺序是 a/b/c 的声明顺序，不是完成时间顺序")
        print("  （c 只睡 1 秒却排最后）—— add 按 LangGraph 处理更新的顺序拼接。")


if __name__ == "__main__":
    main()

import time
from typing import TypedDict, List, Annotated

from langgraph.graph import StateGraph, START, END

def my_merge(a, b):
    # 这里定义合并逻辑
    return b

# 定义状态结构
class MyState(TypedDict):
    menu: Annotated[List[str], my_merge]
    all_ready: bool


# --- 节点定义 ---
def start_cooking(state: MyState):
    print("👨‍🍳 厨房准备开始！")
    # 初始化共享菜单列表
    state["menu"] = []
    return state


def chef_a(state: MyState):
    print("👩‍🍳 A 厨师在做：番茄炒蛋...")
    time.sleep(2)
    # 向共享 list 添加菜品
    state["menu"].append("番茄炒蛋")
    return state


def chef_b(state: MyState):
    print("👨‍🍳 B 厨师在做：宫保鸡丁...")
    time.sleep(3)
    state["menu"].append("宫保鸡丁")
    return state


def chef_c(state: MyState):
    print("👩‍🍳 C 厨师在做：鱼香肉丝...")
    time.sleep(1)
    state["menu"].append("鱼香肉丝")
    return state


def serve_dish(state: MyState):
    print("\n🍽️ 所有菜都准备好了！菜单如下：")
    for dish in state["menu"]:
        print(" -", dish)
    state["all_ready"] = True
    return state


# --- 构建 LangGraph ---
def build_graph():
    graph_builder = StateGraph(MyState)
    graph_builder.add_node("start", start_cooking)
    graph_builder.add_node("chef_a", chef_a)
    graph_builder.add_node("chef_b", chef_b)
    graph_builder.add_node("chef_c", chef_c)
    graph_builder.add_node("serve", serve_dish)

    # 启动 → 并行三个厨师
    graph_builder.add_edge(START, "start")
    graph_builder.add_edge("start", "chef_a")
    graph_builder.add_edge("start", "chef_b")
    graph_builder.add_edge("start", "chef_c")

    # 三个厨师汇总到 serve
    graph_builder.add_edge(["chef_a", "chef_b", "chef_c"], "serve")

    graph_builder.add_edge("serve", END)

    return graph_builder.compile()


# --- 运行 ---
if __name__ == "__main__":
    graph = build_graph()
    result = graph.invoke({})
    print("\n最终结果:", result)

# 自定义 reducer：`return b` 为什么是个陷阱

配套代码：

- `06_langgraph_paramlizm_demo3.py` —— 本文分析的对象，跑出来"看着对了"的那份
- `04_并行与reducer说明.md` —— 先读这篇，reducer 是什么、`add` 版为什么翻倍到 24
- `05_langgraph_paramlizm_demo2.py` —— 追踪工具，本文的日志都是同样手法打出来的

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

## 一句话结论

`06` 把 reducer 从 `add` 换成 `my_merge(a, b): return b`，输出从 24 条变回 3 条 —— **但这不是修好了 bug，是换了个 reducer 把症状盖住了。** `my_merge` 在这份代码里是个空操作，那 3 个菜完全是 `append` 副作用的产物。代价是：顺序变得不确定、节点里埋了竞态、而且一旦有人把节点改成规范写法就立刻静默丢数据。

## 现象：它确实输出了 3 条

```python
def my_merge(a, b):
    # 这里定义合并逻辑
    return b

class MyState(TypedDict):
    menu: Annotated[List[str], my_merge]
    all_ready: bool
```

节点还是 `04` 那份错误写法（`append` 改列表 + `return state`），实跑：

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

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

数量对了。但**顺序不对** —— 鱼香肉丝排在最前面，而它是 `chef_c` 做的、声明在最后。这是第一个信号，说明结果不是按合并规则算出来的。

## 一、`my_merge` 是个空操作

reducer 的契约是 `(通道现在的值, 节点返回的值) -> 通道的新值`。`return b` 意思是"丢掉 old，只要 new"，也就是**覆盖** —— 语义上等价于根本不写 reducer（无 reducer 字段的默认行为就是覆盖）。

但这份代码里连覆盖都没发生。给 `my_merge` 加上日志，把两个参数的 `id()` 也打出来：

```python
def my_merge(a, b):
    print(f"[my_merge] old={a} (id={id(a)})  new={b} (id={id(b)})")
    return b
```

实测：

```
[my_merge] old=[] (id=1259944948544)  new=[] (id=1259940762688)
a  id(menu)= 1259940762688
b  id(menu)= 1259940762688
c  id(menu)= 1259940762688
[my_merge] old=['鱼香肉丝','番茄炒蛋','宫保鸡丁'] (id=1259940762688)
           new=['鱼香肉丝','番茄炒蛋','宫保鸡丁'] (id=1259940762688)
[my_merge] old=[同上] (id=1259940762688)  new=[同上] (id=1259940762688)
[my_merge] old=[同上] (id=1259940762688)  new=[同上] (id=1259940762688)
[my_merge] old=[同上] (id=1259940762688)  new=[同上] (id=1259940762688)
```

**`old` 和 `new` 的 id 完全相同 —— 它们是同一个对象。** 原因是节点 `append` 改的就是通道里那个 list，然后 `return state` 又把这个同一个 list 原样交回来。于是 `a is b` 成立，`return b` 返回的就是通道自己已经有的东西。

推论：`return b` 和 `return a` 应该完全等效。实测确认了：

| reducer | 结果 |
|---|---|
| `lambda a, b: b` | 3 条 `['鱼香肉丝','番茄炒蛋','宫保鸡丁']` |
| `lambda a, b: a` | 3 条 `['鱼香肉丝','番茄炒蛋','宫保鸡丁']` |

`return a` 是连"新值"都不看的 reducer，居然也能得到 3 个菜。**这就证明了那 3 个菜跟 reducer 无关，纯粹是 `append` 的副作用。**

## 二、和 `04` 的关系：盖住症状，不是修复

`04` 的错误写法犯了两件事：

1. `state["menu"].append(...)` —— 就地改了三个并行节点共享的那个 list
2. `return state` —— 又把改完的整份列表当增量交回去

`add` 作为 reducer 时，错误 ② 导致累加，逐级翻倍到 24。换成 `my_merge` 后，**只是让错误 ② 不再造成翻倍**（覆盖不累加），错误 ① 一个字都没动。

数字上确实从 24 变成 3，但 3 这个正确结果是错误 ① 凑出来的，不是任何一条合并规则算出来的。

## 三、四×二对照表

把 reducer 和节点写法两个维度交叉跑一遍（实测，sleep 缩短为 0.1~0.3 秒）：

| reducer | 节点写法 | 结果 | 评价 |
|---|---|---|---|
| `add` | append + return state | 12 条 | `04` 的 bug，翻倍 |
| `add` | 只返回增量 | **3 条 `['番茄炒蛋','宫保鸡丁','鱼香肉丝']`** | ✅ 唯一正确 |
| `my_merge`(`return b`) | append + return state | 3 条 `['鱼香肉丝',...]` | `06` —— 巧合，顺序不定 |
| `my_merge`(`return b`) | 只返回增量 | 1 条 `['鱼香肉丝']` | ❌ 静默丢数据 |
| `return a` | append + return state | 3 条 `['鱼香肉丝',...]` | reducer 完全没参与 |
| `return a` | 只返回增量 | 0 条 `[]` | ❌ 全丢 |

（上表 `add` + append 那行是 12 而非 24，因为测试里的 `serve` 用的是规范写法 `return {"all_ready": True}`，少了第二级翻倍。`04` 的 `serve` 也 `return state`，所以是 24。）

只有第二行是对的：**合并语义用 `add`，节点交增量。** 其他五行要么数量错，要么是靠副作用凑对的。

## 四、三个实际后果

### 1. 顺序变成不确定的

`06` 的输出是 `['鱼香肉丝', '番茄炒蛋', '宫保鸡丁']`，因为 `chef_c` 只 `sleep(1)` 最先跑完，它的 `append` 就落在最前面。**菜品顺序由线程完成的先后决定**，改一下 sleep 时长顺序就变。

规范写法不受这个影响 —— `add` 按 LangGraph 处理节点更新的顺序拼接，与谁先跑完无关，稳定是声明顺序 `['番茄炒蛋','宫保鸡丁','鱼香肉丝']`。`05_langgraph_paramlizm_demo2.py` 加 `--no-sleep` 前后对比就能看到这个差别：错误写法的顺序会变，规范写法不会。

### 2. 节点里埋了竞态

三个线程同时改一个 list，现在侥幸没丢数据，只是因为 CPython 的 `list.append` 恰好是原子操作。节点里只要换成下面任何一种写法，就是标准的 read-modify-write 竞态：

```python
state["menu"] = state["menu"] + ["番茄炒蛋"]   # 读 → 建新 list → 写回
state["menu"] += ["番茄炒蛋"]
current = state["menu"]; ...; current.append(x)  # 中间隔了别的操作
```

这类 bug 不报错，只是偶尔少一个菜，且随机器负载变化。**依赖 `append` 的原子性来保证正确性，是不该写进教学代码的假设。**

### 3. 换成规范写法立刻丢数据 —— 最危险的一点

`my_merge` 和"节点交增量"这个正确习惯是**互斥**的。同一个 `my_merge`，节点改成规范写法：

```
[my_merge] old=[]            new=['番茄炒蛋']  -> ['番茄炒蛋']
[my_merge] old=['番茄炒蛋']   new=['宫保鸡丁']  -> ['宫保鸡丁']    ← 番茄炒蛋没了
[my_merge] old=['宫保鸡丁']   new=['鱼香肉丝']  -> ['鱼香肉丝']

结果: {'menu': ['鱼香肉丝']}
```

这次 `old` 和 `new` 是不同对象了，覆盖语义真正生效，于是每个厨师都把前一个的菜盖掉，只剩最后一个。**不报错，只是少了两个菜。**

这就是它最坏的地方：把一个坏习惯（改共享对象）变成了正确性的**前提**。后来的人按规范改代码，反而把功能改坏了 —— 而且没有任何报错提示他。

## 五、规范写法

并行追加就用 `add`，节点交增量：

```python
from operator import add

class MyState(TypedDict):
    menu: Annotated[List[str], add]      # 合并语义：拼接
    all_ready: bool                      # 无 reducer：覆盖

def chef_a(state: MyState):
    return {"menu": ["番茄炒蛋"]}         # 不 append，不 return state
```

## 六、自定义 reducer 什么时候是对的

自定义 reducer 完全正当，只要它**真的在写合并逻辑**。几个实测可用的例子：

```python
def merge_unique(old: List[str], new: List[str]) -> List[str]:
    """拼接并去重，保留首次出现的顺序。"""
    seen = set(old)
    return old + [x for x in new if not (x in seen or seen.add(x))]

def keep_max(old: int, new: int) -> int:
    """并行打分，取最高的那个。"""
    return max(old, new)

def merge_dict(old: dict, new: dict) -> dict:
    """浅合并，后来的覆盖同名 key。"""
    return {**old, **new}
```

三个节点各交回一份，实测：

```python
def a(s): return {"menu": ["番茄炒蛋", "米饭"], "best": 3, "tags": {"a": 1}}
def b(s): return {"menu": ["宫保鸡丁", "米饭"], "best": 7, "tags": {"b": 2}}
def c(s): return {"menu": ["米饭"],            "best": 5, "tags": {"c": 3}}

merge_unique -> ['番茄炒蛋', '米饭', '宫保鸡丁']    # 三份里的"米饭"合成一个
keep_max     -> 7
merge_dict   -> {'a': 1, 'b': 2, 'c': 3}
```

顺带一个好消息：**不需要给带 reducer 的字段准备初始值。** LangGraph 会按类型注解推出一个空值当通道初值，`keep_max` 第一次被调用时 `old` 就是 `0`（`int` → `0`，`str` → `''`，`List` → `[]`）：

```
[keep_max] old=0 (type=int)  new=3
[keep_max] old=3             new=7
结果: {'best': 7}
```

所以 `invoke({})` 就够了，也不需要在起始节点里写 `state["menu"] = []`。

## 七、判断标准

一条就够用：

> **reducer 里如果没同时用到 `old` 和 `new` 两个参数，就该怀疑它写错了。**

- `return b` —— 没用 `old`，那是覆盖，不是合并。这种情况下 reducer 可以直接删掉，效果一样（而且删掉之后并行写同一字段会正常报 `InvalidUpdateError`，比静默丢数据好）
- `return a` —— 没用 `new`，节点的输出被整个扔掉
- `return old + new` / `max(old, new)` / `{**old, **new}` —— 两个都用了，这才是合并

配套的自查（详见 `04_并行与reducer说明.md` 第七节）：节点里不出现 `state[...].append(...)`，不 `return state`，只返回自己输出 key 组成的 dict。

## 八、三份 demo 的定位

| 文件 | reducer | 节点写法 | 结果 | 用途 |
|---|---|---|---|---|
| `04_langgraph_paramlizm_demo.py` | `add` | append + return state | 24 条 | 反例：翻倍 |
| `05_langgraph_paramlizm_demo2.py` | `traced_add` | 两种都有，`--mode` 切换 | 24 / 3 | 追踪工具 + 正例 |
| `06_langgraph_paramlizm_demo3.py` | `my_merge` | append + return state | 3 条 | 反例：靠副作用凑对 |

`06` 比 `04` 更值得警惕：`04` 数字明显不对，一眼能看出有问题；`06` 数字是对的，藏着顺序不定、竞态、以及"改成规范写法就丢数据"三个隐患。**测试通过不等于写法正确。**