LangGraph State Management — TypedDict and Reducers
You built a LangGraph graph with a few nodes. Each one does its job. But when Node B runs, it has no clue what Node A found. Worse — Node B stomps on Node A’s data, and your chat history is gone.
This happens when you haven’t told LangGraph how to handle your state. And that “how” is the whole game.
In a previous post on graph concepts, we covered nodes, edges, and state at a high level. Here we dig into the engine room — the system that decides whether data gets saved, replaced, or merged as it flows through your graph.
State: The Shared Notebook
State is the shared container that every node reads from and writes to. Picture a whiteboard in a meeting room. Each person walks up, reads what’s there, does their thinking, and writes their findings back.
The key rule: nodes don’t touch state directly. They hand back a dict of changes, and LangGraph folds those changes in.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class MyState(TypedDict):
user_input: str
response: str
step_count: int
def greet(state: MyState) -> dict:
name = state["user_input"]
return {"response": f"Hello, {name}!", "step_count": 1}
graph = StateGraph(MyState)
graph.add_node("greet", greet)
graph.add_edge(START, "greet")
graph.add_edge("greet", END)
app = graph.compile()
result = app.invoke({"user_input": "Alice", "response": "", "step_count": 0})
print(result)
{'user_input': 'Alice', 'response': 'Hello, Alice!', 'step_count': 1}
greet got the full state but only sent back the two keys it cared about. LangGraph merged those in. The user_input field didn’t budge because the node didn’t return it.
Key Insight: From a node’s point of view, state is read-only. You never change it in place — you return a dict of updates and LangGraph does the merge. This makes it easy to trace what each node did, since every change is a clean handoff.
Why TypedDict?
You have three choices for your state schema: plain dicts, TypedDict, or Pydantic models. Why does nearly every LangGraph tutorial pick TypedDict?
Plain dicts have zero safety — misspell a key and nothing warns you. Pydantic models run checks at runtime, which adds overhead you often don’t need. TypedDict sits in the middle: your editor catches typos through type hints, but there’s no speed penalty at runtime.
from typing import TypedDict
from langgraph.graph import StateGraph
class AgentState(TypedDict):
messages: list
current_tool: str
iteration: int
# LangGraph validates keys against this schema
graph = StateGraph(AgentState)
print(f"Graph created with state keys: {list(AgentState.__annotations__.keys())}")
Graph created with state keys: ['messages', 'current_tool', 'iteration']
When you hand this class to StateGraph, LangGraph learns which keys exist. That schema becomes the contract all nodes share.
LangGraph also accepts Pydantic BaseModel and Python dataclass:
from typing import TypedDict
from pydantic import BaseModel
from dataclasses import dataclass
# Option 1: TypedDict (most common -- no runtime overhead)
class StateA(TypedDict):
value: str
# Option 2: Pydantic (adds runtime validation)
class StateB(BaseModel):
value: str
# Option 3: Dataclass (mutable by default)
@dataclass
class StateC:
value: str
print("All three work as state schemas in LangGraph")
All three work as state schemas in LangGraph
Stick with TypedDict unless you truly need Pydantic’s runtime checks. It’s the default in the docs and in practice.
The Default Rule: Last Writer Wins
Here’s what trips up most newcomers. If you don’t set a reducer, LangGraph follows one simple rule: the latest value replaces the old one.
Node A writes count = 5. Node B writes count = 10. End result: count is 10. Node A’s work is gone. Let’s see it happen:
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class CounterState(TypedDict):
count: int
label: str
def node_a(state: CounterState) -> dict:
return {"count": 5, "label": "from A"}
def node_b(state: CounterState) -> dict:
return {"count": 10} # Only updates count, not label
graph = StateGraph(CounterState)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", END)
app = graph.compile()
result = app.invoke({"count": 0, "label": ""})
print(f"count: {result['count']}")
print(f"label: {result['label']}")
count: 10
label: from A
count is 10 — Node B replaced it. But label is still "from A" because Node B didn’t return that key. LangGraph only replaces keys that show up in the returned dict. Everything else stays put.
Warning: If two nodes run at the same time and both return the same key without a reducer, LangGraph throws an InvalidUpdateError. The “last writer wins” rule only works when nodes run one after another. For nodes running side by side, you need a reducer.
Reducers: Merge Instead of Replace
A reducer tells LangGraph how to combine the old value with the new one, rather than tossing the old one out. The idea comes from functional programming: a reducer takes the current value and the incoming update, and returns the merged result.
The syntax uses Annotated to pair a type with a merge function:
from typing import Annotated, TypedDict
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # Concatenate lists
count: int # Overwrite (no reducer)
print("messages uses operator.add reducer")
print("count uses default overwrite")
messages uses operator.add reducer
count uses default overwrite
Annotated[list, operator.add] says: “When a node sends back new items for messages, glue them onto the end of the current list.” Without this, each node would wipe the list clean.
Here’s the proof:
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
import operator
class ListState(TypedDict):
items: Annotated[list, operator.add]
def add_fruits(state: ListState) -> dict:
return {"items": ["apple", "banana"]}
def add_veggies(state: ListState) -> dict:
return {"items": ["carrot", "spinach"]}
graph = StateGraph(ListState)
graph.add_node("fruits", add_fruits)
graph.add_node("veggies", add_veggies)
graph.add_edge(START, "fruits")
graph.add_edge("fruits", "veggies")
graph.add_edge("veggies", END)
app = graph.compile()
result = app.invoke({"items": []})
print(result)
{'items': ['apple', 'banana', 'carrot', 'spinach']}
Without the reducer, only ["carrot", "spinach"] would survive. With operator.add, both lists join into one. That’s the whole point — reducers let you build up data across nodes instead of losing it.
add_messages: Built for Chat
For chatbots and agents, you need a message list that grows as the conversation goes on. LangGraph ships a purpose-built reducer called add_messages for exactly this.
Why not just use operator.add? Because add_messages does something extra — it checks message IDs. If you send back a message with the same ID as one that already exists, it replaces that message instead of creating a copy.
from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage, AIMessage, AnyMessage
from langgraph.graph.message import add_messages
class ChatState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
# Simulating what happens across nodes
existing = [HumanMessage(content="Hello", id="msg-1")]
update = [AIMessage(content="Hi there!", id="msg-2")]
result = add_messages(existing, update)
print(f"Message count: {len(result)}")
for msg in result:
print(f" {msg.__class__.__name__}: {msg.content}")
Message count: 2
HumanMessage: Hello
AIMessage: Hi there!
Where this really helps is when you need to fix a message. Send one with the same ID and the old version gets swapped out:
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph.message import add_messages
existing = [
HumanMessage(content="What's 2+2?", id="msg-1"),
AIMessage(content="It's 5", id="msg-2"),
]
# Same ID as the wrong answer -- triggers replacement
correction = [AIMessage(content="It's 4", id="msg-2")]
result = add_messages(existing, correction)
for msg in result:
print(f" {msg.__class__.__name__} (id={msg.id}): {msg.content}")
HumanMessage (id=msg-1): What's 2+2?
AIMessage (id=msg-2): It's 4
The bad answer got swapped — not stacked on top. With plain operator.add, you’d have both “It’s 5” and “It’s 4” sitting in the list.
Tip: Need to drop a message entirely? Return RemoveMessage(id="msg-2") from langchain_core.messages. Great for trimming chat history when tokens run low.
MessagesState: Less Typing, Same Result
Writing messages: Annotated[list[AnyMessage], add_messages] in every state class gets old fast. LangGraph ships MessagesState — a ready-made class with that field baked in.
from langgraph.graph import MessagesState
# MessagesState already has:
# messages: Annotated[list[AnyMessage], add_messages]
# Extend it with your own fields
class MyAgentState(MessagesState):
current_tool: str
iteration: int
print(f"Inherited keys: {list(MessagesState.__annotations__.keys())}")
Inherited keys: ['messages']
Inherit, add your fields, move on. Here it is in a working graph:
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage, AIMessage
class AgentState(MessagesState):
tool_called: bool
def chatbot(state: AgentState) -> dict:
last_msg = state["messages"][-1]
reply = f"You said: {last_msg.content}"
return {
"messages": [AIMessage(content=reply)],
"tool_called": False,
}
graph = StateGraph(AgentState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
result = app.invoke({
"messages": [HumanMessage(content="How does state work?")],
"tool_called": False,
})
for msg in result["messages"]:
print(f"{msg.__class__.__name__}: {msg.content}")
print(f"Tool called: {result['tool_called']}")
HumanMessage: How does state work?
AIMessage: You said: How does state work?
Tool called: False
I reach for MessagesState in any graph that involves chat. It removes the boilerplate and makes your intent obvious.
Writing Your Own Reducers
When operator.add and add_messages aren’t enough, you write your own. A custom reducer is any function that takes two values — the current one and the incoming one — and returns the merged result.
Here’s a sliding window that keeps only the 3 most recent items:
from typing import Annotated, TypedDict
def keep_last_3(current: list, new: list) -> list:
"""Append new items but keep only the last 3."""
combined = current + new
return combined[-3:]
class BoundedState(TypedDict):
recent_actions: Annotated[list, keep_last_3]
# Simulating sequential updates
step_0 = ["search", "read"]
step_1 = keep_last_3(step_0, ["write"])
print(f"After update 1: {step_1}")
step_2 = keep_last_3(step_1, ["deploy"])
print(f"After update 2: {step_2}")
After update 1: ['search', 'read', 'write']
After update 2: ['read', 'write', 'deploy']
The window slides. Old entries fall off. This pattern saves you from runaway memory in agents that loop many times.
Another handy one — a dict merger that keeps keys from earlier nodes while letting newer values take over:
from typing import Annotated, TypedDict
def merge_dicts(current: dict, new: dict) -> dict:
"""Shallow merge: new values overwrite existing keys."""
return {**current, **new}
class MetadataState(TypedDict):
metadata: Annotated[dict, merge_dicts]
existing = {"source": "api", "confidence": 0.8}
update = {"timestamp": "2026-03-10", "confidence": 0.95}
result = merge_dicts(existing, update)
print(result)
{'source': 'api', 'confidence': 0.95, 'timestamp': '2026-03-10'}
Node A’s source key lived on. The confidence key picked up Node B’s newer number. And timestamp is brand new. With plain overwrite, the whole dict would have been thrown out and replaced.
Key Insight: Keep your reducers pure — no side effects, no API calls, no random values. The function gets the current value and the update, and it returns the merged result. LangGraph calls it behind the scenes every time a node writes to that key.
Tracing What Each Node Did
When your graph spits out something odd, you need to see each node’s contribution step by step. LangGraph’s stream method gives you that view:
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
import operator
class DebugState(TypedDict):
log: Annotated[list[str], operator.add]
value: int
def step_one(state: DebugState) -> dict:
return {"log": ["step_one ran"], "value": 10}
def step_two(state: DebugState) -> dict:
doubled = state["value"] * 2
return {"log": [f"step_two doubled to {doubled}"], "value": doubled}
graph = StateGraph(DebugState)
graph.add_node("step_one", step_one)
graph.add_node("step_two", step_two)
graph.add_edge(START, "step_one")
graph.add_edge("step_one", "step_two")
graph.add_edge("step_two", END)
app = graph.compile()
for event in app.stream({"log": [], "value": 0}):
print(event)
print("---")
{'step_one': {'log': ['step_one ran'], 'value': 10}}
---
{'step_two': {'log': ['step_two doubled to 20'], 'value': 20}}
---
Each event tells you exactly which node wrote which values. step_one set value to 10; step_two doubled it to 20. When a bug shows up, start here.
Tip: For bigger graphs, hook up LangSmith. It captures full state snapshots at every node, draws the path the graph took, and shows timing data. Much more useful than print lines once you have conditional routing in the mix.
The Four Mistakes That Burn People Most
Mistake 1: Changing State in Place
The #1 bug. You reach into the state dict and edit it directly.
# WRONG -- mutating state directly
def bad_node(state):
state["messages"].append("new message") # Direct mutation!
return state
# CORRECT -- return only the updates
def good_node(state):
return {"messages": ["new message"]} # Let the reducer handle it
Direct changes skip the reducer entirely. It may work in toy examples, but it causes silent trouble with checkpoints, parallel nodes, and state replay.
Mistake 2: No Reducer for Side-by-Side Nodes
Two nodes running at the same time both write to the same key? You need a reducer. Without one, LangGraph can’t pick a winner and throws InvalidUpdateError.
from typing import Annotated, TypedDict
import operator
# WRONG -- will crash with InvalidUpdateError in parallel
class BadState(TypedDict):
results: list
# CORRECT -- reducer handles parallel writes
class GoodState(TypedDict):
results: Annotated[list, operator.add]
Mistake 3: Sending Back the Whole State
Nodes should return only what changed. Sending back everything triggers needless overwrites and breaks reducers.
# WRONG -- returning the full state
def bad_node(state):
state_copy = dict(state)
state_copy["status"] = "done"
return state_copy # Overwrites ALL keys, even with reducers
# CORRECT -- return only what changed
def good_node(state):
return {"status": "done"}
Mistake 4: Surprise Behavior with operator.add
operator.add does different things on different types. Make sure the behavior matches what you actually want.
from typing import Annotated, TypedDict
import operator
class ConfusingState(TypedDict):
count: Annotated[int, operator.add] # 5 + 3 = 8 (numeric)
items: Annotated[list, operator.add] # [1] + [2] = [1, 2] (concat)
label: Annotated[str, operator.add] # "hi" + "!" = "hi!" (concat)
print("int: adds numerically")
print("list: concatenates")
print("str: concatenates characters")
int: adds numerically
list: concatenates
str: concatenates characters
That count field with operator.add piles up integers — which might be what you want for a running total, or a bug if you meant to overwrite. Be deliberate.
Warning: Always pass every field when calling app.invoke(). LangGraph won’t set defaults for you — skip a field and you’ll hit a KeyError the moment a node tries to read it.
Full Example: A Research Agent’s State
Let’s bring it all together. Here’s a realistic state schema for a research agent that uses MessagesState for chat, operator.add for source links, and plain overwrite for single-value fields.
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage, AIMessage
import operator
class ResearchAgentState(MessagesState):
"""State for a research agent that searches and summarizes."""
sources: Annotated[list[str], operator.add]
current_query: str
iteration: int
def search(state: ResearchAgentState) -> dict:
query = state["current_query"]
return {
"messages": [AIMessage(content=f"Searching for: {query}")],
"sources": [f"https://example.com/result?q={query}"],
"iteration": state["iteration"] + 1,
}
def summarize(state: ResearchAgentState) -> dict:
num_sources = len(state["sources"])
return {
"messages": [AIMessage(
content=f"Found {num_sources} source(s). Summary complete."
)],
"current_query": "",
}
graph = StateGraph(ResearchAgentState)
graph.add_node("search", search)
graph.add_node("summarize", summarize)
graph.add_edge(START, "search")
graph.add_edge("search", "summarize")
graph.add_edge("summarize", END)
app = graph.compile()
result = app.invoke({
"messages": [HumanMessage(content="Explain LangGraph state")],
"sources": [],
"current_query": "LangGraph state management",
"iteration": 0,
})
print(f"Messages: {len(result['messages'])}")
for msg in result["messages"]:
print(f" {msg.__class__.__name__}: {msg.content}")
print(f"Sources: {result['sources']}")
print(f"Iterations: {result['iteration']}")
Messages: 3
HumanMessage: Explain LangGraph state
AIMessage: Searching for: LangGraph state management
AIMessage: Found 1 source(s). Summary complete.
Sources: ['https://example.com/result?q=LangGraph state management']
Iterations: 1
Each field uses the merge strategy that fits its purpose. Messages grow through add_messages. Sources grow through operator.add. And current_query and iteration overwrite, since only the latest value matters.
When to Keep It Simple
Not every graph needs reducers. For a straight-line pipeline with 2-3 nodes and no shared lists, plain overwrite works fine. Don’t over-build the state for a graph that doesn’t call for it.
Reducers earn their place when you have:
- Side-by-side nodes writing to the same key
- Chat history that must grow, not get wiped
- Multi-step workflows where each node adds to a running list
- Loops where a node runs many times and each pass should append
If your graph is a simple chain with no keys that pile up, skip the reducers. Less moving parts, easier to follow.
Summary
LangGraph state boils down to three ideas:
Schema — usually TypedDict — sets the contract between nodes. Every node knows what fields exist and what types they hold.
Reducers control how updates get merged. No reducer means overwrite. operator.add joins lists (or sums numbers). add_messages grows chat history with built-in dedup. Custom functions let you do anything else.
Partial returns — nodes send back only what they changed, never the full state. LangGraph handles the rest.
For most projects, start with MessagesState, tack on your own fields, and reach for Annotated[list, operator.add] when a field should grow. That covers the vast bulk of real use cases.
Practice Exercise
Build a state schema and three-node graph for a document pipeline. The extract node pulls key phrases. The classify node assigns a category. The enrich node adds more key phrases. Design the state so key phrases build up across nodes, the category overwrites, and a processing log uses a custom reducer that keeps only the last 5 entries.
Click to see the solution
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
import operator
def keep_last_5(current: list, new: list) -> list:
combined = current + new
return combined[-5:]
class DocState(MessagesState):
key_phrases: Annotated[list[str], operator.add]
category: str
processing_log: Annotated[list[str], keep_last_5]
source_text: str
def extract(state: DocState) -> dict:
return {
"key_phrases": ["machine learning", "state management"],
"processing_log": ["extracted key phrases"],
}
def classify(state: DocState) -> dict:
return {
"category": "technical",
"processing_log": ["classified as technical"],
}
def enrich(state: DocState) -> dict:
return {
"key_phrases": ["LangGraph"],
"processing_log": ["enriched with metadata"],
}
graph = StateGraph(DocState)
graph.add_node("extract", extract)
graph.add_node("classify", classify)
graph.add_node("enrich", enrich)
graph.add_edge(START, "extract")
graph.add_edge("extract", "classify")
graph.add_edge("classify", "enrich")
graph.add_edge("enrich", END)
app = graph.compile()
result = app.invoke({
"messages": [],
"key_phrases": [],
"category": "",
"processing_log": [],
"source_text": "LangGraph state management tutorial",
})
print(f"Phrases: {result['key_phrases']}")
print(f"Category: {result['category']}")
print(f"Log: {result['processing_log']}")
`key_phrases` uses `operator.add` to gather entries from `extract` and `enrich`. `category` overwrites since only the final label matters. `processing_log` uses `keep_last_5` to cap growth.
FAQ
Can I use Pydantic instead of TypedDict for state?
Yes. LangGraph takes BaseModel and dataclasses alongside TypedDict. Pydantic adds runtime checks — hand back a wrong type and you get an error right away. The tradeoff is a small speed hit.
What if I skip a field when calling invoke()?
You’ll get a KeyError the first time a node tries to read that field. Always pass every field in your starting state.
Can reducers look at other fields in the state?
No. A reducer only sees the current and incoming values for its own field. If you need cross-field logic, put that in a node function.
How do I shrink a list that uses operator.add?
You can’t — operator.add only grows the list. Write a custom reducer that prunes, or use add_messages with RemoveMessage for chat messages.
Does state carry over between invoke() calls?
Not by default. Each invoke() starts fresh. To keep state across calls, you need a checkpointer — covered in a later post on persistence.
References
- LangGraph State Concepts — Official Documentation
- LangGraph Python API — StateGraph
- LangGraph add_messages Reducer
- Python typing.TypedDict — Official Docs
- Python typing.Annotated — Official Docs
- LangChain Messages API
- LangGraph State Channels
Build a strong Python foundation with hands-on exercises designed for aspiring Data Scientists and AI/ML Engineers.
Start Free Course →
