LangGraph State Schemas: Designing Your Agent's Shared Memory

Why your agent's memory design decides everything else

Most people building their first LangGraph agent spend a week worrying about prompts and about ten minutes deciding what the state schema looks like. Then two weeks later they're debugging why the conversation history keeps overwriting itself, why two parallel nodes clobber each other's writes, or why the supervisor agent can't see what the researcher node just found.

The state schema is not a formality you fill in before the "real" work starts. It is the real work. Every node in a LangGraph graph reads from and writes to a shared state object, and the schema you define determines what data survives between steps, how conflicting updates get merged, and whether your agent can be paused, resumed, or handed off between humans and other agents. Get the schema right, and the rest of the graph — nodes, edges, conditional routing — becomes almost mechanical. Get it wrong, and you'll be firefighting race conditions and silent data loss in production.

This article walks through how LangGraph state schemas actually work under the hood, the difference between plain field replacement and reducer-based merging, how to structure state for multi-agent systems, and the mistakes that trip up almost everyone the first time they build something beyond a toy chatbot.

The building block: TypedDict as your state contract

LangGraph represents state as a structured object that flows through your graph. The most common way to define it is with Python's TypedDict, though dataclasses and Pydantic models work too. TypedDict is the pragmatic default because it's lightweight, has zero runtime overhead, and plays well with LangGraph's StateGraph compiler.

Here's a minimal schema for a research agent:

from typing import TypedDict, List
from langchain_core.messages import BaseMessage

class ResearchState(TypedDict):
    question: str
    messages: List[BaseMessage]
    search_results: List[str]
    draft_answer: str
    final_answer: str

Every node function in your graph takes this state as input and returns a partial dict representing the fields it wants to update:

def search_node(state: ResearchState) -> dict:
    results = run_web_search(state["question"])
    return {"search_results": results}

def draft_node(state: ResearchState) -> dict:
    draft = generate_draft(state["question"], state["search_results"])
    return {"draft_answer": draft}

The critical thing to understand: by default, LangGraph overwrites fields. If search_node returns {"search_results": results}, the new value fully replaces whatever was in search_results before. That's fine for draft_answer — you want the latest draft, not a pile of old drafts. It's a disaster for something like messages, where you want to append, not replace.

This is where reducers come in, and it's the single most important concept in state schema design.

Reducers: controlling how updates merge

A reducer is a function attached to a state field via Annotated that tells LangGraph how to combine the existing value with the new value returned by a node. Instead of blind overwrite, you get controlled merging.

The most common reducer is add_messages, built into LangGraph specifically for chat history:

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

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

With add_messages, when a node returns {"messages": [new_message]}, LangGraph appends new_message to the existing list rather than replacing it. It also handles deduplication by message ID and can update an existing message in place if you return one with a matching ID — useful for streaming token updates into the same message object.

You're not limited to the built-in reducer. Any function with the signature (existing_value, new_value) -> merged_value works. A simple accumulator for a list of tool calls:

import operator
from typing import Annotated

class AgentState(TypedDict):
    question: str
    tool_calls: Annotated[list[str], operator.add]
    total_tokens_used: Annotated[int, operator.add]

Here operator.add on a list concatenates, and on an int it sums. This is a common pattern for running totals — token counts, retry counters, accumulated costs — where every node contributes a delta and the schema handles the aggregation for you. Without the reducer, the last node to write total_tokens_used would silently erase every prior node's count.

You can also write custom reducers for more nuanced merge logic, like deduplicating a set of visited URLs during a crawl:

def merge_unique(existing: list[str], new: list[str]) -> list[str]:
    seen = set(existing)
    merged = list(existing)
    for item in new:
        if item not in seen:
            merged.append(item)
            seen.add(item)
    return merged

class CrawlState(TypedDict):
    visited_urls: Annotated[list[str], merge_unique]
    pages_content: dict[str, str]

