State Reducers 状态归约器
精通 LangGraph 状态更新的核心机制

1 什么是 Reducer？从"默认覆盖"说起

在 LangGraph 中，每个节点执行完后都会返回一个字典，这个字典里是它想要更新的 State 字段。那么问题来了：当节点返回了新值，LangGraph 应该怎样把这个新值写入 State？

Reducer（归约器） 就是回答这个问题的机制。它定义了"对于 State 中的某个字段，当节点返回一个新值时，用什么规则来更新它"。

默认行为：直接覆盖（Last Write Wins）

如果你没有指定任何 Reducer，LangGraph 使用默认覆盖策略：节点返回的新值直接替换掉旧值，就好像做了一次赋值 state['foo'] = 新值。

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

# 最简单的 State：没有任何 Reducer 注解
class State(TypedDict):
    foo: int   # 没有 Annotated，使用默认覆盖行为

def node_1(state):
    print("---Node 1---")
    return {"foo": state['foo'] + 1}  # 返回 foo + 1

builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", END)
graph = builder.compile()

graph.invoke({"foo": 1})

输出是 {'foo': 2}。node_1 返回了 {"foo": 2}，LangGraph 用 2 直接覆盖了旧值 1。这在单节点线性图里完全没问题。

Reducer 的语法：Annotated 注解

给某个 State 字段指定 Reducer，需要使用 Python 的 Annotated 类型注解，格式如下：

from typing import Annotated

class State(TypedDict):
    # 格式：Annotated[字段类型, reducer函数]
    foo: Annotated[list[int], 某个reducer函数]
    #              ↑ 字段实际类型   ↑ 指定如何合并新旧值

Annotated[T, reducer_fn] 告诉 LangGraph：这个字段的类型是 T，当节点返回新值时，用 reducer_fn(旧值, 新值) 来计算最终值，而不是直接覆盖。

2 并行节点的冲突问题：为什么默认行为会报错

默认覆盖在单链路图里工作得很好，但当你的图有分支并行结构时，问题就来了。

场景：node_1 同时分叉到 node_2 和 node_3

class State(TypedDict):
    foo: int  # 没有 Reducer，默认覆盖

def node_1(state): return {"foo": state['foo'] + 1}
def node_2(state): return {"foo": state['foo'] + 1}
def node_3(state): return {"foo": state['foo'] + 1}

builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")  # node_1 → node_2
builder.add_edge("node_1", "node_3")  # node_1 → node_3（并行！）
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
graph = builder.compile()

当你运行 graph.invoke({"foo": 1}) 时，会发生什么？

执行时序如下：

3 operator.add Reducer：用列表追加代替覆盖

解决并行冲突最直接的方式，是把 foo 从 int 改成 list[int]，并使用 operator.add 作为 Reducer。

operator.add 是什么？

Python 内置的 operator 模块里，operator.add 就是 + 运算符的函数形式：

• 对整数：add(1, 2) → 3（数字相加）
• 对列表：add([1, 2], [3]) → [1, 2, 3]（列表拼接）
• 对字符串：add("ab", "cd") → "abcd"（字符串拼接）

当 State 字段类型是 list 时，operator.add 会把节点返回的新列表拼接（append/concat）到旧列表的末尾，而不是覆盖它。

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    # foo 现在是 list[int]，使用 operator.add 来追加
    foo: Annotated[list[int], add]
    #              ↑ 列表类型        ↑ 拼接操作

def node_1(state):
    print("---Node 1---")
    return {"foo": [state['foo'][-1] + 1]}  # 返回一个列表！

def node_2(state):
    print("---Node 2---")
    return {"foo": [state['foo'][-1] + 1]}  # 同样返回一个列表

def node_3(state):
    print("---Node 3---")
    return {"foo": [state['foo'][-1] + 1]}

# 构建与之前相同的并行图
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_1", "node_3")
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
graph = builder.compile()

graph.invoke({"foo": [1]})

执行过程拆解

以初始输入 {"foo": [1]} 为例，完整执行过程如下：

注意到 3 出现了两次，因为 node_2 和 node_3 都基于 node_1 之后的状态（foo[-1] == 2）计算，各自得出 3，然后被 Reducer 全部追加进来。这正是 Reducer 的正确行为：把所有并行节点的输出都保留下来。

operator.add 的局限：不能处理 None

如果初始 State 传入 None 给 foo，会发生什么？

try:
    graph.invoke({"foo": None})
except TypeError as e:
    print(f"TypeError occurred: {e}")

operator.add 尝试执行 None + [2]，Python 无法拼接 None 和 list，因此报错。这就是为什么我们需要自定义 Reducer。

4 自定义 Reducer：处理 None 等边界情况

自定义 Reducer 就是一个普通的 Python 函数，签名为 reducer(left, right) → result，其中：

• left：State 中该字段当前的旧值
• right：节点返回的新值
• return：合并后写入 State 的最终值

实现一个能处理 None 的自定义 Reducer

def reduce_list(left: list | None, right: list | None) -> list:
    """安全地合并两个列表，处理其中任意一个或两个为 None 的情况。

    Args:
        left (list | None): 第一个列表，或 None
        right (list | None): 第二个列表，或 None

    Returns:
        list: 包含两个输入列表所有元素的新列表。
              如果输入为 None，视作空列表处理。
    """
    if not left:
        left = []   # None → 空列表
    if not right:
        right = []  # None → 空列表
    return left + right  # 安全拼接

# 用 operator.add：不能处理 None
class DefaultState(TypedDict):
    foo: Annotated[list[int], add]

