LangGraph State Reducers Explained

If you have built more than one graph in LangGraph, you have probably hit this: two nodes run in parallel, each returns an update to the same key, and one of them just disappears. That is not a bug in LangGraph, it is a state reducer doing exactly what you told it to do, which by default is "last write wins." LangGraph state reducers are the functions that decide how a node's returned state update gets combined with the state that already exists, and understanding them is the difference between a graph that quietly loses data and one that accumulates it correctly across branches, loops, and retries.

This article walks through what reducers are, why LangGraph needs them at all, the built-in ones you get for free, how to write your own, and the mistakes that trip people up most often, especially around parallel fan-out and message history.

Why LangGraph needs reducers in the first place

LangGraph models an agent as a graph of nodes that read and write a shared state object. Each node is just a function: it takes the current state, does some work (call a tool, call an LLM, transform data), and returns a partial update, a dict with only the keys it wants to change.

The tricky part is the word "partial." If node A returns {"count": 5} and the state already has {"count": 2, "name": "alice"}, what should the new state be? Two questions have to be answered:

• Does count become 5 (overwrite) or 7 (add to the existing value)?
• Does name survive, since node A never mentioned it?

The second question is easy: LangGraph merges partial updates into the existing state, so unmentioned keys are untouched. The first question is what a reducer answers. A reducer is a two-argument function, (current_value, new_value) -> merged_value, associated with a specific key in your state schema. LangGraph calls it every time any node writes to that key.

Without a declared reducer, LangGraph uses the default: overwrite. The new value replaces the old one, full stop. That is fine for a field like current_step or last_error, where you only care about the latest value. It is wrong for anything you want to accumulate, like a running message list, a list of tool calls, or a running total, which is why LangGraph lets you attach a custom reducer per field using Annotated types.

The state schema: TypedDict plus Annotated

State in LangGraph is usually a TypedDict (or a Pydantic model), and reducers are attached using typing.Annotated:

from typing import Annotated, TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    step_count: Annotated[int, operator.add]
    last_result: str  # no reducer: default overwrite

Here messages and step_count both use operator.add as their reducer. For lists, operator.add concatenates them; for ints, it sums them. last_result has no annotation, so it falls back to overwrite behavior, whatever a node returns for that key becomes the new value.

This is the core mental model: the reducer lives on the state schema, not on the node. Every node that touches messages is subject to the same merge rule, whether it returns one message or ten. You do not choose the merge behavior per call, you choose it once per field when you define the schema.

`add_messages`: the reducer you will use constantly

If you are building a chat agent, you are almost certainly tracking a conversation history, and LangGraph ships a purpose-built reducer for that: add_messages, from langgraph.graph.message.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

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

add_messages behaves like operator.add for the common case, new messages get appended to the list, but it adds three things a plain concatenation cannot:

1. Deduplication and updates by message ID. If a node returns a message whose id already exists in the current list, add_messages replaces that message in place instead of appending a duplicate. This is how you patch or correct a message that is already in history without rebuilding the whole list.
2. Deletion support. Returning a RemoveMessage(id=...) (also from langchain_core.messages) tells add_messages to delete that message from state. This is the standard way to trim history or drop a message after summarizing it.
3. Coercion. Raw dicts that look like messages get coerced into proper HumanMessage / AIMessage / ToolMessage objects, so nodes that return plain dicts still work.

A minimal node using this pattern:

from langchain_core.messages import AIMessage

def call_model(state: ChatState) -> dict:
    response = AIMessage(content="Here is the answer.")
    return {"messages": [response]}

You never append to state["messages"] yourself and return the whole list. You return only the new message, and add_messages handles merging it into history. Returning the full list back is a common mistake, it works by accident on the first call and then duplicates everything on the second.

Writing a custom reducer

operator.add and add_messages cover most cases, but real agents often need custom merge logic: dedupe a set, keep the max of two scores, merge two dictionaries, or enforce a cap on list length. A reducer is just a plain function:

from typing import Annotated, TypedDict

def merge_unique(current: list[str], new: list[str]) -> list[str]:
    combined = current + new
    seen = set()
    result = []
    for item in combined:
        if item not in seen:
            seen.add(item)
            result.append(item)
    return result

class ResearchState(TypedDict):
    sources: Annotated[list[str], merge_unique]
    findings: Annotated[dict, lambda a, b: {**a, **b}]

Two rules matter here:

• The reducer must be a pure function of (current, new). LangGraph calls it synchronously as part of applying a superstep; it should not do I/O, call an LLM, or depend on anything outside its two arguments.
• The reducer must handle the "no value yet" case correctly if the key can be absent on the first write. In practice this means designing your default state so the key is initialized (an empty list, empty dict) rather than special-casing None inside every reducer.

A reducer that keeps a running maximum, useful for something like a confidence score across retries:

def keep_max(current: float, new: float) -> float:
    return max(current, new)

class ScoredState(TypedDict):
    best_score: Annotated[float, keep_max]

How reducers behave with parallel branches

This is where reducers stop being a convenience and start being a correctness requirement. When a graph fans out, say, a router sends work to three tool nodes at once, all three run in the same "superstep." LangGraph collects every update from every node in that superstep and then applies the reducer once per key, folding all the concurrent updates together.

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

class FanOutState(TypedDict):
    results: Annotated[list[str], operator.add]

def tool_a(state: FanOutState) -> dict:
    return {"results": ["a-done"]}

def tool_b(state: FanOutState) -> dict:
    return {"results": ["b-done"]}

def tool_c(state: FanOutState) -> dict:
    return {"results": ["c-done"]}

builder = StateGraph(FanOutState)
builder.add_node("tool_a", tool_a)
builder.add_node("tool_b", tool_b)
builder.add_node("tool_c", tool_c)
builder.add_edge(START, "tool_a")
builder.add_edge(START, "tool_b")
builder.add_edge(START, "tool_c")
builder.add_edge("tool_a", END)
builder.add_edge("tool_b", END)
builder.add_edge("tool_c", END)

graph = builder.compile()
print(graph.invoke({"results": []}))
# {'results': ['a-done', 'b-done', 'c-done']}

If results had no reducer (plain overwrite), this graph would raise an InvalidUpdateError at runtime, because LangGraph detects that two or more nodes in the same superstep tried to overwrite the same key without a defined way to combine them. That error is doing you a favor: it is telling you that your fan-out pattern needs a reducer, not that something is broken. The fix is always the same, annotate the field with a reducer that knows how to combine concurrent writes, operator.add for lists, a custom merge function for dicts or sets, or route each branch to write to a different key if merging genuinely does not make sense.

Order is not guaranteed to match "logical" branch order under concurrency, so if you need tool_a's result before tool_b's in the merged list, do not rely on the reducer for ordering, tag each result with its source and sort downstream instead.

Reducers, checkpointers, and loops

Reducers also govern what happens across supersteps when a graph loops, which matters a lot for agent loops with retries or tool-call cycles. Every time the graph re-enters a node (say, an LLM-call node that runs again after a tool result comes back), the reducer is applied against whatever is currently checkpointed, not against the original initial state. This is why add_messages keeps growing history correctly turn after turn: each loop iteration's new messages get folded into the accumulated list from the checkpoint, not into a fresh empty list.

If you are using a checkpointer (MemorySaver, a Postgres or SQLite saver) for persistence across sessions, the same rule holds when you resume a thread: the reducer merges new node output into the state that was last checkpointed for that thread ID. This is one of the more subtle correctness points in LangGraph: your reducer choice determines not just how one superstep merges, but how an entire multi-turn, possibly multi-day conversation accumulates state.

Reducers with Pydantic state models

TypedDict is the common choice, but LangGraph also supports Pydantic models for state, which gives you validation on top of merging. Reducers still attach via Annotated, the mechanism does not change:

from pydantic import BaseModel
from typing import Annotated
import operator

class PydanticState(BaseModel):
    messages: Annotated[list, operator.add] = []
    retries: Annotated[int, operator.add] = 0

Be careful with mutable defaults here the same way you would in any Python dataclass: use = [] inside a Pydantic model (Pydantic handles per-instance defaults correctly), but never share a single mutable list object as a class-level default outside of Pydantic's field system, or every graph run will share and corrupt the same list.

Common mistakes

Returning the whole accumulated value instead of the delta. With operator.add on a list, if a node returns the entire history plus one new item instead of just the new item, the reducer concatenates the whole thing again, and your list doubles every step. Nodes should return only what changed.

Forgetting a reducer on a fan-out key. As shown above, this throws InvalidUpdateError the first time two parallel nodes touch the same unreduced key. Treat that error as a design signal, not a nuisance to suppress.

Using `operator.add` on a field that is sometimes a list and sometimes `None`. operator.add will crash concatenating None and a list. Always initialize the field with an empty container in your default state, not None.