The rule of thumb: if a field is something nodes contribute to incrementally (messages, logs, accumulated results, running counters), it needs a reducer. If a field represents "the current value of X" that should just be replaced (a draft answer, a routing decision, a status flag), leave it as plain overwrite.

When TypedDict isn't enough: Pydantic and runtime validation

TypedDict gives you static typing for editor support and readability, but it does nothing at runtime — Python happily lets a node return {"question": 42} even though your schema says question should be a string. For small internal graphs this is a fine tradeoff. For anything customer-facing, or any graph where node outputs come from an LLM's structured output and might not match your expectations, that gap is dangerous.

LangGraph also accepts Pydantic models as the state schema, which gives you actual validation at graph invocation time:

from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class ValidatedState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    question: str
    confidence_score: float = Field(ge=0.0, le=1.0)
    retry_count: int = Field(default=0, ge=0)

graph = StateGraph(ValidatedState)

With this schema, if a node tries to set confidence_score to 1.5, Pydantic raises a validation error immediately rather than letting a nonsensical value silently flow downstream into a routing decision. This matters a lot in graphs where a numeric field drives a conditional edge — a confidence_score above 1.0 might accidentally satisfy a threshold check that should never have passed.

The tradeoff is speed and rigidity. Pydantic validation adds overhead on every state update, and it means every node's return value must satisfy the full model's constraints, not just the fields it touched. In practice, a reasonable rule is: use plain TypedDict for internal, fast-iterating agent graphs where you trust your own node code, and reach for Pydantic when the graph accepts state from external input, from another team's service, or from an LLM's structured output where malformed data is a real risk.

You can also mix approaches — keep the top-level StateGraph schema as a TypedDict for speed, but validate specific high-risk fields with a nested Pydantic model:

from pydantic import BaseModel

class ToolCallPayload(BaseModel):
    tool_name: str
    arguments: dict
    max_retries: int = Field(default=3, ge=0, le=5)

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    pending_tool_call: ToolCallPayload | None

This gives you validation exactly where an LLM's output is least predictable — the arguments to a tool call — without paying the Pydantic overhead across every field in the graph.

Building and wiring the StateGraph

Once the schema exists, wiring it into a graph is straightforward. StateGraph takes your TypedDict class as the type parameter, and every node you add is checked against that schema at compile time for basic shape correctness.

from langgraph.graph import StateGraph, START, END

def build_graph():
    graph = StateGraph(ResearchState)

    graph.add_node("search", search_node)
    graph.add_node("draft", draft_node)
    graph.add_node("review", review_node)

    graph.add_edge(START, "search")
    graph.add_edge("search", "draft")
    graph.add_edge("draft", "review")
    graph.add_edge("review", END)

    return graph.compile()

app = build_graph()
result = app.invoke({
    "question": "What is retrieval-augmented generation?",
    "messages": [],
    "search_results": [],
    "draft_answer": "",
    "final_answer": "",
})

Notice the initial call to invoke supplies every field, even the empty ones. LangGraph doesn't require this in all cases (missing keys are fine if nothing reads them before they're written), but it's good discipline — it documents the full shape of your state up front and avoids KeyError surprises when a node reads a field before another node has populated it.

Input schema, output schema, and internal schema

A pattern that saves a lot of pain in real applications: don't make your full internal state schema the same thing your API consumers see. LangGraph lets you define separate input and output schemas that are subsets of your main state.

class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    final_answer: str

class InternalState(TypedDict):
    question: str
    messages: Annotated[list[BaseMessage], add_messages]
    search_results: List[str]
    draft_answer: str
    final_answer: str

graph = StateGraph(InternalState, input=InputState, output=OutputState)

This matters more than it looks. Your internal state might carry scratch fields — intermediate search results, retry counters, debug traces — that you never want to expose to a caller of your graph, whether that caller is a frontend, an API gateway, or another agent. Defining a narrow output schema means app.invoke(...) only returns final_answer, keeping your public contract stable even as you add more internal bookkeeping fields later. Change the internal schema all you want; the input/output contract stays put.

