AI Agents LangGraph

Reducers

Intermediate

Reducers

This post covers everything you need to know about Reducers in LangGraph — how they manage and merge state updates across nodes, especially during parallel execution. We explore built-in and custom reducers, conflict resolution strategies, reducers for messages and cyclic graphs, common mistakes developers make, and best practices for building reliable and predictable LangGraph workflows.

Reducers in LangGraph

Reducers are one of the most important concepts in LangGraph. They define how state updates from different nodes are merged into the shared state.

What Are Reducers?

A Reducer is a function that tells LangGraph how to combine a new update with the existing value of a state field. When multiple nodes try to update the same field (especially in parallel execution or cycles), LangGraph uses the reducer to decide the final value. Without reducers, LangGraph wouldn’t know how to safely merge updates, especially for lists like messages.

Why Reducers Are Needed

Imagine two nodes running in parallel both trying to update the messages list:
  • Node A returns: {"messages": [msg1]}
  • Node B returns: {"messages": [msg2]}
Without a reducer, LangGraph wouldn’t know whether to:
  • Replace the list?
  • Append both?
  • Take the last one?
Reducers solve this problem by defining clear merging rules. They enable:
  • Safe concurrent updates
  • Predictable behavior in cycles and parallel flows
  • Clean accumulation of messages, documents, logs, etc.

How Reducers Work

You define reducers using Annotated when declaring your state: 
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from operator import add

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]           # ← Reducer
    documents: Annotated[list[dict], add]             # ← Reducer
    iterations: int                                   # No reducer = last write wins
    metadata: Annotated[dict, lambda a, b: {**a, **b}] # Custom reducer
Execution Flow:
  1. Node returns partial update → {"messages": [new_msg]}
  2. LangGraph calls the reducer: add_messages(old_messages, [new_msg])
  3. Result becomes the new value in state

Built-in Reducer Patterns

1. add_messages (Most Important) 

from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
Used for conversation history. Automatically handles AIMessage , HumanMessage , ToolMessage , etc.

2. operator.add (for lists) 

from operator import add

class AgentState(TypedDict):
    documents: Annotated[list[dict], add]        # Concatenates lists
    tool_calls: Annotated[list, add]

3. Last Write Wins (Default)

If no reducer is specified, the last update wins: 
class AgentState(TypedDict):
    current_status: str        # Last node to write wins
    final_answer: str | None

Custom Reducers

You can create your own reducers for complex merging logic. 
def merge_dicts(left: dict, right: dict) -> dict:
    """Merge two dictionaries, right takes precedence"""
    return {**left, **right}

def deduplicate_list(left: list, right: list) -> list:
    """Append only new items"""
    seen = {id(x) for x in left}
    return left + [x for x in right if id(x) not in seen]

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    metadata: Annotated[dict, merge_dicts]
    unique_docs: Annotated[list[dict], deduplicate_list]
Advanced Custom Reducer with Pydantic: 
from pydantic import BaseModel

def merge_tool_results(left: list, right: list):
    return left + [r for r in right if r not in left]

class AgentState(BaseModel):
    tool_results: Annotated[list[dict], merge_tool_results] = Field(default_factory=list)

Reducers in Parallel Execution

Reducers become critical when nodes run in parallel. 
from langgraph.types import Send

def parallel_router(state):
    return [
        Send("web_search", state),
        Send("vector_search", state),
        Send("news_search", state)
    ]

# All three nodes can update the same field safely
class AgentState(TypedDict):
    documents: Annotated[list[dict], add]   # Reducer handles parallel updates
    sources: Annotated[list[str], add]

graph.add_conditional_edges(START, parallel_router)
graph.add_edge("web_search", "aggregate")
graph.add_edge("vector_search", "aggregate")
graph.add_edge("news_search", "aggregate")
Without a proper reducer like add , parallel updates would conflict or overwrite each other.
Key Takeaways:
  • Reducers define how state fields are merged
  • add_messages is the most commonly used reducer
  • Always use reducers for lists and complex data types
  • Custom reducers give you full control over merging logic
  • Reducers are essential for safe parallel and cyclic execution