Assuming reducer order equals wall-clock order. Concurrent branches in the same superstep do not guarantee their updates land in a specific order in the merged output. If order matters, encode it in the data (timestamps, branch IDs) rather than depending on the reducer.

Writing an impure reducer. A reducer that calls an LLM, hits a database, or mutates a global is a design smell. It runs inside LangGraph's internal state-application step and should be as boring and deterministic as operator.add.

Mixing up node-local mutation with state updates. Mutating state["messages"] in place inside a node and returning nothing does not go through the reducer at all, and depending on your runtime it may or may not be picked up. Always return a partial update dict; let the reducer do the merging.

A complete small example

Putting it together: an agent that fans out to two research nodes, merges unique findings, tracks message history, and counts total tool calls.

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, AnyMessage
import operator

def merge_findings(current: dict, new: dict) -> dict:
    merged = dict(current)
    for key, value in new.items():
        merged[key] = value
    return merged

class ResearchAgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    findings: Annotated[dict, merge_findings]
    tool_calls: Annotated[int, operator.add]

def search_docs(state: ResearchAgentState) -> dict:
    return {
        "findings": {"docs": "LangGraph uses supersteps"},
        "tool_calls": 1,
        "messages": [AIMessage(content="Searched docs.")],
    }

def search_web(state: ResearchAgentState) -> dict:
    return {
        "findings": {"web": "State merges via reducers"},
        "tool_calls": 1,
        "messages": [AIMessage(content="Searched web.")],
    }

builder = StateGraph(ResearchAgentState)
builder.add_node("search_docs", search_docs)
builder.add_node("search_web", search_web)
builder.add_edge(START, "search_docs")
builder.add_edge(START, "search_web")
builder.add_edge("search_docs", END)
builder.add_edge("search_web", END)

graph = builder.compile()
result = graph.invoke({"messages": [], "findings": {}, "tool_calls": 0})
print(result["findings"])     # {'docs': ..., 'web': ...}
print(result["tool_calls"])   # 2
print(len(result["messages"]))  # 2

Every field here has a reducer that matches its purpose: messages append and dedupe by ID, findings merge as a dict, tool call counts sum. That is the pattern to reach for by default, pick the reducer based on what "combining two updates" should mean for that specific field, not based on what is convenient to write.

FAQ

What is a state reducer in LangGraph? It is a function attached to a field in your state schema, via Annotated[type, reducer_fn], that defines how a node's returned update for that field gets combined with the existing value. Without one, LangGraph overwrites the field with whatever the node last returned.

What happens if I don't specify a reducer? The field uses default overwrite behavior. The newest write wins. This works fine for single-writer fields (a status string, a final answer) and breaks for fields that multiple nodes or parallel branches need to accumulate into, like message history or a results list.

Why do I get an `InvalidUpdateError` in a fan-out graph? Two or more nodes ran in the same superstep and both tried to write to a key that has no reducer (or an overwrite-only reducer), so LangGraph cannot determine how to merge the concurrent writes. Add a reducer, typically operator.add for lists or a custom merge function, to the field in your TypedDict or Pydantic state model.

Is `add_messages` just `operator.add` for lists? No. operator.add blindly concatenates. add_messages concatenates too, but it also deduplicates by message id (replacing an existing message with the same ID instead of duplicating it) and supports deleting messages via RemoveMessage. Use add_messages for any chat history field.

Can a reducer do async work or call an LLM? Don't. Reducers should be pure, synchronous functions of (current, new). They run as part of LangGraph's internal state application between graph steps. Side effects belong in nodes, not reducers.

Do reducers apply across checkpointed sessions, not just within one invoke call? Yes. If you use a checkpointer and resume a thread, the reducer merges new node output against the last checkpointed state for that thread ID, so accumulation (like growing message history) persists correctly across separate invoke calls on the same thread.

Can I use a different reducer for the same field in different nodes? No, the reducer is defined once on the state schema for a given field, and every node that writes to that field is subject to the same merge rule. If you need different merge behavior in different situations, use separate state fields and combine them in a later node instead of trying to vary the reducer per call site.

How do reducers interact with `operator.add` on non-list types? operator.add works on anything Python's + operator supports: ints and floats sum, strings concatenate, lists concatenate. It fails on None, so make sure the field's initial value is a proper empty container or zero, never None, if you plan to reduce it with operator.add.