Advanced State Management
Master state schemas with TypedDict, state reducers (add, replace, merge), and custom reducer functions for sophisticated state logic.
Advanced State Management
State management is the heart of LangGraph. This lesson covers advanced state schemas, built-in reducers, and custom reducer functions that give you fine-grained control over how state updates are applied.
State Schema Approaches
LangGraph supports three schema definition approaches. Choose based on your needs:
from typing_extensions import TypedDict
from dataclasses import dataclass, field
from pydantic import BaseModel, Field
from typing import List, Optional
# 1. TypedDict — lightweight, no validation
class AgentState(TypedDict):
messages: List[str]
turn_count: int
# 2. dataclass — mutable, default values
@dataclass
class AgentState:
messages: List[str] = field(default_factory=list)
turn_count: int = 0
# 3. BaseModel — validation, serialization
class AgentState(BaseModel):
messages: List[str] = Field(default_factory=list)
turn_count: int = Field(default=0)| Feature | TypedDict | dataclass | BaseModel |
|---|---|---|---|
| Type hints only | Yes | Yes | Yes |
| Default values | No | Yes | Yes |
| Runtime validation | No | No | Yes |
| Serialization | Manual | Manual | Built-in |
| Performance | Fastest | Fast | Slightly slower |
| IDE support | Good | Excellent | Excellent |
Use TypedDict for simple agents, dataclass for agents needing defaults, and BaseModel for production systems requiring validation and serialization.
State Reducers
Reducers control how multiple writes to the same state key are combined. Without a reducer, the last write wins.
The Default: Replace
class DefaultState(TypedDict):
items: List[str]
def node_a(state: DefaultState) -> dict:
return {"items": ["a"]} # Replaces items
def node_b(state: DefaultState) -> dict:
return {"items": ["b"]} # Replaces items again — "a" is lost
# Result: items = ["b"]The Add Reducer
Annotated[type, add] uses operator.add to combine values:
from typing import Annotated
from operator import add
class AppendState(TypedDict):
items: Annotated[List[str], add]
def node_a(state: AppendState) -> dict:
return {"items": ["a"]} # Appended
def node_b(state: AppendState) -> dict:
return {"items": ["b"]} # Appended
# Result: items = ["a", "b"]The add reducer requires both values to be of the same type that supports +. For lists, both must be lists; for ints, both must be ints.
Built-in Reducers
operator.add
Works on lists (concatenation), numbers (addition), and strings:
from operator import add
class CounterState(TypedDict):
count: Annotated[int, add]
logs: Annotated[List[str], add]
text: Annotated[str, add]
def increment(state: CounterState) -> dict:
return {"count": 1} # Adds 1 to current count
def add_log(state: CounterState) -> dict:
return {"logs": ["Processing step"]} # Appends to list
def add_text(state: CounterState) -> dict:
return {"text": " more"} # Concatenates stringsreplace (default)
The default behavior — no annotation needed:
class NoReducerState(TypedDict):
value: str # Last write wins by default
def first(state: NoReducerState) -> dict:
return {"value": "first"}
def second(state: NoReducerState) -> dict:
return {"value": "second"} # Overwrites
# Result: value = "second"Custom Reducers
For complex merge logic, define a custom function:
from typing import Annotated
def merge_counts(old: dict, new: dict) -> dict:
"""Deep merge two count dictionaries."""
result = old.copy()
for key, value in new.items():
if key in result:
result[key] = result[key] + value
else:
result[key] = value
return result
class CustomState(TypedDict):
counts: Annotated[dict, merge_counts]
def node_a(state: CustomState) -> dict:
return {"counts": {"apples": 5, "oranges": 3}}
def node_b(state: CustomState) -> dict:
return {"counts": {"apples": 2, "bananas": 4}}
# Result: counts = {"apples": 7, "oranges": 3, "bananas": 4}Custom Reducer Signature
A reducer function receives (current_value, new_update) and returns the merged value:
def my_reducer(current: ValueType, update: ValueType) -> ValueType:
# Combine current and update
return merged_valueThe reducer receives the existing state value and the new update from the node. It must return the new value for that key. Reducers run every time a node writes to that key.
Advanced Reducer Patterns
Max Reducer
def max_reducer(current: int, update: int) -> int:
return max(current, update)
class MaxState(TypedDict):
highest_score: Annotated[int, max_reducer]
def player_1(state: MaxState) -> dict:
return {"highest_score": 85}
def player_2(state: MaxState) -> dict:
return {"highest_score": 92}
# Result: highest_score = 92 (max of all writes)List Dedup Reducer
def dedup_append(current: List[str], update: List[str]) -> List[str]:
seen = set(current)
result = current[:]
for item in update:
if item not in seen:
seen.add(item)
result.append(item)
return result
class DedupState(TypedDict):
unique_items: Annotated[List[str], dedup_append]
def add_items(state: DedupState) -> dict:
return {"unique_items": ["a", "b", "a"]}
# Result: unique_items = ["a", "b"] (duplicate "a" removed)Timestamp Merger
def latest_wins(current: dict, update: dict) -> dict:
current.update(update)
return current
class MetadataState(TypedDict):
metadata: Annotated[dict, latest_wins]
def node_a(state: MetadataState) -> dict:
return {"metadata": {"status": "processing", "started_at": "2024-01-01"}}
def node_b(state: MetadataState) -> dict:
return {"metadata": {"status": "completed", "completed_at": "2024-01-02"}}
# Result: metadata = {status: "completed", started_at: "2024-01-01", completed_at: "2024-01-02"}Reducer Composition
Combine multiple reducers across different fields:
from typing import Annotated
from operator import add
def merge_dicts(a: dict, b: dict) -> dict:
result = a.copy()
result.update(b)
return result
class CompositeState(TypedDict):
messages: Annotated[List[str], add] # Append
score: Annotated[int, add] # Sum
config: Annotated[dict, merge_dicts] # Shallow merge
status: str # Replace (default)
max_val: Annotated[float, max_reducer] # Custom: maxState Validation with Pydantic
Add runtime validation to state fields:
from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
class ValidatedState(BaseModel):
messages: List[str] = Field(default_factory=list)
temperature: float = Field(default=0.7, ge=0.0, le=2.0)
max_tokens: int = Field(default=1024, gt=0, le=8192)
@field_validator("messages")
@classmethod
def check_message_limit(cls, v: List[str]) -> List[str]:
if len(v) > 100:
raise ValueError("Too many messages (max 100)")
return vPydantic validation runs on every state update. For high-throughput graphs, this adds overhead. Use it judiciously in production.
State Persistence Across Runs
Without persistence, state is ephemeral. To persist state:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = builder.compile(checkpointer=checkpointer)
# State is saved after each node execution
config = {"configurable": {"thread_id": "user-session-1"}}
result = app.invoke(initial_state, config)
# Retrieve state for a thread
saved_state = app.get_state(config)
print(saved_state.values)Persistence is covered in depth in Lesson 2. It's mentioned here because it interacts with state reducers — persisted state is restored and reducers continue from where they left off.
Practice Questions
Which state definition approach provides runtime validation?
What is the default state update behavior without a reducer?
What typing construct applies a reducer to a state field?
What is the signature of a custom reducer function?
Which operator does the 'add' reducer use?
What does a custom reducer returning max(current, update) accomplish?
When using add reducer on a list, what type must node returns be?
How do you apply different reducers to different fields in the same state?
Which state approach is recommended for production systems needing serialization?
What happens when Pydantic validation fails on a state update?
Key Takeaways
- Three state approaches: TypedDict (lightweight), dataclass (defaults), BaseModel (validation)
- Default reducer is "last write wins" (replace)
Annotated[type, add]appends with operator.add- Custom reducers:
def reducer(current, update) -> new_value - Different fields can have different reducers via Annotated
- Pydantic BaseModel adds runtime validation at the cost of performance
- Custom reducers enable patterns like max, dedup, and deep merge