Reducers: How Parallel Writes to the Same State Field Combine

Module 10 ended with an unresolved, real bug: two parallel branches, both writing to findings, with no explanation of what actually happens. Let’s not explain reducers abstractly before seeing that problem clearly. Let’s watch it actually break first.

Watching the collision happen, directly

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

class State(TypedDict):
    topics: list[str]
    results: list[str]

class BranchState(TypedDict):
    topic: str
    results: list[str]

def research_one(state: BranchState) -> dict:
    return {"results": [f"Result for {state['topic']}"]}

def fan_out(state: State) -> list[Send]:
    return [Send("research_one", {"topic": t, "results": []}) for t in state["topics"]]

builder = StateGraph(State)
builder.add_node("research_one", research_one)
builder.add_conditional_edges(START, fan_out, ["research_one"])
builder.add_edge("research_one", END)

graph = builder.compile()
print(graph.invoke({"topics": ["LLMs", "RAG"], "results": []}))

Run this and look closely at results in the printed output. Depending on your current LangGraph version, this either raises a real, genuine error about concurrent updates to the same key, or — more dangerously — silently keeps only one of the two branches’ results, with the other one simply vanishing. Neither is what you actually wanted. Both branches genuinely computed a real, useful result; the state schema, as written, gave LangGraph no instructions for what to do when two updates arrive for the same field at the same time.

The fix: telling LangGraph how to combine, not just what to store

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

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

class BranchState(TypedDict):
    topic: str
    results: list[str]

def research_one(state: BranchState) -> dict:
    return {"results": [f"Result for {state['topic']}"]}

def fan_out(state: State) -> list[Send]:
    return [Send("research_one", {"topic": t, "results": []}) for t in state["topics"]]

builder = StateGraph(State)
builder.add_node("research_one", research_one)
builder.add_conditional_edges(START, fan_out, ["research_one"])
builder.add_edge("research_one", END)

graph = builder.compile()
print(graph.invoke({"topics": ["LLMs", "RAG"], "results": []}))

The only real change is Annotated[list[str], operator.add] on the results field. Run this, and both results genuinely survive: ["Result for LLMs", "Result for RAG"]. Annotated[type, reducer_function] is telling LangGraph something completely different from a plain type hint: when more than one update arrives for this field, don’t just take the newest one — combine them, using this specific function. operator.add, applied to two lists, is simply concatenation — exactly the accumulation behavior this field genuinely needed.

Seeing this visually, before and after

flowchart TD
    A["Without a reducer:\nsecond write silently\nreplaces the first"] --> B["results: ['Result for RAG']\n(LLMs result lost)"]
    C["With Annotated + operator.add:\nwrites genuinely combine"] --> D["results: ['Result for LLMs',\n'Result for RAG']\n(both survive)"]

Example: a custom reducer for merging dictionaries

operator.add handles lists genuinely well. Some real accumulation needs are more specific, and you can write your own reducer function directly.

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

def merge_dicts(existing: dict, new: dict) -> dict:
    return {**existing, **new}

class State(TypedDict):
    topics: list[str]
    findings_by_topic: Annotated[dict, merge_dicts]

class BranchState(TypedDict):
    topic: str
    findings_by_topic: dict

def research_one(state: BranchState) -> dict:
    return {"findings_by_topic": {state["topic"]: f"Detail on {state['topic']}"}}

def fan_out(state: State) -> list[Send]:
    return [Send("research_one", {"topic": t, "findings_by_topic": {}}) for t in state["topics"]]

builder = StateGraph(State)
builder.add_node("research_one", research_one)
builder.add_conditional_edges(START, fan_out, ["research_one"])
builder.add_edge("research_one", END)

graph = builder.compile()
print(graph.invoke({"topics": ["LLMs", "RAG"], "findings_by_topic": {}}))

merge_dicts is genuinely just a plain Python function — take the existing dictionary, take the new one, combine them. Annotated[dict, merge_dicts] tells LangGraph to use this specific function whenever two updates to findings_by_topic need to be reconciled, rather than the default overwrite behavior.

Example: a reducer that keeps the maximum value

Not every accumulation need is “combine everything.” Sometimes you genuinely want to keep only the most significant of several parallel results.

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

def keep_max(existing: int, new: int) -> int:
    return max(existing, new)

class State(TypedDict):
    checks: list[str]
    highest_risk_score: Annotated[int, keep_max]

class BranchState(TypedDict):
    check: str
    highest_risk_score: int

def run_check(state: BranchState) -> dict:
    scores = {"fraud_check": 20, "credit_check": 75, "identity_check": 40}
    return {"highest_risk_score": scores.get(state["check"], 0)}

def fan_out(state: State) -> list[Send]:
    return [Send("run_check", {"check": c, "highest_risk_score": 0}) for c in state["checks"]]

builder = StateGraph(State)
builder.add_node("run_check", run_check)
builder.add_conditional_edges(START, fan_out, ["run_check"])
builder.add_edge("run_check", END)

graph = builder.compile()
print(graph.invoke({"checks": ["fraud_check", "credit_check", "identity_check"], "highest_risk_score": 0}))

Three parallel checks, each computing its own real risk score — and keep_max ensures the final state reflects the single highest score among all three, genuinely useful for a real risk-assessment workflow where you care about the worst case, not an accumulated total.

A preview: reducers and message history

You’ll meet this properly in the next module, but it’s worth naming now, since it’s exactly this same mechanism, applied to the single most common real use case: add_messages, a genuine, prebuilt reducer specifically for message lists — appending new messages to existing conversation history, rather than replacing it. Everything you just learned about writing your own reducer is the same real mechanism add_messages uses internally.

Common mistakes worth avoiding

Assuming every list field needs operator.add. Recall the maximum-score example — not every accumulation problem is “combine everything.” A field genuinely representing “the single most important result” needs a reducer like keep_max, not concatenation.

Writing a reducer with the wrong argument order. A reducer function’s real signature is (existing_value, new_value) -> combined_value — reversing this, or assuming it works like Python’s built-in sum(), produces a genuinely confusing, silently wrong result rather than an obvious error.

Forgetting a reducer entirely on a field multiple parallel branches will write to. Recall this module’s opening example — the default behavior when no reducer exists isn’t graceful; it’s either a real error or silent data loss, neither of which announces itself clearly until you specifically go looking for missing results.

What you should take away from this module

• Without a reducer, two parallel writes to the same field either raise a real error or silently overwrite each other — this is the exact bug Module 10 left open, and it’s genuinely dangerous precisely because it can fail silently.
• Annotated[type, reducer_function] tells LangGraph how to combine multiple updates to one field, rather than simply replacing the old value with the new one.
• operator.add handles list concatenation; custom functions like merge_dicts or keep_max handle genuinely different, more specific accumulation needs.
• add_messages, covered fully next, is this exact same mechanism, applied specifically to conversation history.

Where this goes next

The next module covers Messages and MessagesState properly — connecting the LangChain messages you already know deeply to LangGraph state, using exactly the reducer mechanism you just learned, built specifically for the most common real accumulation need of all: a growing conversation.