自定义数据整合方式Reducer

LangGraph 包内公开预构建的归约器（reducer）只有 add_messages。无 Annotated 时走默认覆盖；要绕过已注册的 reducer 用 Overwrite。并行 worker 写计数器、标签集合或嵌套 dict 时，没有现成内置函数，需自定义 reducer(left, right) -> merged。

段末注释：reducer = 字段级合并函数 (当前值, 本次更新) → 新值。operator.add 是 Python 标准库，官方示例常用，但不是 langgraph 包导出的 API。

社区方案：消息字段只用官方 add_messages（Reducers）；其余字段用 operator.add 或自写纯函数。不要期待包里还有 dict_merge / set_union。风险：MessageGraph 自 1.0 弃用；实验项 _messages_delta_reducer 未进公开 API。

---

1. 定位

维度	内容
角色	字段级合并策略，支撑 Send 并行与多写
输入 → 输出	(旧值, 新 partial) → 合并值
核心 API	add_messages、Overwrite、Annotated[T, reducer_fn]
依赖 LangChain	add_messages 消费 AnyMessage / RemoveMessage

---

2. 图拓扑

节点表

节点名	职责	读 State	写 State
w1	worker 1	—	total, tags
w2	worker 2	—	total, tags
sum_label	格式化	total, tags	label

边表

源	目标	类型
START	w1, w2	固定并行（简化：顺序边演示 reducer）
w1	w2	固定
w2	sum_label	固定
sum_label	END	固定

为清晰演示 reducer，本篇用顺序两节点模拟两次写入；Send 并行场景合并规则相同。

---

3. invoke 生命周期与 superstep

Superstep 1: w1 写 total=10, tags=["a"]
Superstep 2: w2 写 total=5,  tags=["b"]
             → sum_reducer: total=15
             → tag_reducer: tags=["a","b"]（去重）
Superstep 3: sum_label → END

---

4. 原理

4.1 reducer 签名

def sum_reducer(left: int, right: int) -> int:
    return left + right

LangGraph 对 Annotated 字段在合并 update 时调用。

4.2 列表 vs 集合

去重用 set 或自定义逻辑；注意 JSON 序列化 checkpoint 时 set 可能需转 list。

4.3 与 operator.add

Annotated[list, operator.add] 等价于列表拼接 reducer。operator.add 来自标准库，对 list 拼接、int/float 相加、str 拼接都成立。

4.4 LangGraph 本身支持的 reducer

对照 langgraph 1.2.x 源码（langgraph.graph.message / langgraph.types）：名为 reducer、由本包导出的公开函数只有 add_messages。其余是运行时默认行为、绕过包装，或标准库/用户函数。

名称	来源	用法	合并语义	适用字段	注意
（默认，无 Annotated）	运行时	字段不写 reducer	last-write-wins：右值覆盖左值	status 等标量	并行多写只留最后一次
add_messages	langgraph.graph.message	Annotated[list[AnyMessage], add_messages]	无相同 id 则追加；相同 id 则替换；可反序列化 dict / tuple	对话 messages	包内唯一公开预构建 reducer
add_messages(format="langchain-openai")	同上（偏函数）	Annotated[list, add_messages(format="langchain-openai")]	先按 id 合并，再转成 OpenAI 文本/图片块	多模态消息	需 langchain-core>=0.3.11
MessagesState	langgraph.graph.message	子类或直接当 State	预置 messages: Annotated[list[AnyMessage], add_messages]	对话图骨架	不是 reducer，是套装 schema
Overwrite(value)	langgraph.types	节点返回 {"k": Overwrite([...])} 或 {"k": {"__overwrite__": [...]}}	绕过该字段 reducer，整表替换	任意已挂 reducer 的字段	不是 reducer；同一 superstep 同一 key 只能一个 Overwrite，否则 InvalidUpdateError
RemoveMessage(id)	langchain_core.messages	{"messages": [RemoveMessage(id="x")]}	从列表删掉该 id	仅配合 add_messages	删除不存在的 id → ValueError
REMOVE_ALL_MESSAGES	langgraph.graph.message	RemoveMessage(id=REMOVE_ALL_MESSAGES)	清空历史，只保留该标记之后的新消息	仅配合 add_messages	常量 "__remove_all__"
operator.add	Python operator	Annotated[list, operator.add]	left + right	日志 list、计数 int、拼接 str	不是 langgraph 导出；官方文档当通用加法用
自定义 (left, right)	用户	Annotated[T, my_reducer]	任意（求和、去重、deep merge）	标签、嵌套 dict	包内没有 dict_merge / set_union

