Chapter 4: Reducers and State Aggregation

In Chapter 3, we mastered Overwrite Semantics. When a node returns {"status": "failed"}, the Graph Runtime overwrites the existing status key in the global state.

This works perfectly for strings, booleans, and integers. But what happens when you need to maintain a list of items? What if you want to keep track of every error that occurs during a complex execution?

If you rely on Overwrite Semantics for lists, you will experience catastrophic data loss.

In this chapter, we will solve the list problem using Reducers and State Aggregation.

---

4.1 The Problem with Lists

Let's imagine we define a state that tracks errors during execution:

from typing import TypedDict

class SystemState(TypedDict):
    errors: list[str]

Watch what happens when two nodes execute sequentially under default Overwrite Semantics.

Incoming State:

{
    "errors": ["Database timeout"]
}

Node B Encounters a Bug and Returns:

# Node B returns a dictionary to update the state
return {"errors": ["API key invalid"]}

The Graph Runtime Executes the Merge: Because the default behavior is Overwrite, the runtime takes the new list and completely replaces the old list.

  ["Database timeout"]
           ↓ Overwrites existing list!
  ["API key invalid"]

Final State (Data Loss!):

{
    "errors": ["API key invalid"] // We lost the "Database timeout" error!
}

We do not want to overwrite the list. We want to append to the list.

---

4.2 Introducing Reducers (operator.add)

A Reducer is a function attached to a specific field in your State schema. It tells the Graph Runtime exactly how to merge incoming updates for that specific key.

Crucial Concept: Reducers are per-key. Attaching a reducer to one field does not change how the rest of your graph behaves. One field can use overwrite semantics while another uses aggregation semantics.

To instruct LangGraph to append to a list instead of overwriting it, we use Python's built-in operator.add function combined with the Annotated type hint.

from typing import TypedDict, Annotated
import operator

class ResilientState(TypedDict):
    # Standard field: Uses default Overwrite semantics
    status: str 
    
    # Reduced field: Uses 'operator.add' to append new items to the existing list!
    errors: Annotated[list[str], operator.add]

When you wrap a field in Annotated[..., reducer_function], you change the fundamental physics of how that key is merged.

---

4.3 Visualizing Reducers

Let's watch the magic of Reducers in action with our new ResilientState.

Incoming State:

{
    "status": "running",
    "errors": ["Database timeout"]
}

Node B Returns:

return {
    "status": "failed",
    "errors": ["API key invalid"]
}

The Graph Runtime Executes the Merge:

1. For status, it sees no reducer. It overwrites.
2. For errors, it sees operator.add. It physically executes: operator.add(["Database timeout"], ["API key invalid"]). In Python, adding two lists together concatenates them!

Current Errors: ["Database timeout"]

Node Update:    ["API key invalid"]

Reducer (Add):  ["Database timeout"] + ["API key invalid"]
                      ↓

Final State:    ["Database timeout", "API key invalid"]

By simply adding Annotated[list, operator.add], we have completely solved the data loss problem. Multiple nodes can now safely contribute items to the same list.

---

4.4 Why Reducers Matter for Parallel Graphs

While reducers solve data loss in sequential executions, their true superpower is unlocked during Parallel Execution.

Imagine an agent that needs to search both a SQL database and the public Web concurrently to answer a user's question.

        Search Web
       /          \
Question           Combine Results
       \          /
        Search DB

Both the Web Node and the DB Node will finish at the exact same time and both will attempt to update the documents list in the State simultaneously.

If you don't use a Reducer:

• Last Writer Wins. Whichever node finishes 1 millisecond later will blindly overwrite the other node's documents. Data is lost.

If you DO use a Reducer (Annotated[list, operator.add]):

• Results are Combined. The Graph Runtime will capture both incoming lists and safely concatenate them together. [Web Docs] + [DB Docs].

---

4.5 Custom Reducers (Preserving Order)

operator.add is just a standard Python function. You can write your own custom reducers to implement advanced state aggregation logic.

What if you want to keep a list of errors, but you want to ensure there are no duplicate errors? You could use a set(), but sets destroy the original order of the list, which is catastrophic for logs or chat histories.

Here is a proper custom reducer that deduplicates while preserving order:

from typing import TypedDict, Annotated

