LangGraph——reducer

作者：CSDN博客
在 LangGraph 中，reducer 是一个核心概念，它定义了当图（graph）中的某个节点（或检查点 checkpoint）接收到多个值时，如何将这些值合并成一个单一的值。这在处理并行分支、循环或消息传递等场景中至关重要，因为它决定了状态如何被更新和累积。
简单来说，reducer 就是一个合并函数：reducer(old_value, new_value) -> combined_value。
一、Reducer 的作用

在 LangGraph 的 StateGraph 中，状态（state）通常是一个字典。每个键（key）可以关联一个 reducer，用于定义该键的值在多次更新时如何合并。
例如：

你有一个 messages 列表，每次节点运行都会追加新消息。reducer 应该是 add_messages，将新消息添加到旧列表末尾。你有一个 total_cost 数值，每次计算都累加。reducer 应该是 sum 或自定义累加函数。

---

二、常见的 Reducer 类型和使用方法

LangGraph 提供了多种内置 reducer，也支持自定义。以下是主要的几种：
1. add / operator.add

作用：对数值或字符串进行加法合并。适用类型：int, float, str, list（等支持 + 操作的类型）

• 行为：old + new
  1. from langgraph.graph import StateGraph, ADD
  2. from operator import add
  4. class State(TypedDict):
  5.     count: Annotated[int, add]  # 或 ADD
  6.     # 或者使用：count: Annotated[int, "add"]
  8. graph = StateGraph(State)

  复制代码

  注意：ADD 是 LangGraph 对 operator.add 的别名。
  

  2. add_messages最常用于聊天应用中消息列表的合并。作用：将新消息（单个或列表）追加到已有消息列表中，支持去重和更新（如替换 AIMessage）。来自：from langgraph.types import add_messages
  
• 1. from langgraph.types import add_messages
  3. class State(TypedDict):
  4.     messages: Annotated[List[BaseMessage], add_messages]
  6. graph = StateGraph(State)

  复制代码

  这是处理 LLM 聊天历史的标准做法，能智能处理消息追加。
  

  
  

3. set / 直接赋值（无 reducer）

作用：新值完全覆盖旧值。适用：当你不希望累积，而是只保留最新值。实现方式：不指定 reducer，或使用 lambda old, new: new

• 1. class State(TypedDict):
  2.     current_topic: str  # 默认行为是覆盖（等价于 set）
  4. # 或显式定义
  5. from typing import Annotated
  6. current_topic: Annotated[str, lambda old, new: new]

  复制代码

4. 自定义 Reducer 函数

你可以定义任意逻辑的 reducer。
示例：只保留最长的字符串

1. def keep_longest_str(old: str, new: str) -> str:
2.     return old if len(old) >= len(new) else new
4. class State(TypedDict):
5.     summary: Annotated[str, keep_longest_str]

复制代码

示例：累加字典中的数值

1. def merge_dicts(old: dict, new: dict) -> dict:
2.     result = old.copy()
3.     for k, v in new.items():
4.         result[k] = result.get(k, 0) + v
5.     return result
7. class State(TypedDict):
8.     stats: Annotated[dict, merge_dicts]

复制代码

三、Reducer 的使用方法

方法 1：使用 Annotated 在 State 定义中声明（推荐）

这是最清晰、最常用的方式。

1. from typing import TypedDict, Annotated, List
2. from langgraph.types import add_messages
3. from typing_extensions import TypedDict
5. class State(TypedDict):
6.     messages: Annotated[List[dict], add_messages]
7.     count: Annotated[int, sum]
8.     notes: Annotated[List[str], list.extend]  # 注意：需包装成函数

复制代码

注意：像 list.extend 这种方法不能直接用，需包装：

1. def extend_list(old: list, new: list) -> list:
2.     return old + new  # 更安全的不可变方式

复制代码

方法 3：使用 LangGraph 内置常量（如 ADD）

1. from langgraph.graph import ADD
3. class State(TypedDict):
4.     score: Annotated[int, ADD]  # 等价于 operator.add

复制代码

四、Reducer 的执行时机

当一个节点返回状态更新时，LangGraph 会检查该字段是否有 reducer。如果有，就调用 reducer(old_value, new_value) 来更新状态。如果没有，就直接覆盖（new_value 替换 old_value）。

Reducer 类型	用途	示例
add_messages	合并聊天消息列表	聊天机器人
ADD / add	数值、字符串、列表相加	累加计数、拼接字符串
set (默认)	覆盖旧值	更新当前状态、主题
自定义函数	特殊合并逻辑	合并统计、去重、筛选等

原文地址：https://blog.csdn.net/qq_50863584/article/details/151157914