Reducer 归约器

前面多次提到 reducer——它决定状态某字段如何被多个节点更新合并。本篇彻底讲清 reducer 是什么、怎么用、什么时候需要自定义。

一、reducer 是什么

Reducer 是一个函数 (old_value, new_value) -> merged_value，决定状态某字段被节点更新时的合并策略。

graph LR
    S1[节点1 返回 step=1] --> R{Reducer add}
    S0[旧 step=0] --> R
    R --> N[新 step=1]
    S2[节点2 返回 step=1] --> R2{Reducer add}
    N --> R2
    R2 --> N2[新 step=2]

每个字段都可以挂一个 reducer：

• 没挂 reducer：默认策略是"覆盖"——新值直接替换旧值。
• 挂了 reducer：用 reducer(old, new) 合并。

二、为什么需要 reducer

设想一个对话场景：每个节点都返回新消息，希望消息历史追加而不是被覆盖。

如果默认覆盖：

class BadState(TypedDict):
    messages: list   # 没挂 reducer，覆盖

def node_a(state): return {"messages": [msg_a]}  # 历史 = [msg_a]
def node_b(state): return {"messages": [msg_b]}  # 历史 = [msg_b]，msg_a 丢了！

挂了 reducer（追加）：

class GoodState(TypedDict):
    messages: Annotated[list, add]   # 挂 operator.add

def node_a(state): return {"messages": [msg_a]}  # 历史 = [msg_a]
def node_b(state): return {"messages": [msg_b]}  # 历史 = [msg_a, msg_b] ✅

reducer 让你声明字段语义：是要覆盖、要追加、要累加，还是更复杂的合并逻辑。

三、Annotated 语法

from typing import Annotated
from operator import add

class State(TypedDict):
    counter: Annotated[int, add]              # 累加
    history: Annotated[list, add]             # 追加
    messages: Annotated[list, add_messages]   # 消息专用追加+去重
    answer: str                                # 默认覆盖

Annotated[T, reducer_fn] 的含义：

• 类型仍是 T（类型检查器眼里）。
• LangGraph 编译时读第二个元素 reducer_fn，作为该字段的更新函数。
• 第三个及之后的元数据被忽略（可写注释）。

四、内置 reducer

1. operator.add

Python 内置的 + 运算符。对 int 是加法，对 list 是拼接：

from operator import add
add(3, 5)        # 8
add([1, 2], [3]) # [1, 2, 3]

适合：计数器累加、列表追加。

2. add_messages

LangGraph 专门给消息列表设计的 reducer，不是简单追加：

• 同 id 的消息会被更新（不是追加）。
• 不同 id 的消息追加。
• 支持 RemoveMessage 删除指定 id 的消息。

from langgraph.graph.message import add_messages

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

为什么需要这个？因为流式输出时同一条消息会被多次更新（增量内容），如果用普通 add，会看到一堆同 id 的半成品消息。add_messages 让"同 id 更新"自动覆盖旧版。

from langchain_core.messages import AIMessage

# 第一条 AI 消息
msg1 = AIMessage(content="你好", id="m1")
# 流式更新：同一个 id，内容变长
msg1_updated = AIMessage(content="你好，世界", id="m1")

# add_messages 行为：
add_messages([msg1], [msg1_updated])
# 结果：[AIMessage(content="你好，世界", id="m1")]  ——更新而非追加

MessagesState 内置就是用 add_messages：

from langgraph.graph import MessagesState
# 等价于
# class MessagesState(TypedDict):
#     messages: Annotated[list, add_messages]

3. 默认覆盖（不挂 reducer）

class State(TypedDict):
    answer: str   # 没 Annotated，覆盖

适合"最后写入者获胜"的字段，如答案、最终结果。

五、自定义 reducer

写一个普通函数，签名 (old, new) -> merged：

from typing import Annotated, TypedDict

def keep_max(old: int, new: int) -> int:
    """保留较大值。"""
    return max(old, new)

class State(TypedDict):
    best_score: Annotated[int, keep_max]

或者更复杂的合并：

def merge_dict(old: dict, new: dict) -> dict:
    """dict 浅合并，new 覆盖 old。"""
    merged = dict(old or {})
    merged.update(new or {})
    return merged

class State(TypedDict):
    config: Annotated[dict, merge_dict]

注意：自定义 reducer 必须能处理 old=None 的情况——因为字段初次更新时旧值可能不存在。

def keep_max(old, new):
    if old is None:
        return new
    return max(old, new)

六、覆盖 vs 追加对比示例

完整可运行：

# reducer_demo.py
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    overwrite_field: str                 # 覆盖
    append_list: Annotated[list, add]    # 追加
    counter: Annotated[int, add]         # 累加


def node_a(state: State) -> dict:
    return {
        "overwrite_field": "A 写的内容",
        "append_list": ["A 的项"],
        "counter": 1,
    }


def node_b(state: State) -> dict:
    return {
        "overwrite_field": "B 写的内容（覆盖 A）",
        "append_list": ["B 的项"],
        "counter": 10,
    }


g = StateGraph(State)
g.add_node("a", node_a)
g.add_node("b", node_b)
g.add_edge(START, "a")
g.add_edge("a", "b")
g.add_edge("b", END)
app = g.compile()