Conflict Resolution in Parallel Updates

One of the most powerful features of reducers is how they handle parallel execution . When multiple nodes run at the same time and update the same state field, reducers resolve conflicts automatically. 
from langgraph.types import Send
from operator import add

class AgentState(TypedDict):
    documents: Annotated[list[dict], add]           # Safe for parallel updates
    sources: Annotated[list[str], add]
    messages: Annotated[list, add_messages]

def parallel_router(state):
    """Launch multiple searches in parallel"""
    return [
        Send("web_search", state),
        Send("vector_search", state),
        Send("news_search", state)
    ]

graph.add_conditional_edges(START, parallel_router)

# All three nodes can safely return updates
def web_search(state):
    results = ...
    return {"documents": results, "sources": ["web"]}

def vector_search(state):
    results = ...
    return {"documents": results, "sources": ["vector"]}
How Conflict Resolution Works:
  • add → Concatenates lists from all parallel branches
  • add_messages → Intelligently merges messages (preserves order and metadata)
  • Custom reducer → You define the exact merging logic
Without reducers, parallel updates would cause data loss or race conditions.
Reducers with Messages ( add_messages )
This is the most important reducer in LangGraph. 
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
How add_messages works:
  • Appends new messages to the conversation history
  • Handles different message types (HumanMessage, AIMessage, ToolMessage)
  • Preserves message metadata and tool call IDs
  • Prevents duplicate messages in some cases
Example in Action: 
def agent_node(state: AgentState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}          # add_messages will append

def tools_node(state: AgentState):
    tool_results = execute_tools(...)
    return {"messages": [ToolMessage(...)]}  # Automatically appended
Advanced Usage: 
# You can also pass multiple messages
return {
    "messages": [
        AIMessage(content="Let me check that..."),
        ToolMessage(content="Tool result here", tool_call_id="123")
    ]
}

Reducers in Cyclic Graphs

Reducers are essential in loops because the same fields (especially messages) get updated repeatedly. 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    iterations: int = 0

def agent_node(state: AgentState):
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "iterations": state.get("iterations", 0) + 1
    }

def router(state: AgentState):
    if state.get("iterations", 0) >= 12:
        return "END"
    if state["messages"][-1].tool_calls:
        return "tools"
    return "END"
Why reducers shine in cycles:
  • add_messages safely accumulates conversation history across many iterations
  • You can track progress with counters
  • State remains consistent even after many loops

Common Reducer Mistakes

1. Forgetting Reducers on Lists 

# Wrong
documents: list[dict]          # Last write wins → loses data in parallel

2. Using Mutable Defaults 

# Dangerous
documents: list = []           # Shared mutable default!
Correct: 
documents: list[dict] = Field(default_factory=list)   # Pydantic
# or
documents: Annotated[list, add]                       # TypedDict

3. Overcomplicating Custom Reducers

Creating overly complex reducers instead of using simple ones like add .

4. Returning Full State Instead of Partial 

# Bad
return state.copy()          # Risky and inefficient

Best Practices for Reducers

Always use reducers for lists and complex types 

messages: Annotated[list, add_messages]
documents: Annotated[list[dict], add]

Prefer built-in reducers ( add_messages , add ) when possible

Keep custom reducers simple and pure 

def merge_metadata(left: dict, right: dict):
    return {**left, **right}

Use Pydantic Field(default_factory=...) for better defaults

Document your reducers 

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]           # Conversation history
    tool_outputs: Annotated[list, add]                # All tool results
    metadata: Annotated[dict, merge_metadata]         # Control data

Test reducers independently during development

Reducers are the glue that makes stateful, parallel, and cyclic execution safe and predictable in LangGraph.
Mastering reducers is essential for building robust, scalable AI agents.

AI agent LangChain LangGraph Python

← All training