State Reducer

什么是State Reducer

State Reducer 是 LangGraph 中用于合并状态更新的核心机制。在 LangGraph 的StateGraph中，每个节点可以读取和写入共享状态，而 Reducer 定义了如何将多个节点对同一状态键的更新合并

Reducer 的核心特征：

• 函数签名：(Value, Value) -> Value，接收当前值和更新值，返回合并后的新值
• 注解定义：通过Annotated[Type, reducer_function]为状态键指定 Reducer
• 默认行为：未指定 Reducer 的状态键使用覆盖策略（Last-Write-Wins）

如何定义Reducer

定义Reducer函数

Reducer 本质上是一个二元合并函数，用于定义当同一个字段产生多个更新值时，LangGraph 应该如何将这些值合并为一个最终结果。

函数签名：(Value, Value) -> Value

示例代码如下：

defmy_reducer(left:list[str],right:list[str])->list[str]:returnleft+right left=['a','b']right=['c']print(my_reducer(left,right))

其中，my_reducer用于处理list[str]类型的数据。它接收两个列表参数：

• left：当前已累计的状态值；
• right：本次待合并的新值。

函数内部通过left + right将两个列表合并，并返回合并后的结果。

因此，该 Reducer 的作用是：当某个状态字段存在多次列表更新时，将这些列表内容追加合并，而不是直接覆盖原值。

运行结果如下

['a', 'b', 'c']

将Reducer和状态字段关联

在 LangGraph 中，Reducer 通常通过 Python 的typing.Annotated与状态字段进行关联。

Annotated[]是 Python 提供的一种类型注解扩展机制，用于在原始类型之外附加额外的元数据信息。需要注意的是，Annotated[]本身并不规定这些元数据的具体含义，它只负责在类型注解中保留这些信息。

严格来说，Annotated的第一个参数是被注解的原始类型，后续参数是附加的元数据。至于这些元数据表示什么、如何解析，则由使用它的框架或工具自行决定。

在 LangGraph 中，框架利用这一机制，将状态字段的类型和 Reducer 规则同时声明在字段定义中。其基本形式如下：

Annotated[Type,reducer_function]

其中：

• Type：表示状态字段的数据类型；
• reducer_function：表示该字段对应的 Reducer 函数。

示例代码如下：

from typing import TypedDict, Annotated class OverAllState(TypedDict): logs: Annotated[list[str], my_reducer] cur_id: str

在上述代码中：

• logs字段的类型是list[str]；
• my_reducer是与logs字段关联的 Reducer 函数；
• 当多个节点同时更新logs字段时，LangGraph 会使用my_reducer将多个列表合并；
• cur_id字段没有指定 Reducer，因此采用默认更新规则。

常用内置Reducer函数

1. operator.add

operator.add是 Python 内置的加法操作函数，底层由 C 实现

它接收两个参数，等价于a（第一个参数）+b（第二个参数）

代码如下

fromoperatorimportaddprint(f"{add(1,2)=}")print(f"{add([1,2],[3,4])=}")print(f"{add(['a','b'],['c'])=}")

输出如下

add(1,2) = 3 add([1,2], [3,4]) = [1, 2, 3, 4] add(['a','b'], ['c']) = ['a', 'b', 'c']

langgraph.graph.message.add_messages

add_messages是 LangGraph 中专用于合并消息列表的 Reducer 函数，常用于维护对话历史类的状态字段。其函数签名如下：

defadd_messages(left:Messages,right:Messages,*,format:Literal["langchain-openai"]|None=None,)->Messages:...returnmerged

参数说明：

• left：状态中已有的消息列表；
• right：当前节点返回的消息更新值；
• format：可选参数，用于指定返回消息的格式，通常无需手动设置。

left与right的类型均为Messages。Messages可以理解为 LangChain 消息对象的列表，其中每个元素都是BaseMessage或其子类的实例，常见子类包括：

• HumanMessage：用户的输入消息；
• AIMessage：AI 的回复消息；
• SystemMessage：系统提示消息；
• ToolMessage：工具调用的结果消息。

add_messages处理的是对话消息序列，而非普通的字符串列表。

BaseMessage包含一个可选的id属性，用于唯一标识一条消息。add_messages在合并left与right时，不是简单地执行列表拼接，而是依据消息的id进行合并：

• 若right中的某条消息的id在left中不存在，则将该消息追加到结果列表末尾；
• 若right中的某条消息的id与left中已有消息的id相同，则使用right中的新消息替换left中的旧消息。

因此，add_messages的作用可以概括为：在保留历史消息的基础上追加新消息，并允许通过相同的消息id覆盖已有消息。

需要特别说明，add_messages并非简单地对left与right求“并集”。更准确地说，它是一个基于消息id的消息列表合并函数：既支持追加新消息，也支持更新已有消息。

可以理解为：

merged=left+right

但若right中存在与left相同id的消息，则最终结果中不会出现重复消息，而是用right中的消息覆盖left中对应的旧消息。

示例代码如下

fromlanggraph.graph.messageimportadd_messagesfromlangchain.messagesimportHumanMessage,AIMessage,SystemMessage left=[SystemMessage(content="你是个善解人意的助手",id='1'),HumanMessage(content="你好",id='2'),AIMessage(content="你好~",id='3'),]right=[HumanMessage(content="我是老王，你是小王",id='2'),AIMessage(content="好的，我记住啦",id='3'),HumanMessage(content="你是谁？",id='4'),AIMessage(content="我是小王",id='5'),]merged=add_messages(left,right)formsginmerged:print(msg)

输出如下

content='你是个善解人意的助手' additional_kwargs={} response_metadata={} id='1' content='我是老王，你是小王' additional_kwargs={} response_metadata={} id='2' content='好的，我记住啦' additional_kwargs={} response_metadata={} id='3' tool_calls=[] invalid_tool_calls=[] content='你是谁？' additional_kwargs={} response_metadata={} id='4' content='我是小王' additional_kwargs={} response_metadata={} id='5' tool_calls=[] invalid_tool_calls=[]

默认行为

如果某个 State 字段没有显式定义Reducer，LangGraph 会使用默认的状态更新行为：后一次更新值会覆盖该字段原有的状态值。

换句话说，当节点返回的更新结果中包含某个字段时，如果该字段没有配置 Reducer，LangGraph 不会对新旧值进行追加、合并或累加，而是直接使用本次返回的新值替换原来的旧值。

示例代码如下：

fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDictclassOverAllState(TypedDict):logs:list[str]id:strdefnode_a(state:OverAllState):return{"logs":["node_a"],"id":"node_a"}defnode_b(state:OverAllState):return{"logs":["node_b"],"id":"node_b"}builder=StateGraph(state_schema=OverAllState)builder.add_node("node_a",node_a)builder.add_node("node_b",node_b)builder.add_edge(START,"node_a")builder.add_edge("node_a","node_b")builder.add_edge("node_b",END)graph=builder.compile()result=graph.invoke({"logs":["START"],"id":"start"})print('='*30,'-> result <-','='*30)print(result)

输出如下

============================== -> result <- ============================== {'logs': ['node_b'], 'id': 'node_b'}

可以看到，logs字段和id字段都没有定义Reducer，因此，节点返回的新值会覆盖初始状态中的旧值，图运行结果中的状态值和最后一次更新保持一致。