LangGraph 中状态的更新策略

LangGraph 的核心在于状态在计算图中的流转与更新。状态的更新策略在源码层面通过 Reducer（归约函数） 来实现——即通过 Annotated[type, reducer] 语法为字段指定一个 (old, new) -> merged 函数来定义如何合并新旧值。若不显式指定，默认采用”覆盖更新”（由 LastValue 通道实现），即节点返回的新值直接替换该字段的旧值。本文梳理几种内置 Reducer（更新策略）及其适用场景，并介绍如何自定义 Reducer。

默认覆盖更新

覆盖更新是 LangGraph 的默认行为，无需通过 Annotated 声明。它的语义很直接：节点返回什么，对应字段就更新为什么。需要特别注意的是，它只覆盖字段级别的值，而非整个状态字典。以下是一个简单的示例：

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


class MyState(TypedDict):  
    foo: str  
    bar: str  

def node_1(state: MyState):  
    return {"foo": "foo"}  

def node_2(state: MyState):  
    return {"bar": "bar"}  

graph = StateGraph(MyState)  

graph.add_node(node_1)  
graph.add_node(node_2)  

graph.add_edge(START, "node_1")  
graph.add_edge("node_1", "node_2")  
graph.add_edge("node_2", END)  

app = graph.compile()  

res = app.invoke({"foo": "", "bar": ""})  

print(res)

# 输出
# {'foo': 'foo', 'bar': 'bar'}

示例中，node_1 将 foo 更新为 "foo"，node_2 将 bar 更新为 "bar"。由于覆盖策略是基于字段的，两个节点的返回结果最终合并为完整的字典输出。

add_messages 追加消息

add_messages 是 LangGraph 内置的 Reducer，专为聊天/消息场景设计。与默认覆盖不同，它将节点返回的新消息追加到历史消息列表之后。它还具备按 ID 去重更新的能力：若新消息的 ID 已存在于列表中，则原地更新该消息；若不存在，则追加。以下是一个示例：

from typing import TypedDict, Annotated  
from langgraph.graph import add_messages, StateGraph, START, END  


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

def chat_node_1(state: AgentState):  
    return {"messages": [("assistant", "Hello from node 1")]}  

def chat_node_2(state: AgentState):  
    return {"messages": [("assistant", "Hello from node 2")]}  

graph = StateGraph(AgentState)  

graph.add_node("chat1", chat_node_1)  
graph.add_node("chat2", chat_node_2)  

# 两个节点并行执行，各自追加消息
graph.add_edge(START, "chat1")  
graph.add_edge(START, "chat2")  
graph.add_edge("chat1", END)  
graph.add_edge("chat2", END)  

app = graph.compile()  

res = app.invoke({"messages": [("human", "Hello!")]})  

print(res)

# 输出（简化）
# {'messages': [HumanMessage(content='Hello!', ...), AIMessage(content='Hello from node 1', ...), AIMessage(content='Hello from node 2', ...)]}

可以看到，最终结果保留了初始的 human 消息，并追加了两条来自不同节点的 AI 回复——这正是 add_messages 的效果。LangGraph 还会自动将 ("role", "content") 元组转为对应的 BaseMessage 子类（如 HumanMessage、AIMessage），因此无需手动转换。

operator.add 累加或拼接

operator.add 是 Python 内置的加法函数，作为 Reducer 使用时将更新视为”累加”操作。不同类型的表现不同：对于数值，执行加法运算；对于列表和字符串，执行拼接操作。

下面的示例通过三个并行分支，同时演示了这三种类型的 add 效果：

import operator  
from typing import TypedDict, Annotated  
from langgraph.graph import StateGraph, START, END  


class MyState(TypedDict):  
    tags: Annotated[list, operator.add]      # 列表 → 拼接  
    greeting: Annotated[str, operator.add]   # 字符串 → 拼接  
    count: Annotated[int, operator.add]      # 数值 → 相加  

def add_list_node(state: MyState):  
    return {"tags": ["langgraph"]}  

def add_str_node(state: MyState):  
    return {"greeting": ",world"}  

def add_int_node(state: MyState):  
    return {"count": 2}  

graph = StateGraph(MyState)  

graph.add_node(add_int_node)  
graph.add_node(add_str_node)  
graph.add_node(add_list_node)  

graph.add_edge(START, "add_list_node")  
graph.add_edge(START, "add_str_node")  
graph.add_edge(START, "add_int_node")  

