LangGraph 状态与 reducer

第 2 篇收尾留了一句话：黑板字段的默认合并语义是覆盖，str 这类单值字段刚好合适，换成消息列表就是灾难——后一个节点会把前一个节点攒下的对话历史整个顶掉。当时只让 add_messages 露了个名字，本篇把这个坑正面填完：reducer 是什么、标准写法怎么写、内置规则有哪些、自定义怎么下手，以及并发写撞车怎么破。

先把术语钉死。reducer 一词借自 MapReduce，在 LangGraph 里指挂在 State 字段上的合并函数：节点照旧返回增量，框架调用 reducer 把旧值与增量算成新值写回黑板。版本口径与全系列一致：langgraph 1.2.x，PyPI 独立发布，示例代码只依赖 langgraph 及其自带的 langchain-core。全文按"翻车现场 → Annotated 写法 → add_messages 规则 → 自定义 reducer → State 载体取舍 → 并发写冲突 → 实战 → 总结"展开。

1、默认覆盖的翻车现场

先把翻车完整复现一遍。两节点小图：ask 抛出一个翻译请求，translate 给出译文，messages 字段不挂任何 reducer——正是第 2 篇的标准写法。代码如下：

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

class State(TypedDict):
    messages: list                    # 没有 reducer：默认覆盖

def ask(state: State) -> dict:
    return {"messages": [("user", "把这句话翻译成英文")]}

def translate(state: State) -> dict:
    return {"messages": [("assistant", "Translate this sentence")]}

builder = StateGraph(State)
builder.add_node("ask", ask)
builder.add_node("translate", translate)
builder.add_edge(START, "ask")
builder.add_edge("ask", "translate")
builder.add_edge("translate", END)
print(builder.compile().invoke({})["messages"])

输出如下：

[('assistant', 'Translate this sentence')]

问题一目了然：用户的话被译文顶掉了。问句进黑板是第一次写，译文进黑板是第二次写，覆盖语义下后者直接替换前者，messages 里只剩一条。这里节点返回的还是元组，换成真模型的消息一个道理——下一轮模型读黑板，对话里根本没有用户的提问，答非所问从此开始。

根子是语义错位：消息列表的天然语义是"追加历史"，覆盖的天然语义是"以最新为准"。出路有两条。要么每个节点自己拼整份列表再返回（state["messages"] + [新消息]），全量搬运又蠢又容易漏；要么告诉框架"这个字段按追加合并"——这就是 reducer，Graph API 文档对整套合并规则有正式定义。

2、Annotated 挂载 reducer

LangGraph 没有为合并函数另造配置项，而是借用 Python 类型注解的标准扩展：Annotated[类型, reducer]，把合并函数直接写进字段的类型里。改造如下：

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 挂上合并函数

def ask(state: State) -> dict:
    return {"messages": [("user", "把这句话翻译成英文")]}

def translate(state: State) -> dict:
    return {"messages": [("assistant", "Translate this sentence")]}

解读：两个节点函数一个字没改，改的只有 State 里一行的类型注解。重跑第 1 节的建图代码，输出变成两条消息——先问后答，历史完整。原理是节点返回增量后，框架执行 add_messages(旧列表, 增量列表)，把算出的结果写回黑板。

节点写增量，reducer 定语义，框架管合并。

两个工程细节。其一，add_messages 的家在 langgraph.graph.message 模块，官方 API 参考的示例同时也从 langgraph.graph 直接导入，两条路径都有效，团队里统一一条即可。其二，State 里若只有 messages 一个字段，可以直接用内置的 MessagesState——它就是 Annotated[list, add_messages] 这一行的成品，第 1 篇的 Agent 循环示例用的正是它；字段一多就回到自定义，黑板不会只为消息存在。

3、add_messages 的合并规则

add_messages 不是简单的列表相加，完整规则是按消息 ID 合并，官方 API 参考给出三条判定，整理如下：

最值得咀嚼的是第二条：同 ID 不是报错、不是重复，而是替换。直接调用它验证：

from langchain_core.messages import HumanMessage
from langgraph.graph.message import add_messages

old = [HumanMessage(content="初稿：今天聊状态管理", id="m1")]
new = [HumanMessage(content="定稿：今天聊 reducer", id="m1")]