result = app.invoke({
    "overwrite_field": "",
    "append_list": [],
    "counter": 0,
})
print(result)

输出：

{'overwrite_field': 'B 写的内容（覆盖 A）', 'append_list': ['A 的项', 'B 的项'], 'counter': 11}

观察：

• overwrite_field：A 写的"B 写的"——只保留最后值（B）。
• append_list：A 和 B 的项都在。
• counter：0 + 1 + 10 = 11。

这就是 reducer 的差异。

七、消息合并去重示例

# messages_reducer.py
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage


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


def user_turn(state: State) -> dict:
    # 模拟用户消息
    return {"messages": [HumanMessage(content="你好", id="u1")]}


def ai_turn(state: State) -> dict:
    # 第一次生成（id 相同）
    msg1 = AIMessage(content="你好", id="a1")
    # 流式更新（id 相同，内容变长）
    msg2 = AIMessage(content="你好，我是 AI！", id="a1")
    return {"messages": [msg1, msg2]}


g = StateGraph(State)
g.add_node("u", user_turn)
g.add_node("a", ai_turn)
g.add_edge(START, "u")
g.add_edge("u", "a")
g.add_edge("a", END)
app = g.compile()

result = app.invoke({"messages": []})
for m in result["messages"]:
    print(f"{m.__class__.__name__}(id={m.id}): {m.content}")

输出：

HumanMessage(id=u1): 你好
AIMessage(id=a1): 你好，我是 AI！

注意 AI 消息只有一条（内容是更新后的版本），不是两条。这就是 add_messages 的去重更新能力。如果用普通 add，会看到两条同 id 的 AI 消息（半成品 + 完整）。

八、reducer 决定状态如何进化

回到核心抽象：状态在节点间不是简单替换，而是按 reducer 进化。

graph LR
    S0[初始 State] -->|节点A 返回 dict| R[reducer 合并]
    R --> S1[State v1]
    S1 -->|节点B 返回 dict| R2[reducer 合并]
    R2 --> S2[State v2]
    S2 -->|节点C 返回 dict| R3[reducer 合并]
    R3 --> S3[最终 State]

每个字段都有独立的进化策略，让你能精细控制不同字段的合并语义。这跟 Redux 的 reducer 思想一致，但 LangGraph 用 Annotated 把它声明化、字段化。

九、Reducer 与 LangGraph4j 对比

维度	Python	Java (LangGraph4j)
标注方式	Annotated[T, fn]	@Reducer 注解 + 类继承
默认策略	覆盖	覆盖
内置 add_messages	from langgraph.graph.message import add_messages	内置
自定义	写函数	实现 Reducer 接口
多 reducer 字段	每个 Annotated 独立	每个字段独立注解

Java 因静态类型，自定义 reducer 要实现接口；Python 因函数一等公民，直接传函数更灵活。

十、常见踩坑

1. 没用 reducer 导致消息被覆盖丢失历史：messages: list 而不是 messages: Annotated[list, add_messages]——只留最后一条消息。对话场景必须配 reducer。
2. 计数器被覆盖而不是累加：counter: int 而非 Annotated[int, add]——每次新值覆盖旧值。
3. 自定义 reducer 不处理 None：第一次更新时 old 可能是 None，没处理就抛 TypeError。if old is None: return new。
4. reducer 函数有副作用：reducer 应该是纯函数。在里面调 LLM、写日志都不行（会被多次调用、并发场景出错）。
5. 消息没设 id 用了 add_messages：消息没 id 时 add_messages 当作"不同消息"全部追加——不会去重。流式场景务必给消息设 id：

   AIMessage(content="...", id=str(uuid4()))

6. 想要"清空"列表字段：返回 {"append_list": []} 不会清空——add(old, []) 还是 old。要清空得用别的方法（如用覆盖型字段或自定义 reducer）。
7. reducer 返回了和原对象相同引用：reducer 应返回新对象，原地修改可能引起并发问题。
8. Annotated 顺序写反：Annotated[add, list] 是错的——类型在前，reducer 在后。
9. Pydantic 状态用 Annotated 不生效：Pydantic v2 配合 LangGraph 时 reducer 也能用，但要确保字段有默认值，否则首次合并报错。
10. 想要"先到先得"策略：默认覆盖是"后到先得"。想反过来要自定义 reducer：

    def keep_first(old, new):
        return old if old is not None else new

十一、调试 reducer

想知道 reducer 实际怎么合并的，加 print：

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

class State(TypedDict):
    counter: Annotated[int, debug_add]

调试完换成 add 即可。

十二、小结

• reducer 是 (old, new) -> merged 函数，决定字段如何被多个节点合并。
• 默认策略是覆盖；用 Annotated[T, reducer] 指定合并策略。
• 内置：operator.add（累加/拼接）、add_messages（消息追加+同 id 更新）。
• 自定义 reducer 写普通函数即可，注意处理 old=None。
• 对话/智能体场景的 messages 字段必须配 add_messages，否则历史丢失。
• reducer 让你能为每个字段声明独立的进化语义。

下一篇讲 编译与运行——把图跑起来。