graph.add_edge("add_list_node", END)  
graph.add_edge("add_str_node", END)  
graph.add_edge("add_int_node", END)  

app = graph.compile()  

res = app.invoke({"tags": ["langchain"], "greeting": "hello", "count": 1})  
print(res)

# 输出
# {'tags': ['langchain', 'langgraph'], 'greeting': 'hello,world', 'count': 3}

三个结果分别验证了 add 对不同类型的效果。对应的计算图如下所示：

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	add_int_node(add_int_node)
	add_str_node(add_str_node)
	add_list_node(add_list_node)
	__end__([<p>__end__</p>]):::last
	__start__ --> add_int_node;
	__start__ --> add_list_node;
	__start__ --> add_str_node;
	add_int_node --> __end__;
	add_list_node --> __end__;
	add_str_node --> __end__;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

operator.mul 相乘

operator.mul 在实际使用中很容易给人造成困惑。先看一个例子：

import operator  
from typing import TypedDict, Annotated  
from langgraph.graph import StateGraph, START, END  


class MyState(TypedDict):  
    factor: Annotated[float, operator.mul]  

def mul_node(state: MyState) -> MyState:  
    return {"factor": 2.0}  

graph = StateGraph(MyState)  
graph.add_node(mul_node)  
graph.add_edge(START, "mul_node")  
graph.add_edge("mul_node", END)  

app = graph.compile()  
res = app.invoke({"factor": 2.0})  
print(res)

# 输出
# {'factor': 0.0}

期望的结果是 4.0（2.0 × 2.0），但实际输出是 0.0。问题出在 LangGraph 的底层实现上。

当 Reducer 是一个二元操作符（接受两个参数）时——无论是 operator.add、operator.mul 还是自定义的 lambda——LangGraph 都会为对应字段创建 BinaryOperatorAggregate 通道。该通道在初始化时通过调用类型的无参构造函数来设定初始值——对于 int 得到 0，对于 float 得到 0.0。后续所有的写入都通过二元操作符与当前值进行运算：

对 add 来说 0 是加法单位元，所以表现正常；但对 mul 来说 0 是”吸收元”——任何数乘以 0 都得 0，导致 operator.mul 对数值字段几乎不可用。此外，operator.mul 对列表也不适用（list * list 会抛出 TypeError）。

如果需要乘法语义，可以用自定义 Reducer 配合其他通道类型，或在节点中手动计算后直接覆盖字段值。

自定义策略

当内置策略无法满足需求时，可以自定义 Reducer。自定义 Reducer 是一个满足以下签名的函数：

def my_reducer(current_value, update_value):  
    return merged_value

Reducer 拿到当前旧值与本次节点更新的新值，返回合并后的结果——本质上是定义”如何合并新旧数据”。

自定义相乘 Reducer

前面的 operator.mul 因底层零初始化而失效。自定义 Reducer 可以绕过这个限制：

from typing import TypedDict, Annotated  
from langgraph.graph import StateGraph, START, END  


def mul_reducer(current: float, update: float) -> float:  
    # BinaryOperatorAggregate 初始值为 0.0，将其视为"未初始化"
    return current * update if current != 0.0 else update  

class MyState(TypedDict):  
    factor: Annotated[float, mul_reducer]  

def mul_node(state: MyState) -> MyState:  
    return {"factor": 2.0}  

graph = StateGraph(MyState)  
graph.add_node(mul_node)  
graph.add_edge(START, "mul_node")  
graph.add_edge("mul_node", END)  

app = graph.compile()  
res = app.invoke({"factor": 2.0})  
print(res)

# 输出
# {'factor': 4.0}

注意 这种”跳过零值”的做法是一个实用的变通方案，但并非代数上严格的乘法 Reducer——若某一步的合法计算结果恰好为 0.0，后续的乘法也会被跳过。在大多数业务场景下这不会造成问题，但如果需要严格语义，应考虑在节点中手动计算后直接覆盖字段值，或选择其他架构方案。

总结

LangGraph 中状态的更新策略可以归纳为三类：

选择时的几点建议：

• 默认覆盖是最直观的策略，能满足大部分串行节点的需求。
• 需要合并而非替换时，考虑内置 Reducer：聊天场景用 add_messages，数值或列表累加用 operator.add。
• operator.mul 因底层零初始化问题不适用于数值字段，如有乘法需求请自定义 Reducer。
• 自定义 Reducer 只需一个 (old, new) -> merged 函数，但要注意 BinaryOperatorAggregate 的初始化行为对结果的影响。