不要当成包内 reducer 的：

• push_message：节点内往 messages 流式通道塞一条，不是合并函数。
• _messages_delta_reducer：实验批处理，配合 DeltaChannel，未进公开 __all__。
• MessageGraph：整图 state 就是带 add_messages 的消息列表；1.0 弃用，2.0 删除，改 StateGraph + messages 键。

from typing import Annotated, TypedDict

from langchain_core.messages import AnyMessage, RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages
from langgraph.types import Overwrite


class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


def reset(state: ChatState) -> dict:
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}


def replace_all(state: ChatState) -> dict:
    return {"messages": Overwrite([])}  # 绕过 add_messages，直接清空

---

5. 最小可运行示例

from typing import Annotated, TypedDict

from langgraph.graph import StateGraph, START, END


def sum_reducer(left: int, right: int) -> int:
    return left + (right or 0)


def tag_reducer(left: list[str], right: list[str]) -> list[str]:
    seen = set(left)
    out = list(left)
    for t in right:
        if t not in seen:
            seen.add(t)
            out.append(t)
    return out


class State(TypedDict):
    total: Annotated[int, sum_reducer]
    tags: Annotated[list[str], tag_reducer]
    label: str


def w1(_: State) -> dict:
    return {"total": 10, "tags": ["api"]}


def w2(_: State) -> dict:
    return {"total": 5, "tags": ["api", "db"]}


def sum_label(state: State) -> dict:
    return {"label": f"total={state['total']} tags={state['tags']}"}


builder = StateGraph(State)
builder.add_node("w1", w1)
builder.add_node("w2", w2)
builder.add_node("sum_label", sum_label)
builder.add_edge(START, "w1")
builder.add_edge("w1", "w2")
builder.add_edge("w2", "sum_label")
builder.add_edge("sum_label", END)

graph = builder.compile()
print(graph.invoke({"total": 0, "tags": [], "label": ""}))
# total=15, tags=['api','db']

---

6. 执行追踪

步骤	total	tags
初始	0	[]
w1 后	10	[“api”]
w2 后	15	[“api”,”db”]
sum_label	—	label 写入

---

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点 / 常用范围	配置指导
add_messages	(left, right, *, format=None)	包内唯一公开预构建 reducer；按消息 id 追加或替换	无 id 则追加；同 id 覆盖旧条；可吞 dict/tuple	messages 字段	对话/Agent 标配；勿用 operator.add 管消息
add_messages(format=...)	None / "langchain-openai"	合并后再规范成 OpenAI 块	多模态图块对齐厂商格式	含 image 的消息	需 langchain-core>=0.3.11
Overwrite(value)	包装类型	节点返回值上绕过该字段 reducer	整表替换；同拍同 key 多个 Overwrite 报错	重置 logs/messages	不是 Annotated 上的 reducer
RemoveMessage / REMOVE_ALL_MESSAGES	消息 / 常量	作为 add_messages 的 right 做删除或清空	清错 id 会 ValueError	HITL 改写历史	只对 add_messages 字段有效
operator.add	标准库	left + right：list 拼接、int 相加	不去重；消息场景会重复同 id	日志、计数	非 langgraph API
sum_reducer / tag_reducer	用户 callable	本篇示例：求和 / 去重追加	不处理 None 会 TypeError	计数、标签	checkpoint 用 list 不要用 set
无 Annotated	last-write-wins	运行时默认覆盖	并行写标量只留最后一次	status、label	单写 OK

---

7. 易踩坑

1. reducer 内改 mutable 入参：应返回新值，避免副作用。
2. 并行写无 reducer 的 int：只保留最后一次写入。
3. set 作 state 字段：checkpoint 序列化可能失败，用 list + 去重 reducer。
4. 以为 langgraph 还内置 dict/set reducer：没有。字典合并、集合并交差都要自写。
5. 用 operator.add 管 messages：同 id 会再追加一条，流式替换/HITL 改写会重复。

---

小结

• 包内公开预构建 reducer 只有 add_messages；默认覆盖、Overwrite 绕过、RemoveMessage 是配套机制。
• 计数/标签/嵌套 dict 没有内置函数，写 (left, right) -> merged。
• 消息不要用 operator.add；Send 并行前先定各字段 reducer。

---

参考链接

• Reducers
• add_messages
• Overwrite
• Send API