# A custom reducer function
def deduplicate_merge(existing_list: list[str], new_list: list[str]) -> list[str]:
    """Combines two lists and removes duplicates while preserving order."""
    if existing_list is None:
        existing_list = []
    if new_list is None:
        new_list = []
        
    seen = set()
    result = []
    
    # Iterate through both lists in order
    for item in existing_list + new_list:
        if item not in seen:
            seen.add(item)
            result.append(item)
            
    return result

class UniqueErrorState(TypedDict):
    # Attach our custom reducer function!
    errors: Annotated[list[str], deduplicate_merge]

Now, if Node A returns ["Timeout"] and Node B returns ["Timeout"], the final state will only contain ["Timeout"] once, exactly in the order it was received.

---

4.6 Warning: Avoid Mutating State Directly

A massive mistake beginners make is treating the incoming state dictionary as a mutable variable that they can change directly inside the node.

ANTI-PATTERN (DO NOT DO THIS):

def bad_node(state: ResilientState):
    # Modifying the incoming state in-place is highly dangerous!
    state["errors"].append("My new error")
    return state

PRO-PATTERN:

def good_node(state: ResilientState):
    # Nodes should ONLY return the updates. The Graph Runtime handles the merging.
    return {"errors": ["My new error"]}

If you mutate the state in place, you completely bypass the Graph Runtime's checkpointer, ruin parallel execution, and create nightmare bugs. Always return a dictionary of updates.

---

4.7 Reducers vs Overwrite Semantics (Reference Table)

As a rule of thumb, use this table when designing your State schema:

Field Type	Recommended Merge	Example Use Case
str	Overwrite	User's name, active department
bool	Overwrite	Is the ticket resolved?
int / float	Overwrite	Product price, active retry count
list	Reducer (operator.add)	Error logs, retrieved documents
messages	Specialized Reducer	LLM Chat History (See Chapter 6)

---

4.8 Building an Error Aggregation Graph

Let's build a functional graph that proves how reducers allow multiple isolated nodes to safely aggregate data into a shared list.

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

# 1. State Definition using a Reducer!
class AggregationState(TypedDict):
    ticket_id: int
    # Without operator.add, this graph would lose data!
    logs: Annotated[list[str], operator.add]

# 2. Define the Nodes
def database_check_node(state: AggregationState):
    # Simulate some work
    print("Checking Database...")
    
    # We return a list containing ONE new log entry
    return {"logs": ["Database connected successfully."]}

def api_check_node(state: AggregationState):
    # Simulate some work
    print("Checking API...")
    
    # We return a list containing ONE new log entry
    return {"logs": ["API responded in 200ms."]}

def final_report_node(state: AggregationState):
    print("\n--- FINAL SYSTEM LOGS ---")
    for log in state["logs"]:
        print(f"- {log}")
    return {}

# 3. Build the Graph
builder = StateGraph(AggregationState)

builder.add_node("db_checker", database_check_node)
builder.add_node("api_checker", api_check_node)
builder.add_node("reporter", final_report_node)

# Execution: Start -> DB -> API -> Reporter -> End
builder.add_edge(START, "db_checker")
builder.add_edge("db_checker", "api_checker")
builder.add_edge("api_checker", "reporter")
builder.add_edge("reporter", END)

app = builder.compile()

# 4. Execute!
result = app.invoke({
    "ticket_id": 999,
    "logs": ["System Boot Sequence Initiated."] # Initial state
})

# Output:
# Checking Database...
# Checking API...
# 
# --- FINAL SYSTEM LOGS ---
# - System Boot Sequence Initiated.
# - Database connected successfully.
# - API responded in 200ms.

Notice that neither the DB Node nor the API Node had to read the existing logs. They simply returned {"logs": ["My new log"]}. The Graph Runtime and the operator.add reducer handled the complex merging behind the scenes.

4.9 Summary

Reducers are the secret to building massive, resilient graphs where multiple autonomous nodes contribute information to a shared pool over time.

You now understand Overwrite Semantics and Reducer Aggregation. You possess the complete mental and technical model of LangGraph's State Layer.

Now that we understand how data flows, we need to master how control flows. In the next chapter, we will learn the APIs for building topology: Nodes, Edges, Branching, and Loops.