merged = add_messages(old, new)
print(len(merged))           # 1：原位替换，没有追加
print(merged[0].content)     # 定稿：今天聊 reducer

解读：两条消息的 ID 同为 m1，合并结果只有一条，内容取新值。这个规则在生产里非常值钱——改写历史而不是追加更正。典型场景：某个节点发现黑板里上一条总结说错了话，与其追加一条"更正：刚才说错了"让上下文越来越脏，不如按原 ID 返回一条修正版，历史干净，token 也省。第三条规则里的 RemoveMessage 专司删除，对话历史裁剪时它会登场，第 7 篇持久化再见。顺带说明：HumanMessage 来自 langchain-core，装 langgraph 时会作为依赖一并装上；("user", "...") 这种元组写法也会被 add_messages 自动转成消息对象。

4、自定义 reducer 两例

add_messages 只管消息，黑板上的其他字段各有合并诉求。reducer 本身就是普通函数——接收旧值与增量两个参数、返回新值，谁都能写。看两个最常用的例子。

例一，数字累加。审校轮数、重试次数这类计数字段，诉求是"每次加一"而不是"覆盖成最新"，Python 标准库 operator 里的 add 现成可用：

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

class State(TypedDict):
    rounds: Annotated[int, operator.add]   # 整数相加

def review(state: State) -> dict:
    return {"rounds": 1}                   # 每审一轮贡献 1

builder = StateGraph(State)
builder.add_node("review", review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
print(builder.compile().invoke({"rounds": 0}))

输出为 {'rounds': 1}。解读：operator.add 等价于 lambda a, b: a + b，挂在 int 字段上就是累加器，图里有几次写 rounds 的增量，就累加几次，第 7 节实战里它管审校轮数。

例二，按值去重合并。并行分支各自产出标签，汇合时重复的只留一份，这类语义没有现成函数，手写也就六行：

from typing import Annotated, TypedDict

def merge_tags(existing: list | None, update: list | None) -> list:
    """按值去重合并：先来先到，重复不再进黑板"""
    existing, update = existing or [], update or []
    return existing + [t for t in update if t not in existing]

class State(TypedDict):
    tags: Annotated[list, merge_tags]

解读：reducer 的契约只有两条——两个参数（旧值、增量）、返回合并结果；官方示例的防御写法是参数可能为 None 时先兜底，照抄这个习惯能避开首写空值的坑。另有铁律一条：reducer 必须是纯函数——同样的输入永远返回同样的输出，且绝不能原地修改 existing。持久化时框架会重放这些函数（第 7 篇），带副作用的 reducer 会把历史状态污染掉。

5、TypedDict 与 Pydantic 取舍

State 的定义载体不止 TypedDict 一种，官方同样支持 Pydantic BaseModel 与 dataclass，reducer 在三种载体上的挂法一模一样，差别在类型之外的待遇。两种主流载体对比如下：

Pydantic 版的 State 长这样：

from pydantic import BaseModel, Field

class State(BaseModel):
    topic: str
    score: float = Field(default=0.0, ge=0, le=10)  # 默认值加区间校验
    rounds: int = 0

解读：score 声明了默认 0.0 且必须在 0 到 10 之间，invoke 传入越界值时校验当场报错，而不是等下游节点算出垃圾结果再排查；需要 reducer 的字段照旧用 Annotated[list, add_messages] 挂在类型上。取舍口径给一个：内部实验图用 TypedDict，图的输入边界用 Pydantic 挡一道，把脏数据拦在进黑板之前。本系列后续示例默认 TypedDict，遇输入校验需求再切换。

6、并发写冲突与破解

第 2 篇报错表里那条 InvalidUpdateError 埋了伏笔，这里正面拆。触发条件一句话：同一个超步里，多个节点写了同一个没有 reducer 的字段。串行图永远不会撞——前一个超步写完、后一个超步覆盖，是合法的"以最新为准"；并行分支才会撞。复现如下：

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)   # State.tags 沿用 4 节定义，不挂 reducer
builder.add_node("pick_a", lambda s: {"tags": ["AI"]})
builder.add_node("pick_b", lambda s: {"tags": ["Agent"]})
builder.add_edge(START, "pick_a")   # 同一入口扇出两路
builder.add_edge(START, "pick_b")
builder.add_edge("pick_a", END)
builder.add_edge("pick_b", END)
builder.compile().invoke({"tags": []})   # 抛 InvalidUpdateError

