首页 > AI教程 >LangGraph Reducer详解

LangGraph Reducer详解

来源：互联网 2026-07-23 06:29:13

Reducer是LangGraph中定义State字段更新规则的函数，默认采用覆盖方式，后执行的节点会替换前一个节点的值；通过指定Reducer可改为合并或累加，其参数为当前值与新值，返回合并结果。由于图存在分支与循环，Reducer能确保数据累积而非丢失，内置operator.add、add_messages等常用归约器。

前情回顾：LangGraph 入门基础全解析

LangGraph Reducer详解

Reducer（归约器）本质上是一个函数，专门解决一个核心问题：当多个节点同时尝试修改 State 中的同一字段时，如何公平合理地合并这些修改。

长期稳定更新的攒劲资源： >>>点此立即查看<<<

更通俗地讲，Reducer 决定了 State 中某个字段的更新策略——是直接覆盖、按顺序追加、累加计算，还是采用完全自定义的合并规则。

从实际问题理解 Reducer 的作用

核心矛盾：多个节点同时写入同一字段

在 LangGraph 中，多个节点按顺序执行，每个节点都可能修改 State。这就引出一个经典问题：

假设有两个返回值：

2（后一个值直接覆盖前一个）—— 这是 LangGraph 的默认行为
3（1 + 2 累加，两个值都保留）—— 这是 Reducer 介入后的结果

Reducer 就是帮助你做出正确选择的机制。

默认行为与 Reducer 行为的对比

默认行为：覆盖（Override）

未指定 Reducer 时，LangGraph 采用后来者居上的覆盖策略。

class MyState(TypedDict):counter: int# 未使用 Reducername: str # 未使用 Reducer

执行流程如下：

初始状态：counter = 0节点 A 返回：{"counter": 1}→ 状态更新为：counter = 1节点 B 返回：{"counter": 2}→ 状态更新为：counter = 2（节点 A 的修改被覆盖）最终结果：counter = 2

这种模式适用于只需要最终值的场景，例如用户名、当前状态标志等，以最后一次写入为准。

Reducer 行为：合并（Merge）

指定 Reducer 后，行为发生改变：

class MyState(TypedDict):counter: Annotated[int, operator.add] # 使用加法 Reducer

执行流程变为：

初始状态：counter = 0节点 A 返回：{"counter": 1}→ 状态更新为：counter = 1节点 B 返回：{"counter": 2}→ 状态更新为：counter = 3（1 + 2 = 3，累加前序修改）最终结果：counter = 3

此模式适用于需要累积的场景，例如计数器、总分、消息列表等，每个节点的贡献都被保留。

Reducer 的本质：一个普通函数

Reducer 本质上就是普通的 Python 函数，接收两个参数：

def reducer_func(current_value, new_value) -> final_value:"""current_value：当前 State 中该字段的现有值new_value：节点返回的该字段的新值return：合并后的最终值"""# 在此定义合并逻辑return merged_result

LangGraph 内置了多个常用 Reducer，可直接使用：

Reducer 函数	效果	等价实现
`operator.add`	数值相加	`lambda a, b: a + b`
`operator.set`	集合合并	`lambda a, b: a \| b`
`add_messages`	消息列表追加	特殊处理，含 ID 去重
不指定	覆盖	`lambda a, b: b`

通过生活案例深入理解 Reducer

场景：共享记账本

你和室友共用一本账本，记录每日开销。这个场景能直观说明 Reducer 的作用。

未使用 Reducer（覆盖模式）：

周一：小明记录"餐饮 50元" → 账本：[餐饮 50元]周二：小红记录"交通 30元" → 账本：[交通 30元]← 周一内容被覆盖周三：两人因账目丢失发生争执

使用 Reducer（追加模式）：

周一：小明记录"餐饮 50元" → 账本：[餐饮 50元]周二：小红记录"交通 30元" → 账本：[餐饮 50元, 交通 30元]← 追加保留周三：月底对账，清晰明了

此例中，add 充当了 Reducer 的角色，规则是：新记录追加到旧记录之后，而非替换。

LangGraph 为何需要 Reducer

原因一：图结构并非线性执行

LangGraph 中的图可能包含分支和循环，执行路径并非单一线性：

┌→ 节点 B ─→┐START ─→┤├→ 节点 D └→ 节点 C ─→┘

节点 B 和节点 C 都可能修改同一字段。若无 Reducer，后执行的节点会覆盖前序节点的修改。Reducer 确保两者的贡献共存，互不覆盖。

原因二：循环结构需要累积

节点 A → 条件判断 → 未满足 → 返回节点 A（循环）

循环中，每次经过节点 A 都可能产生新数据。Reducer 确保数据累积而非每次循环重置，否则循环失去意义。

原因三：状态变化可预测

Reducer 使状态变化透明且可预测。每个字段的更新规则明确，避免出现数值意外丢失的情况。

实战：自定义 Reducer

内置 Reducer 无法满足需求时，可自行编写：

示例1：限制列表长度

def last_n_reducer(n: int):"""仅保留最近 n 条记录"""def reducer(old: list, new: list) -> list:combined = old + newreturn combined[-n:]return reducerclass MyState(TypedDict):recent_logs: Annotated[List, last_n_reducer(10)]# 仅保留最近10条

示例2：字典合并

def dict_merge(old: dict, new: dict) -> dict:"""合并两个字典，新值覆盖旧值"""merged = old.copy()merged.update(new)return mergedclass MyState(TypedDict):metadata: Annotated[Dict, dict_merge]# 字典合并操作

示例3：取最大值

def max_reducer(old: float, new: float) -> float:"""保留最大值"""return max(old, new)class MyState(TypedDict):max_score: Annotated[float, max_reducer]# 保留最高分

Reducer 的完整工作流程

为便于理解，将 LangGraph 内部处理 Reducer 的完整流程分解如下：

1. 节点执行完毕，返回字典，例如 {"counter": 5}2. LangGraph 遍历该字典的每个键值对3. 针对每个键（如 "counter"）： a. 查询 State 定义中该键是否配置了 Reducer b. 若配置了 Reducer：- 获取当前 State 中 "counter" 的旧值（例如 3）- 调用 Reducer 函数：reducer(旧值=3, 新值=5)- 将 Reducer 返回值（例如 8）写入 State c. 若未配置 Reducer：- 直接用新值（5）覆盖旧值（3）4. 更新后的 State 传递给下一个节点

面试级知识点总结

问题	答案
Reducer 是什么？	定义 State 字段更新规则的函数
默认行为是什么？	覆盖（新值替换旧值）
Reducer 改变了什么？	从"覆盖"变为"合并/累加/自定义"
Reducer 的参数？	`(current_value, new_value) → merged_value`
为什么需要 Reducer？	图包含分支和循环，需要累积而非覆盖
内置 Reducer 有哪些？	`operator.add`、`operator.set`、`add_messages`
能否自定义 Reducer？	可以，任何符合签名的函数均可