# 用自定义 reduce_list：能处理 None
class CustomReducerState(TypedDict):
    foo: Annotated[list[int], reduce_list]  # 换成自定义函数

对比：传入 None 时的表现差异

Reducer 函数的完整调用时机

每次节点返回一个包含该字段的字典，Reducer 都会被调用一次：

• 1

  节点执行完毕，返回更新字典

  例如 return {"foo": [2]}

• 2

  LangGraph 识别出 foo 有 Reducer 注解

  读取当前 State 中 foo 的旧值（left）和节点返回的新值（right）

• 3

  调用 Reducer 函数

  reduce_list(旧值, 新值) 返回合并后的结果

• 4

  将 Reducer 返回值写入 State

  如有多个并行节点，先收集所有节点的返回值，再依次经过 Reducer 合并

5 add_messages Reducer：专为消息列表设计

在 LangGraph 的实际使用中，最常见的 State 字段是对话消息列表。LangGraph 为此提供了一个内置的、功能强大的 Reducer：add_messages。

MessagesState 与自定义 State 的等价写法

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import MessagesState
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

# 方式一：手动定义（完整控制）
class CustomMessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    added_key_1: str   # 可以继续添加其他字段
    added_key_2: str

# 方式二：继承 MessagesState（推荐，更简洁）
class ExtendedMessagesState(MessagesState):
    # MessagesState 已内置：messages: Annotated[list[AnyMessage], add_messages]
    added_key_1: str   # 只需添加额外字段即可
    added_key_2: str

add_messages 的基本行为：追加消息

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

# 模拟当前 State 中已有的消息列表
initial_messages = [
    AIMessage(content="Hello! How can I assist you?", name="Model"),
    HumanMessage(content="I'm looking for information on marine biology.", name="Lance")
]

# 节点返回了一条新消息
new_message = AIMessage(
    content="Sure, I can help with that. What specifically are you interested in?",
    name="Model"
)

# add_messages(旧列表, 新消息) → 追加后的列表
result = add_messages(initial_messages, new_message)
# result 包含 3 条消息：原来 2 条 + 新的 1 条

这正是聊天机器人所需要的行为：每轮对话产生的新消息都被追加到历史记录中，而不是替换掉整个历史。

add_messages 与 operator.add 的对比

6 消息覆写：用相同 ID 替换已有消息

add_messages 有一个强大的隐藏特性：如果你传入的新消息与现有消息有相同的 ID，它会用新消息替换（覆写）旧消息，而不是追加。

为什么需要消息覆写？

典型场景：用户发出了一条消息，但随后想要修改它（比如纠正一个错误的问题）。或者 AI 生成了一条消息，你想在保存前替换掉其中的内容。消息覆写让你能够精准地修改对话历史中的某一条记录。

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

# 当前 State 中的消息（注意：都有显式的 id）
initial_messages = [
    AIMessage(content="Hello! How can I assist you?", name="Model", id="1"),
    HumanMessage(content="I'm looking for information on marine biology.", name="Lance", id="2")
]

# 新消息的 id="2"，与上面第二条消息的 id 相同！
new_message = HumanMessage(
    content="I'm looking for information on whales, specifically",
    name="Lance",
    id="2"   # ← 关键：相同的 ID
)

result = add_messages(initial_messages, new_message)
# 结果：id="2" 的那条消息被替换成了新的内容

def fix_last_message(state):
    messages = state["messages"]
    last_msg = messages[-1]  # 拿到最后一条 AI 消息

    # 构造一条相同 ID 的新消息，内容做了修正
    corrected = AIMessage(
        content=last_msg.content + " [已审核]",
        id=last_msg.id  # 相同 ID → 触发覆写
    )
    return {"messages": [corrected]}
    # add_messages 看到相同 ID，用 corrected 替换掉原来那条

7 消息删除：用 RemoveMessage 精准移除

add_messages 还支持一种特殊的"删除"操作：传入 RemoveMessage 对象，可以从消息列表中移除指定 ID 的消息。

RemoveMessage 的工作原理

from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage

# 当前消息列表（4 条消息）
messages = [
    AIMessage("Hi.", name="Bot", id="1"),
    HumanMessage("Hi.", name="Lance", id="2"),
    AIMessage("So you said you were researching ocean mammals?", name="Bot", id="3"),
    HumanMessage("Yes, I know about whales. But what others should I learn about?", name="Lance", id="4")
]

# 构造删除指令：删除前两条消息（id="1" 和 id="2"）
delete_messages = [RemoveMessage(id=m.id) for m in messages[:-2]]
print(delete_messages)
# [RemoveMessage(id="1"), RemoveMessage(id="2")]

# 执行删除
result = add_messages(messages, delete_messages)
# result 中只剩 id="3" 和 id="4" 的消息

删除消息的典型使用场景

def trim_old_messages(state):
    messages = state["messages"]
    if len(messages) > 10:  # 超过 10 条则删除最旧的
        to_delete = messages[:-10]
        return {"messages": [RemoveMessage(id=m.id) for m in to_delete]}
    return {}

from langchain_core.messages import ToolMessage

def clean_tool_messages(state):
    messages = state["messages"]
    # 删除所有工具调用相关的中间消息
    tool_msgs = [m for m in messages if isinstance(m, ToolMessage)]
    return {"messages": [RemoveMessage(id=m.id) for m in tool_msgs]}

8 总结：Reducer 选型指南与最佳实践

四种 Reducer 策略对比

选型决策树

关键概念总结

一张图看懂 Reducer 的作用