Private state per node with schema composition

Not every field belongs in the global state. Sometimes a node needs scratch space that no other node should see or depend on. LangGraph supports this by letting individual nodes declare their own narrower state type that only overlaps with the global schema on the fields they actually need.

class SearchOnlyState(TypedDict):
    question: str
    search_results: List[str]
    retry_count: int

def search_node(state: SearchOnlyState) -> dict:
    if state.get("retry_count", 0) > 3:
        return {"search_results": []}
    results = run_web_search(state["question"])
    return {"search_results": results, "retry_count": state.get("retry_count", 0) + 1}

Because StateGraph merges whatever a node returns into the shared state dict, you can type a node's function signature narrowly without declaring retry_count in the graph's top-level schema — as long as you do want it tracked globally, add it there too. The practical benefit is readability: a node's type signature becomes documentation of exactly what it reads and writes, instead of every node silently depending on the entire god-object state.

Designing state for multi-agent systems

Multi-agent graphs — a supervisor delegating to a researcher, a coder, and a reviewer — are where state schema design either saves you or buries you. The naive approach is one giant shared state dict that every subagent reads and writes. It works for demos and becomes unmanageable fast: agents overwrite each other's scratch fields, and it's unclear who "owns" which part of the state.

A more robust pattern is to nest per-agent state under its own key, using a reducer only where it's genuinely shared:

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

class ResearcherState(TypedDict):
    findings: List[str]
    sources_checked: int

class CoderState(TypedDict):
    code_draft: str
    test_results: List[str]

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    next_agent: str
    researcher: ResearcherState
    coder: CoderState

Each subgraph agent only touches its own nested slice, researcher or coder, while messages and next_agent remain the shared coordination surface the supervisor uses to route. This keeps ownership explicit: if you're debugging why code_draft is empty, you know exactly one node writes to it.

For graphs built from subgraphs (a common pattern when each agent is itself a compiled StateGraph), LangGraph requires at least one shared key between the parent graph's schema and the subgraph's schema so state can flow between them:

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str

class ResearchSubgraphState(TypedDict):
    messages: Annotated[list, add_messages]
    search_queries: List[str]
    findings: List[str]

research_graph = StateGraph(ResearchSubgraphState)
# ... add nodes, compile ...
compiled_research = research_graph.compile()

parent_graph = StateGraph(ParentState)
parent_graph.add_node("research", compiled_research)

The shared messages key is the bridge. The subgraph's extra fields (search_queries, findings) live entirely inside its own execution and never leak into the parent unless you explicitly map them back out in a wrapper node.

Persistence, checkpoints, and why schema shape matters for memory

LangGraph's checkpointing system serializes your entire state object at every super-step so a graph can be paused, resumed, replayed from history, or forked into a new branch of execution (the "time travel" feature). This is where sloppy schema design bites hardest: if your state contains unserializable objects — open file handles, database connections, non-picklable custom classes — checkpointing breaks or silently drops data.

Keep state schemas built from serializable primitives: strings, numbers, lists, dicts, and LangChain's own message types. If a node genuinely needs a live object like a database connection, pass it through config or a Runnable's injected context, not through the state schema.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-42-session-1"}}
app.invoke({"question": "Summarize our last conversation"}, config=config)

Because state is checkpointed per thread_id, a clean schema is also what lets you build durable, resumable multi-turn agents — the messages field with its add_messages reducer keeps accumulating across separate invoke calls on the same thread, giving you persistent conversational memory without hand-rolling a database layer yourself.

Common mistakes and how to avoid them

