langgraph-state-management
LangGraph State Management
State Design Workflow
Follow this workflow when designing or modifying state for a LangGraph application:
- Identify data requirements — What data flows through the graph?
- Choose a schema pattern — Match the use case to a template
- Define reducers — Decide how concurrent updates merge
- Configure persistence — Select and set up a checkpointer
- Validate and test — Run schema validation and reducer tests
Quick Start
Python — Minimal Chat State
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_core.messages import AIMessage
class State(MessagesState):
pass
def chat_node(state: State):
return {"messages": [AIMessage(content="Hello!")]}
graph = StateGraph(State).add_node("chat", chat_node)
graph.add_edge(START, "chat").add_edge("chat", END)
app = graph.compile()
Python — Subclass MessagesState
For convenience, subclass the built-in MessagesState (includes messages with add_messages reducer):
from langgraph.graph import MessagesState
class State(MessagesState):
documents: list[str]
query: str
TypeScript — StateSchema with Zod
import { StateGraph, StateSchema, MessagesValue, ReducedValue, START, END } from "@langchain/langgraph";
import { AIMessage } from "@langchain/core/messages";
import { z } from "zod/v4";
const State = new StateSchema({
messages: MessagesValue,
documents: z.array(z.string()).default(() => []),
count: new ReducedValue(
z.number().default(0),
{ reducer: (current, update) => current + update }
),
});
const graph = new StateGraph(State)
.addNode("chat", (state) => ({ messages: [new AIMessage("Hello!")] }))
.addEdge(START, "chat")
.addEdge("chat", END)
.compile();
Schema Patterns
Choose the pattern matching the application type. See references/schema-patterns.md for complete examples with both Python and TypeScript.
| Pattern | Use Case | Key Fields |
|---|---|---|
| Chat | Conversational agents | Built-in messages from MessagesState |
| Research | Information gathering | query, search_results, summary |
| Workflow | Task orchestration | task, status (Literal), steps_completed |
| Tool-Calling | Agents with tools | messages, tool_calls_made, should_continue |
| RAG | Retrieval-augmented generation | query, retrieved_docs, response |
Template files are available in assets/ for each pattern:
assets/chat_state.py— Chat applicationassets/research_state.py— Research agentassets/workflow_state.py— Workflow orchestrationassets/tool_calling_state.py— Tool-calling agent
For RAG state patterns, use reference examples in references/schema-patterns.md.
Reducers
Reducers control how state updates merge when nodes write to the same field.
Key Concepts
- No reducer → value is overwritten (last-write-wins)
- With reducer → values are merged using the reducer function
- A reducer takes
(existing_value, new_value)and returns the merged result
Python: Annotated Type with Reducer
from typing import Annotated
import operator
from langgraph.graph import MessagesState
class State(MessagesState):
# Overwrite (no reducer)
query: str
# Sum integers
count: Annotated[int, operator.add]
# Custom reducer
results: Annotated[list[str], lambda left, right: left + right]
TypeScript: ReducedValue and MessagesValue
const State = new StateSchema({
query: z.string(), // Last-write-wins
messages: MessagesValue, // Built-in message reducer
count: new ReducedValue( // Custom reducer
z.number().default(0),
{ reducer: (current, update) => current + update }
),
});
Built-in Reducers
| Reducer | Import | Behavior |
|---|---|---|
add_messages |
langgraph.graph.message |
Append, update by ID, delete |
operator.add |
operator |
Numeric addition or list concatenation |
MessagesValue |
@langchain/langgraph |
JS equivalent of add_messages |
Bypass Reducers with Overwrite
Replace accumulated state instead of merging:
from langgraph.types import Overwrite
def reset_messages(state: State):
return {"messages": Overwrite(["fresh start"])}
Delete Messages
from langchain_core.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
# Delete specific message
{"messages": [RemoveMessage(id="msg_123")]}
# Delete all messages
{"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}
For advanced reducer patterns (deduplication, deep merge, conditional update, size-limited accumulators), see references/reducers.md.
Persistence
Persistence enables multi-turn conversations, human-in-the-loop, time travel, and crash recovery.
Choosing a Backend
| Backend | Package | Use Case |
|---|---|---|
| InMemorySaver | langgraph-checkpoint (included) |
Development, testing |
| SqliteSaver | langgraph-checkpoint-sqlite |
Local workflows, single-instance |
| PostgresSaver | langgraph-checkpoint-postgres |
Production, multi-instance |
| CosmosDBSaver | langgraph-checkpoint-cosmosdb |
Azure production |
Agent Server note: When using LangGraph Agent Server, checkpointers are configured automatically — no manual setup needed.
Python Setup
# Development
from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver())
# Production (PostgreSQL)
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:pass@host:5432/db"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# checkpointer.setup() # Run once for initial schema
graph = builder.compile(checkpointer=checkpointer)
result = graph.invoke(
{"messages": [{"role": "user", "content": "Hi"}]},
{"configurable": {"thread_id": "session-1"}}
)
TypeScript Setup
// Development
import { MemorySaver } from "@langchain/langgraph";
const graph = builder.compile({ checkpointer: new MemorySaver() });
// Production (PostgreSQL)
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
const checkpointer = PostgresSaver.fromConnString(DB_URI);
// await checkpointer.setup(); // Run once
const graph = builder.compile({ checkpointer });
Thread Management
Every invocation requires a thread_id to identify the conversation:
config = {"configurable": {"thread_id": "user-123-session-1"}}
result = graph.invoke({"messages": [...]}, config)
Subgraph Persistence
Provide the checkpointer only on the parent graph — LangGraph propagates it to subgraphs automatically:
parent_graph = parent_builder.compile(checkpointer=checkpointer)
# Subgraphs inherit the checkpointer
To give a subgraph its own separate memory:
subgraph = sub_builder.compile(checkpointer=True)
For backend-specific configuration, migration between backends, and TTL settings, see references/persistence-backends.md.
State Typing
Python: TypedDict (Recommended)
from typing import TypedDict, Annotated, Literal
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
next: Literal["agent1", "agent2", "FINISH"]
context: dict
Note:
create_agentstate schemas supportTypedDictfor custom agent state. Prefer TypedDict for agent state extensions.
TypeScript: StateSchema with Zod
import { StateSchema, MessagesValue, ReducedValue, UntrackedValue } from "@langchain/langgraph";
import { z } from "zod/v4";
const AgentState = new StateSchema({
messages: MessagesValue,
currentStep: z.string(),
retryCount: z.number().default(0),
// Custom reducer
allSteps: new ReducedValue(
z.array(z.string()).default(() => []),
{ inputSchema: z.string(), reducer: (current, newStep) => [...current, newStep] }
),
// Transient state (not checkpointed)
tempCache: new UntrackedValue(z.record(z.string(), z.unknown())),
});
// Extract types for use outside the graph builder
type State = typeof AgentState.State;
type Update = typeof AgentState.Update;
For Pydantic validation, advanced type patterns, and migration from untyped state, see references/state-typing.md.
Validation and Debugging
Validate State Schema
Run the validation script to check schema structure:
uv run scripts/validate_state_schema.py my_agent/state.py:MyState --verbose
Checks for: schema parsing issues, empty schemas, reducer annotation problems, message fields without reducers, routing fields without Literal types, and unsupported/unclear schema class patterns.
Test Reducers
Test reducer functions for correctness and edge cases:
uv run scripts/test_reducers.py my_agent/reducers.py:extend_list --verbose
Tests: basic merge, empty inputs, None handling, type consistency, nested structures, large inputs.
Inspect Checkpoints
Debug state evolution by inspecting saved checkpoints:
# List recent checkpoints
uv run scripts/inspect_checkpoints.py ./checkpoints.db
# Inspect specific checkpoint
uv run scripts/inspect_checkpoints.py ./checkpoints.db --checkpoint-id abc123 --thread-id thread-1
# View full history for a thread
uv run scripts/inspect_checkpoints.py ./checkpoints.db --thread-id thread-1 --history
inspect_checkpoints.py accepts either a direct SQLite DB path or a directory containing checkpoints.db.
Migrate Persisted State
When state shape changes require updating persisted checkpoint values:
# Dry run first
uv run scripts/migrate_state.py ./checkpoints.db migrations/add_field.py --dry-run
# Apply migration
uv run scripts/migrate_state.py ./checkpoints.db migrations/add_field.py
Migration script format:
def migrate(old_state: dict) -> dict:
new_state = old_state.copy()
new_state["new_field"] = "default_value" # Add field
new_state.pop("deprecated_field", None) # Remove field
return new_state
Common State Issues
| Symptom | Likely Cause | Fix |
|---|---|---|
| State not updating | Missing reducer | Add Annotated[type, reducer] |
| Messages overwritten | No add_messages reducer |
Use MessagesState (or Annotated[list[BaseMessage], add_messages]) |
| Duplicate entries | Reducer appends without dedup | Use dedup reducer from references/reducers.md |
| State grows unbounded | No cleanup | Use RemoveMessage or trim strategy |
| Agent state schema rejected | Non-TypedDict state_schema in create_agent |
Use a TypedDict agent state schema |
| Parallel update conflict | Multiple Overwrite on same key |
Only one node per super-step can use Overwrite |
For detailed debugging techniques, LangSmith tracing, and checkpoint inspection patterns, see references/state-debugging.md.
Resources
Scripts
| Script | Purpose |
|---|---|
scripts/validate_state_schema.py |
Validate schema structure and typing |
scripts/test_reducers.py |
Test reducer functions |
scripts/inspect_checkpoints.py |
Inspect checkpoint data |
scripts/migrate_state.py |
Migrate checkpoint state values |
References
| File | Content |
|---|---|
| references/schema-patterns.md | Schema examples for chat, research, workflow, RAG, tool-calling |
| references/reducers.md | Reducer patterns, Overwrite, custom reducers, testing |
| references/persistence-backends.md | Backend setup, thread management, migration |
| references/state-typing.md | TypedDict, Pydantic, Zod, validation strategies |
| references/state-debugging.md | Debugging techniques, LangSmith tracing, common issues |
State Templates
| File | Pattern |
|---|---|
assets/chat_state.py |
Chat with MessagesState |
assets/research_state.py |
Research with custom reducers |
assets/workflow_state.py |
Workflow with Literal status |
assets/tool_calling_state.py |
Tool-calling agent with MessagesState |
More from lubu-labs/langchain-agent-skills
langgraph-agent-patterns
Implement multi-agent coordination patterns (supervisor-subagent, router, orchestrator-worker, handoffs) for LangGraph applications. Use when users want to (1) implement multi-agent systems, (2) coordinate multiple specialized agents, (3) choose between coordination patterns, (4) set up supervisor-subagent workflows, (5) implement router-based agent selection, (6) create parallel orchestrator-worker patterns, (7) implement agent handoffs, (8) design state schemas for multi-agent systems, or (9) debug multi-agent coordination issues.
43langgraph-error-handling
Implement LangGraph error handling with current v1 patterns. Use when users need to classify failures, add RetryPolicy for transient issues, build LLM recovery loops with Command routing, add human-in-the-loop with interrupt()/resume, handle ToolNode errors, or choose a safe strategy between retry, recovery, and escalation.
25langgraph-testing-evaluation
Use this skill when you need to test or evaluate LangGraph/LangChain agents: writing unit or integration tests, generating test scaffolds, mocking LLM/tool behavior, running trajectory evaluation (match or LLM-as-judge), running LangSmith dataset evaluations, and comparing two agent versions with A/B-style offline analysis. Use it for Python and JavaScript/TypeScript workflows, evaluator design, experiment setup, regression gates, and debugging flaky/incorrect evaluation results.
24langgraph-project-setup
Initialize and configure LangGraph projects with proper structure, langgraph.json configuration, environment variables, and dependency management. Use when users want to (1) create a new LangGraph project, (2) set up langgraph.json for deployment, (3) configure environment variables for LLM providers, (4) initialize project structure for agents, (5) set up local development with LangGraph Studio, (6) configure dependencies (pyproject.toml, requirements.txt, package.json), or (7) troubleshoot project configuration issues.
21deepagents-setup-configuration
Initialize, validate, and troubleshoot Deep Agents projects in Python or JavaScript using the `deepagents` package. Use when users need to create agents with built-in planning/filesystem/subagents, configure middleware/backends/checkpointing/HITL, migrate from `create_react_agent` or `create_agent`, scaffold projects with repo scripts, validate agent config files, and confirm compatibility with current LangChain/LangGraph/LangSmith docs.
21langsmith-trace-analyzer
Fetch, organize, and analyze LangSmith traces for debugging and evaluation. Use when you need to: query traces/runs by project, metadata, status, or time window; download traces to JSON; organize outcomes into passed/failed/error buckets; analyze token/message/tool-call patterns; compare passed vs failed behavior; or investigate benchmark and production failures.
19