LangGraph state and reducers

You build your first LangGraph chatbot, run two turns, and the second turn’s reply has no idea what the first turn said. You log the state: the messages list contains exactly one entry — the most recent one. The previous user message is gone. You didn’t lose it; the default reducer silently overwrote it. Welcome to your first lesson in reducers.

Every LangGraph workflow is glued together by state — a typed dict that flows from node to node. When a node returns an update, LangGraph has to decide how to merge it into the existing state. That merging logic is called a reducer: a function (old_value, new_value) -> merged_value registered for each field. Picking the right reducer for each field is one of the first real LangGraph design decisions you make, and getting it wrong is the most common beginner bug.

One important thing state is not: it is not a global mutable object. Nodes never modify state in place — they return a new dict of updates, and LangGraph applies the reducer. Treat your node functions as pure transformations.

State is a TypedDict

from typing import TypedDict

class State(TypedDict):
    question: str
    intermediate: list[str]
    answer: str

The TypedDict is your contract. Every node sees the same shape; every node’s return value is type-checked against it.

A node returns a partial update — just the keys it wants to change. LangGraph merges this back into the state before passing it to the next node.

def think(state: State) -> dict:
    return {"answer": "42"}   # only updates `answer`; other keys untouched

The default merge is overwrite: the returned answer replaces the existing one. That’s fine for fields that should hold a single latest value.

Append, not overwrite — the add_messages reducer

Chat-style agents accumulate messages across turns. If every node overwrote the message list, you’d lose history. The fix is to declare a custom reducer with Annotated.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

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

add_messages is a reducer function. When a node returns {"messages": [new_msg]}, LangGraph calls add_messages(old_list, [new_msg]) and uses the result. The reducer appends instead of overwriting.

This is the canonical agent state shape — you’ll see it in practically every LangGraph chatbot. Other useful built-in reducers and patterns:

• add_messages — append messages, also dedupes by ID and supports message removal.
• operator.add — concatenate lists (e.g. for accumulating tool results).
• A custom function — anything (old, new) -> merged.

Two nodes, one state

# Mocked LangGraph with state + reducer. The TypedDict / Annotated /
# add_messages pattern matches real LangGraph code.

from typing import TypedDict

# --- mocked LangGraph runtime ---
START, END = "__start__", "__end__"

def add_messages(old, new):
    """Append reducer — what langgraph.graph.message.add_messages does."""
    return (old or []) + (new or [])

class StateGraph:
    def __init__(self, schema, reducers=None):
        self.schema = schema
        self.reducers = reducers or {}
        self.nodes = {}
        self.edges = {}
    def add_node(self, name, fn): self.nodes[name] = fn
    def add_edge(self, src, dst): self.edges[src] = dst
    def compile(self): return CompiledGraph(self)

class CompiledGraph:
    def __init__(self, g): self.g = g
    def _merge(self, state, update):
        merged = dict(state)
        for k, v in update.items():
            if k in self.g.reducers:
                merged[k] = self.g.reducers[k](state.get(k), v)
            else:
                merged[k] = v   # overwrite (default)
        return merged
    def invoke(self, state):
        current = START
        while True:
            if current in self.g.edges:
                current = self.g.edges[current]
                if current == END: return state
                update = self.g.nodes[current](state)
                state = self._merge(state, update)
            else:
                return state

# --- the actual graph (this is the shape of real LangGraph code) ---
class State(TypedDict):
    messages: list   # annotated with add_messages in real LG
    turn: int        # plain — default overwrite reducer

def user_turn(state):
    # Imagine this came from a UI. We append a user message.
    return {
        "messages": [{"role": "user", "content": "What's 7 * 6?"}],
        "turn": state.get("turn", 0) + 1,
    }

def assistant_turn(state):
    last = state["messages"][-1]["content"]
    answer = "42" if "7 * 6" in last else "[mock answer]"
    return {
        "messages": [{"role": "assistant", "content": answer}],
        "turn": state["turn"] + 1,
    }

# Declare which keys use which reducer
graph = StateGraph(State, reducers={"messages": add_messages})
graph.add_node("user", user_turn)
graph.add_node("assistant", assistant_turn)
graph.add_edge(START, "user")
graph.add_edge("user", "assistant")
graph.add_edge("assistant", END)

app = graph.compile()
final = app.invoke({"messages": [], "turn": 0})

print(f"turns taken: {final['turn']}")
print("conversation:")
for m in final["messages"]:
    print(f"  [{m['role']}] {m['content']}")

turns taken: 2
conversation:
  [user] What's 7 * 6?
  [assistant] 42

Trace the messages key: each node returns a one-element list, and because of add_messages, both messages survive in the final state — the conversation accumulated. The turn key, with no reducer, overwrites — so the final value is 2 (the assistant’s last write), not 1 + 2 = 3. That split is the whole lesson: same graph, two fields, two different merge behaviours.

If you forgot the reducer on messages, the assistant turn would overwrite the user message and the history would only have one entry. This is the most common LangGraph beginner bug.

Multi-channel state

Real agent state typically has several “channels,” each with its own reducer:

class AgentState(TypedDict):
    messages:       Annotated[list, add_messages]   # conversation history
    scratchpad:     Annotated[list, operator.add]   # internal reasoning steps
    retrieved_docs: list                            # latest retrieval only
    user_metadata:  dict                            # set once at start

Each field’s reducer should match its semantics:

• Append history → add_messages or operator.add.
• Latest value only → no reducer (overwrite).
• Set-once → no reducer; just don’t update it after the first node.

Picking these correctly up front saves a lot of debugging later.

The shape of a real agent

For a tool-using chat agent, the state is almost always:

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

That’s it. Everything else — the user question, the model’s tool calls, the tool results, the final answer — lives as messages in the list. The conversation is the state.

The graph has two nodes:

• agent — calls the LLM with the current messages, gets back either a text answer or one or more tool calls.
• tools — looks at the last message’s tool calls, executes them, returns the results as new messages.

A conditional edge from agent routes to either tools (if the LLM called a tool) or END (if it returned text). After tools, the edge goes back to agent. That loop is the entire structure of every chat agent you’ll build.

START → agent → (has tool calls?) → tools → agent → ... → END
                                          (no tool calls) → END

The add_messages reducer is what makes this work — the conversation accumulates as nodes append messages, and the LLM sees the full history every time the agent node runs.

Updating state without a node

Sometimes you want to modify state from outside the graph — for example, after a human edits a draft. The update_state method on a compiled graph lets you do that.

config = {"configurable": {"thread_id": "user-42"}}
app.update_state(config, {"messages": [{"role": "user", "content": "actually, in French"}]})

This requires a checkpointer (next lesson). Without one, there’s no “saved state” to update.

In one breath

• State is a TypedDict that flows node to node; a node returns a partial update, and a reducer (old, new) -> merged decides how each field merges.
• The default reducer is overwrite — fine for “latest value only” fields; the silent overwrite of a field that should accumulate is the #1 beginner bug.
• add_messages is the append (and dedupe-by-id) reducer that makes conversation history accumulate — declare it with Annotated[list, add_messages].
• Real state is multi-channel: each field’s reducer should match its semantics (append history, overwrite latest, set-once).
• The canonical agent state is just messages: Annotated[list, add_messages] — the conversation is the state; never mutate it in place, return a new dict.

Quick check

Next

State that lives only in memory is fine for demos. Real agents need to survive process restarts, pause for hours, and let you debug what happened on turn 17. That’s what checkpointers do — the next lesson.