State reducers State reducer
Proficient in the core mechanism of LangGraph state update

1What is a reducer? Let’s start with “default coverage”

In LangGraph, each node will return a dictionary after execution. This dictionary contains the State field it wants to update. So here comes the question:When a node returns a new value, how should LangGraph write the new value to State?

reducerIt’s the mechanism that answers this question. It defines "what rules are used to update a field in State when the node returns a new value."

Default behavior: direct overwrite (Last Write Wins)

If you don't specify any reducer, LangGraph usesDefault coverage strategy: The new value returned by the node directly replaces the old value, as if an assignment was made.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})

The output is{'foo': 2}. node_1 returned{"foo": 2}, LangGraph uses2Directly overwrites the old value1. This is perfectly fine in single-node linear graphs.

reducer syntax: Annotated annotations

To specify a reducer for a State field, you need to use Python'sAnnotatedType annotation, the format is as follows:

from typing import Annotated

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

Annotated[T, reducer_fn]Tell LangGraph:The type of this field is T. When the node returns a new value, reducer_fn(old value, new value) is used to calculate the final value., rather than covering it directly.

2Conflict problem of parallel nodes: why the default behavior reports an error

The default overlay works fine in single-link graphs, but when your graph hasbranch parallelismThe problem comes when it comes to structure.

Scenario: node_1 forks to node_2 and node_3 at the same time

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()

When you rungraph.invoke({"foo": 1})What will happen?

The execution sequence is as follows:

3operator.add reducer: Use list append instead of overwrite

The most direct way to resolve parallel conflicts is tofoofromintChange tolist[int], and useoperator.addAs a reducer.

What is operator.add?

Python built-inoperatorIn the module,operator.addthat is+Functional form of operator:

• For integers:add(1, 2) → 3(numbers added)
• pair list:add([1, 2], [3]) → [1, 2, 3](list splicing)
• For strings:add("ab", "cd") → "abcd"(String concatenation)

When the State field type islisthour,operator.addwill return a new list of nodesSplicing(append/concat) to the end of the old list instead of overwriting it.

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]})

Execution process breakdown

with initial input{"foo": [1]}For example, the complete execution process is as follows:

noticed3appears twice because both node_2 and node_3 are based onnode_1The subsequent state (foo[-1] == 2) calculation, each can get3, and then reducerAppend allCome in. This is exactly the correct behavior of the reducer: keep the output of all parallel nodes.

Limitation of operator.add: Cannot handle None

If the initial State is passed inNoneGivefoo,what happens?

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

operator.addtry to executeNone + [2], Python cannot spliceNoneandlist, so an error is reported. That's why we needCustom reducer。

4Custom reducer: handle edge cases such as None

A custom reducer is an ordinary Python function with a signature ofreducer(left, right) → result,in:

• left: The current old value of this field in State
• right: the new value returned by the node
• return: The final value written to State after merging

Implement a custom reducer that can handle None

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]  # 换成自定义函数

Comparison: Performance difference when passing in None

The complete calling timing of the reducer function

Each time the node returns a dictionary containing the field, the reducer is called once:

• 1

  After the node is executed, the update dictionary is returned.

  For examplereturn {"foo": [2]}

• 2

  LangGraph recognizes that foo has the reducer annotation

  Read the old value of foo in the current State (left) and the new value returned by the node (right)

• 3

  Call the reducer function

  reduce_list(旧值, 新值)Return the merged result

• 4

  Write the reducer return value to State

  If there are multiple parallel nodes, first collect the return values ​​of all nodes, and then merge them through the reducer.

5add_messages reducer: specially designed for message lists

In actual use of LangGraph, the most common State field isConversation message list. LangGraph provides a built-in, powerful reducer for this purpose:add_messages。

Equivalent writing method of MessagesState and custom 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

Basic behavior of add_messages: Append 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 条

This is exactly the desired behavior of a chatbot: new messages generated by each round of conversation areAppendinto the history, rather than replacing the entire history.

Comparison of add_messages and operator.add

6Message overwriting: replace existing messages with the same ID

add_messagesThere is a powerful hidden feature:If you pass in a new message with the same ID as an existing message, it will replace (overwrite) the old message with the new message instead of appending。

Why do you need message overwriting?

Typical scenario: A user sends a message but then wants to modify it (such as correcting a wrong question). Or the AI ​​generated a message and you want to replace the content before saving. Message overwrite allows you to precisely modify a record in the conversation history.

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 替换掉原来那条

7Message deletion: precise removal with RemoveMessage

add_messagesAlso supports a special "delete" operation: passing inRemoveMessageObject, which can be selected from the message listRemove message with specified ID。

How RemoveMessage works

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" 的消息

Typical usage scenarios for deleting messages

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]}

8Summary: reducer Selection Guide and Best Practices

Comparison of four reducer strategies

Selection decision tree

Summary of key concepts

Understand the role of reducer in one picture