State and reducers

State is the only thing every node shares. Get the schema right and the graph writes itself; get it wrong and you will fight the runtime for a week.

The schema

from typing_extensions import TypedDict

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

TypedDict is the recommended shape. A dataclass gives you defaults, and a Pydantic BaseModel gives you validation on every update — useful when untrusted input reaches the graph, at the cost of validation on each step.

A node returns a partial update, not the whole state:

def retrieve(state: State) -> dict:
    docs = search(state["question"])
    return {"documents": docs}

Return only what changed. The runtime merges it. Returning the full state works and is a habit worth breaking — it makes the diffs in a trace useless.

Reducers: the concept that trips everyone

By default, an update to a key replaces that key. That is fine for answer, and wrong for anything accumulating.

A reducer says how to combine the old value with the update:

from typing import Annotated
from operator import add

class State(TypedDict):
    question: str
    logs: Annotated[list[str], add]

Now return {"logs": ["retrieved 5 docs"]} appends instead of replacing.

That last row is the one that bites. Two nodes running in parallel both writing the same key without a reducer is a conflict, and the runtime will tell you so. With a reducer it is a merge. So the reducer is not a convenience — it is how you declare that a key is safe to write concurrently.

add_messages, the one you will actually use

Conversation history needs more than append: it needs updates by id, so a node can revise a message rather than duplicate it.

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

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

add_messages appends new messages and replaces any message whose id matches one already present. That is what lets you correct or remove a turn without rebuilding the list.

The shortcut, when messages are all you need:

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

MessagesState is a TypedDict with messages already annotated. Subclass it and add your own keys.

Designing the schema

Three rules that prevent most rewrites.

1. State is the contract between nodes, not a scratchpad. If only one node reads a key, it should probably be a local variable. Everything in state is checkpointed, streamed and inspected, so a key that exists for one node’s convenience is noise in every trace.

2. Keep it serialisable. State is persisted between steps, so a database connection or an open file handle does not belong in it. Put the id in state and resolve the object inside the node.

3. Watch the size. Every checkpoint writes the whole state. Accumulating raw documents in state across twenty steps means twenty copies. Store an id or a summary and fetch the body when needed — the same discipline as Context engineering.

Gotcha: a key with no reducer written by two parallel branches raises an error at runtime, not at build time. If you fan out, every key the branches write needs a reducer — that is the design question fan-out forces.

Custom reducers

When add is not the semantics you want, write one. It takes the current value and the update, and returns the new value:

def keep_best(current: dict, update: dict) -> dict:
    """Keep whichever candidate scored higher."""
    if not current:
        return update
    return max(
        current, update, key=lambda d: d["score"]
    )

class State(TypedDict):
    best: Annotated[dict, keep_best]

This is the clean way to express “several workers propose, one answer survives” — the merge rule lives in the schema rather than in a node that has to know about its siblings.

Related

• Nodes, edges and routing — what consumes this state
• Persistence and threads — where it gets written
• Context engineering — why state size is a cost