解读：pick_a 与 pick_b 在同一个超步被并行调度，各自带回一份 tags 更新，覆盖语义下框架无法回答"该留哪份"——与其悄悄丢数据，不如直接抛错。破解恰恰是本篇主角：给 tags 挂上第 4 节的 merge_tags，两路更新交给 reducer 汇合，冲突变协作，并行加 reducer 正是表达"分头干活、汇合结果"的标准姿势。另一条路是让各分支写不同的键、在下游节点手动汇总，第 2 篇练习二的提示说的就是它。扇出扇入的完整玩法第 4 篇专讲。

7、多轮翻译润色图实战

把三种合并语义放进同一张图。业务：一段文字先机翻，再审校；审校不通过就带着意见回炉重译，通过则收工——一张带回边的两节点循环图。黑板三个字段各配各的合并策略，选型如下：

State 与两个节点如下：

import operator
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 追加
    rounds: Annotated[int, operator.add]      # 累加
    done: bool                                # 覆盖

def draft(state: State) -> dict:
    verb = "初译" if state["rounds"] == 0 else "按意见重译"
    return {"messages": [("assistant", f"{verb}：{state['messages'][0].content}")]}

def review(state: State) -> dict:
    ok = state["rounds"] >= 1              # 演示口径：第二轮通过
    note = "定稿：术语统一，已润色" if ok else "初审：术语不统一"
    return {"messages": [("assistant", note)], "rounds": 1, "done": ok}

解读：draft 读首条用户消息与轮数，决定初译还是重译，只写 messages；review 写三样——追加一条审校意见、给 rounds 贡献 1、给出完成判断。done 不挂 reducer 是刻意的：布尔标记要的就是"最新判断覆盖旧判断"，挂累加反而出错。建图连边与运行如下：

from langgraph.graph import StateGraph, START, END

def route(state: State) -> str:
    return END if state["done"] else "draft"

builder = StateGraph(State)
builder.add_node("draft", draft)
builder.add_node("review", review)
builder.add_edge(START, "draft")
builder.add_edge("draft", "review")
builder.add_conditional_edges("review", route)   # 通过收工，不通过回炉
graph = builder.compile()
result = graph.invoke({"messages": [("user", "把这段话翻译成英文")],
                       "rounds": 0, "done": False})
print(result["rounds"], result["done"])

运行输出 2 True。解读：第一轮 review 见轮数为 0 判不通过，rounds 累加到 1；route 读 done 为 False 把流程送回 draft 重译；第二轮 review 判通过，rounds 到 2、done 翻真，route 送至 END。messages 里躺着五条消息——提问、初译、初审、重译、定稿，全程留痕。整个流程画出来是这样：

解读：一条回边表达回炉，黑板上的三种合并语义各司其职——messages 记录来龙去脉，rounds 数出两轮审校，done 始终是最新一次判断。第 1 节那个问句被顶掉的翻车现场，在这张图里被三行类型注解彻底解决。真实场景把 review 换成审校模型、通过条件换成质量评分即可，骨架不变。

8、总结

reducer 回答一个问题：节点带回增量之后，黑板按什么规则合并。没有 reducer 就是覆盖，适合单值、最新为准的字段；消息历史用 add_messages，按 ID 追加、替换、删除三态合并；计数字段用 operator.add 累加；更复杂的语义六行函数自己写，纯函数与防御空值两条纪律。载体上 TypedDict 轻快、Pydantic 带校验与默认值，输入边界交给后者。并发写同一个键，先想 reducer——冲突就变成了汇合。

练习两道。练习一：给第 4 节的 merge_tags 加上"忽略空串标签"的过滤，再用两个并行节点各写一批重复标签，验证去重与过滤同时生效。练习二：把第 7 节实战图的通过条件从轮数改成评分——黑板加一个 score: Annotated[float, max] 字段，用内置 max 当 reducer，多次评审只留最高分，review 节点按分数是否达到 8 分决定去留。对于打算继续深入的读者，建议下一站进第 4 篇——节点与边的完整控制流，并行、扇出扇入都在那里展开。