• Forgetting a reducer on an accumulating field. If a list or counter is supposed to grow across nodes and you didn't annotate it with a reducer, every node's write silently erases the previous one's contribution. This is the single most common LangGraph bug reported by teams new to the framework.
• One giant flat state for a multi-agent graph. Without nested ownership, it becomes impossible to tell which agent is responsible for which field, and parallel branches will race to overwrite shared fields that should have been reducer-managed or namespaced.
• Putting non-serializable objects in state. Open connections, threads, or custom classes without a __reduce__/pickle implementation will break checkpointing the moment you turn on persistence.
• No `input`/`output` schema separation. Exposing your entire internal scratch state to callers locks you into that shape forever, since anyone consuming the graph's output may start depending on fields you meant to be temporary.
• Mutating state in place inside a node. Nodes should return new values to merge, not mutate the state dict argument directly — in-place mutation bypasses reducers entirely and can produce inconsistent behavior depending on how LangGraph's runtime handles the object internally.
• Using `Annotated` reducers on fields that should just overwrite. Not everything needs operator.add. A current_status or next_agent field should replace, not merge — adding a reducer where none is needed usually produces state that grows unbounded or nonsensically combines values that were never meant to be combined.

A complete, runnable example

Putting it together: a small two-agent graph where a router decides between a math tool and a general chat response, using reducers for messages and a plain field for routing decisions.

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

class RouterState(TypedDict):
    messages: Annotated[list, add_messages]
    route: str
    calculation_result: float

def router_node(state: RouterState) -> dict:
    last = state["messages"][-1].content
    if any(op in last for op in ["+", "-", "*", "/"]):
        return {"route": "math"}
    return {"route": "chat"}

def math_node(state: RouterState) -> dict:
    expression = state["messages"][-1].content
    result = eval(expression, {"__builtins__": {}})
    reply = AIMessage(content=f"The answer is {result}")
    return {"messages": [reply], "calculation_result": result}

def chat_node(state: RouterState) -> dict:
    reply = AIMessage(content="I can only do math right now, try an expression.")
    return {"messages": [reply]}

def route_decision(state: RouterState) -> Literal["math", "chat"]:
    return state["route"]

graph = StateGraph(RouterState)
graph.add_node("router", router_node)
graph.add_node("math", math_node)
graph.add_node("chat", chat_node)

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_decision, {"math": "math", "chat": "chat"})
graph.add_edge("math", END)
graph.add_edge("chat", END)

app = graph.compile()

output = app.invoke({
    "messages": [HumanMessage(content="12 * 4")],
    "route": "",
    "calculation_result": 0.0,
})
print(output["messages"][-1].content)

Run this and messages accumulates the human input plus the AI reply via add_messages, while route and calculation_result simply hold their latest values — exactly the mixed reducer/overwrite behavior a real schema needs.

Testing your state schema before you trust it

Before wiring a schema into a large graph, test the merge behavior in isolation. A quick sanity check for a reducer:

def test_add_messages_reducer():
    state = {"messages": [HumanMessage(content="hi")]}
    update = {"messages": [AIMessage(content="hello")]}
    merged = add_messages(state["messages"], update["messages"])
    assert len(merged) == 2

def test_operator_add_counter():
    import operator
    total = operator.add(5, 3)
    assert total == 8

It's a small habit, but it catches the exact class of bug — silent overwrite instead of merge — that otherwise only shows up after a demo in front of a client goes sideways because the agent "forgot" what it said three turns ago.

Where to go from here

State schema design is the part of LangGraph that separates agents that work reliably in production from agents that work great in a five-minute demo and fall apart under real, concurrent, multi-turn use. The core ideas are small — TypedDict, Annotated reducers, input/output separation, nested ownership for multi-agent graphs — but applying them consistently across a real system takes practice, especially once you're juggling subgraphs, checkpointing, and human-in-the-loop interrupts at the same time.

If you want to go from reading about these patterns to actually building a production-grade multi-agent system with proper state design, checkpointing, and tool orchestration, our LangGraph Tutorial course on teachyou.ai walks through all of it hands-on, from a single-node graph to a fully checkpointed multi-agent supervisor architecture.