接着上一篇继续啃 LangGraph。这次的主题是我踩过最早、也最容易踩的一个坑。
写 LangGraph 的人应该都遇过这个：
def collect(state):
logs = state["logs"]
logs.append("我干了点事") # 明明 append 了
return {"logs": logs}
# 三个节点都这么写，跑完 logs 里只有一条
或者更隐蔽的版本：三个节点各自 return {"logs": ["..."]}，跑完只剩最后一条。
原因很简单：LangGraph 合并状态时，每个键的默认规则是"覆盖"，不是"累加"。
这个开关叫 Reducer。而我一开始理解错的地方是 —— 它写在 State 上，不写在节点里。
一个实验：节点代码一行不改，结果完全不同
三个评审节点，张三 70 分、李四 85 分、王五 60 分，各留一条意见。
下面三个版本，节点代码完全复用、一个字没改，只改了 State 的注解：
# 版本 A：什么都不加
class StatePlain(TypedDict):
reviews: list[str]
score: int
# → reviews = ['王五：风险评估太薄'] score = 60
三条意见交上去，只剩最后一条。这就是"我明明写了怎么没了"的现场。
# 版本 B：给 reviews 挂上累加
class StateAdd(TypedDict):
reviews: Annotated[list[str], operator.add]
score: int
# → reviews = [张三, 李四, 王五] score = 60
意见齐了。但注意 score 还是 60 —— 想要的是最高分 85。
这是我觉得整件事里最值得记的地方：只给"看起来该累加"的字段加了 reducer，
漏掉的字段还在默默用覆盖语义，跑出来是个不报错、但业务错误的结果。60 分和 85 分，
在评审系统里就是"驳回"和"通过"的差别。
# 版本 C：score 也挂一个
def keep_max(left: int, right: int) -> int:
return max(left, right)
class StateMax(TypedDict):
reviews: Annotated[list[str], operator.add]
score: Annotated[int, keep_max]
# → reviews = [张三, 李四, 王五] score = 85 ✅
一个七个字符的函数，把"最后一个人说了算"改成了"取最高分"。节点代码一行没动。
Annotated[类型, reducer] 的语义：类型给人和 IDE 看，第二个参数给引擎看。
引擎每次合并这个键，就调一次 reducer(旧值, 新值)。
所以我现在读一张 LangGraph 图，先读 State 定义，再读节点 ——
State 定义就是这张图的数据契约。写 State 时给每个字段问一句：
这个键被多个节点写的时候，我要什么行为？
reducer 什么时候被调用？我打印了一下
文档里不太强调，但这两件事会咬人。我写了个会打印参数的 reducer，初始输入 60，
三个节点依次返回 70、40、和"不返回这个键"：
↳ reducer 被调用：left=0 right=60 → 60 ← 初始输入也要过 reducer！
↳ reducer 被调用：left=60 right=70 → 70
↳ reducer 被调用：left=70 right=40 → 70
-
初始输入也要过 reducer，而且第一次的
left是引擎按注解类型推的空值（int→0）。
所以def r(l, r): return l["k"] + r["k"]这种直接下标访问的 reducer，第一次就炸。 -
节点没返回这个键，reducer 就不调用（一共 3 次不是 4 次）。
它不是"每步都跑"，是"这个键有人提交才跑"。
顺带一个好消息：带 reducer 的键，输入不给也不会 KeyError ——
引擎会按注解类型推个空值（list→[]、int→0、dict→{}）。
没 reducer 的键不给初始值，节点一读就 KeyError。
这解释了一个我一直没想明白的现象：为什么 messages 键从来不用先传个空列表进去。
消息列表千万别用 operator.add
消息看起来就是个往后追加的 list，我一开始顺手就写了 Annotated[list, operator.add]。这是错的。
传一条 AIMessage(content="改好了", id="a1")，而历史里已经有一条 id="a1"：
add_messages ： [('human','你好','h1'), ('ai','改好了','a1')] ← 2 条，原地更新
operator.add ： [('human','你好','h1'), ('ai','草稿','a1'), ('ai','改好了','a1')] ← 3 条，两个 a1
add_messages 认 id。 它的完整行为：
| 输入 | 行为 |
|---|---|
| 新 id 的消息 | 追加 |
| 已存在 id 的消息 | 原地替换那一条 |
RemoveMessage(id=...) |
删掉那一条 |
("user", "hi") 元组 |
自动转成 HumanMessage
|
| 没写 id | 自动补 uuid |
"按 id 原地更新"不是锦上添花，是必需的：流式输出要不断更新同一条 AI 消息、
工具结果要回填、消息内容要修订。用 operator.add 的话这些全变成"堆重复消息"，
而且不报错 —— 只会在某天发现历史里怎么有两条一样的。
顺带说，MessagesState 没有魔法，打印一下就知道：
MessagesState.__annotations__
# {'messages': Annotated[list[AnyMessage], add_messages]}
就一个键。要加业务字段就继承它：
class TicketState(MessagesState):
ticket_id: str # 覆盖语义
tools_used: Annotated[list[str], operator.add] # 想累加就自己挂
四个比 operator.add 有用的自定义 reducer
# 合并字典：多个节点各写几个字段，不互相覆盖
def merge(left, right): return {**left, **right}
# 累加去重且保序：多路检索召回同一篇文档
def add_unique(left, right):
out = list(left)
for x in right:
if x not in out: out.append(x)
return out
# 只留最后 3 条：上下文长度护栏
def keep_last_3(left, right): return (left + right)[-3:]
# 取最大：评分、优先级、置信度
def keep_max(left, right): return max(left, right)
第三个是我觉得最有意思的：把上下文护栏写进 State 的 reducer，比在每个节点里判断长度可靠得多。
节点会越写越多，总有一天有人忘了加判断；reducer 是这个键的唯一入口，漏不掉。
写 reducer 的三条纪律（我自己踩过前两条）：必须是纯函数（不改 left、不做 IO，因为重试和回放时会被重复调用）、
必须能处理 left 为空值、别在里面写业务逻辑（它只管"怎么合并"，不管"该不该合并"）。
两个坑，一个响一个哑
响的那个 —— 两个并行节点写同一个键、没加 reducer：
InvalidUpdateError: At key 'x': Can receive only one value per step.
Use an Annotated key to handle multiple values.
同一个超步里一个键收到两份提交，引擎不知道听谁的。串行不会遇到（一步一份），并行必然撞上。
reducer 是并行的前置条件。
哑的那个 —— 加了 reducer 之后，"顺手返回整个 state"这个坏习惯开始静默出错：
def lazy(state):
s = dict(state) # 只想改 n，把整个 state 交回去
s["n"] = 1
return s
# 结果：{'log': ['good 干的', 'good 干的'], 'n': 1} ← log 凭空多一条
因为它把 log 的旧值原样交了回去，而 log 的 reducer 是累加，旧值又被追加一遍。
在真实项目里，这表现为「消息列表莫名翻倍」，非常难查。
上一篇里"节点只返回要改的键"这条纪律，在有 reducer 之后才真正开始收费。
顺带一个我之前不知道的功能：把内部字段藏起来
上一篇里 invoke 把 state 里 7 个键全倒给了调用方，其中 draft、error 这种中间产物本不该外泄：
g = StateGraph(Overall, input_schema=In, output_schema=Out)
app.invoke({"question": "多少钱"}) → {'answer': '定稿:多少钱'} ← 只有 answer
app.invoke({"question": "q", "draft": "我塞的脏数据"}) → 节点只看到 {'question': 'q'} ← 入口就被过滤
输入被过滤，输出被裁剪。 这才像是能对外暴露的 API 形态。
还能更细 —— add_node("narrow", fn, input_schema=In) 让单个节点只看见 state 的一部分：
拿不到的字段就不可能被它误改。
这次的收获小结
- 默认是覆盖，不是累加；reducer 写在 State 上，不写在节点里
- 读图先读 State 定义 —— 它是数据契约
- 逐字段问「多个节点写它时我要什么」，漏掉的字段会静默出错
- 消息永远用
add_messages（它认 id，能原地更新），别用operator.add - reducer 必须是纯函数、必须能处理空
left - reducer 是并行的前置条件；加了 reducer 之后，"返回整个 state"会让数据翻倍
接下来想搞明白的
到这里图还是一条直线，每条边都无条件 —— 上一篇那个「校验失败了还照样算钱」的问题还没解决。
而真实业务全是分支：VIP 走快速通道、金额超阈值要审批。更关键的是——
Agent 的本质就是一个环：调模型 → 要调工具吗 → 调完回到模型 → 再判断。
没有条件边，就没有 Agent。 下一篇就啃这个。
我在从零重学 LangGraph，边学边把笔记整理出来，这是第 2 篇。
环境：Python 3.13 + LangGraph 1.2.10，文中所有输出都是实跑出来的。
有理解错的地方欢迎指出，也欢迎交流你踩过的坑。
