LangGraph State Management

Design and manage state schemas for LangGraph workflows.

TypedDict Approach (Simple)

from typing import TypedDict, Annotated from operator import add

class WorkflowState(TypedDict): input: str output: str agent_responses: Annotated[list[dict], add] # Accumulates metadata: dict

MessagesState Pattern (2026 Best Practice)

from langgraph.graph import MessagesState from langgraph.graph.message import add_messages from typing import Annotated

Option 1: Use built-in MessagesState (recommended)

class AgentState(MessagesState): """Extends MessagesState with custom fields.""" user_id: str context: dict

Option 2: Define messages manually with add_messages reducer

class CustomState(TypedDict): messages: Annotated[list, add_messages] # Smart append/update by ID metadata: dict

Why add_messages matters:

• Appends new messages (doesn't overwrite)

• Updates existing messages by ID

• Handles message deduplication automatically

Note: MessageGraph is deprecated in LangGraph v1.0.0. Use StateGraph with a messages key instead.

Pydantic Approach (Validation)

from pydantic import BaseModel, Field

class WorkflowState(BaseModel): input: str = Field(description="User input") output: str = "" agent_responses: list[dict] = Field(default_factory=list)

def add_response(self, agent: str, result: str):
    self.agent_responses.append({"agent": agent, "result": result})

Accumulating State Pattern

from typing import Annotated from operator import add

class AnalysisState(TypedDict): url: str raw_content: str

# Accumulate agent outputs
findings: Annotated[list[Finding], add]
embeddings: Annotated[list[Embedding], add]

# Control flow
current_agent: str
agents_completed: list[str]
quality_passed: bool

Key Pattern: Annotated[list[T], add]

• Without add : Each node replaces the list

• With add : Each node appends to the list

• Critical for multi-agent workflows

Custom Reducers

from typing import Annotated

def merge_dicts(a: dict, b: dict) -> dict: """Custom reducer that merges dictionaries.""" return {**a, **b}

class State(TypedDict): config: Annotated[dict, merge_dicts] # Merges updates

def last_value(a, b): """Keep only the latest value.""" return b

class State(TypedDict): status: Annotated[str, last_value] # Overwrites

State Immutability

def node(state: WorkflowState) -> WorkflowState: """Return new state, don't mutate in place.""" # Wrong: state["output"] = "result" # Right: return { **state, "output": "result" }

Context Schema (2026 Pattern)

Pass runtime configuration without polluting state:

from dataclasses import dataclass from langgraph.graph import StateGraph

@dataclass class ContextSchema: """Runtime configuration, not persisted in state.""" llm_provider: str = "anthropic" temperature: float = 0.7 max_retries: int = 3 debug_mode: bool = False

Create graph with context schema

graph = StateGraph(WorkflowState, context_schema=ContextSchema)

Access context in nodes

def my_node(state: WorkflowState, context: ContextSchema): if context.llm_provider == "anthropic": response = call_claude(state["input"], context.temperature) else: response = call_openai(state["input"], context.temperature)

if context.debug_mode:
    logger.debug(f"Response: {response}")

return {"output": response}

Invoke with context

graph.invoke( {"input": "Hello"}, context={"llm_provider": "openai", "temperature": 0.5} )

Node Caching (2026 Pattern)

Cache expensive node results with TTL:

from langgraph.cache.memory import InMemoryCache from langgraph.types import CachePolicy

Add node with cache policy

builder.add_node( "embed_content", embed_content_node, cache_policy=CachePolicy(ttl=300) # Cache for 5 minutes )

builder.add_node( "llm_call", llm_node, cache_policy=CachePolicy(ttl=60) # Cache for 1 minute )

Compile with cache

graph = builder.compile(cache=InMemoryCache())

RemainingSteps (Proactive Recursion Handling)

Check remaining steps to wrap up gracefully:

from langgraph.types import RemainingSteps

def agent_node(state: WorkflowState, remaining: RemainingSteps): """Proactively handle recursion limit.""" if remaining.steps < 5: # Running low on steps, wrap up return { "action": "summarize_and_exit", "reason": f"Only {remaining.steps} steps remaining" }

# Continue normal processing
return {"action": "continue"}

Key Decisions

Decision Recommendation

TypedDict vs Pydantic TypedDict for internal state, Pydantic at boundaries

Messages state Use MessagesState or add_messages reducer

Accumulators Always use Annotated[list, add] for multi-agent

Nesting Keep state flat (easier debugging)

Immutability Return new state, don't mutate

Runtime config Use context_schema for non-persistent config

Expensive ops Use CachePolicy to cache node results

Recursion Use RemainingSteps for proactive handling

2026 Guidance: Use TypedDict inside the graph (lightweight, no runtime overhead). Use Pydantic at boundaries (inputs/outputs, user-facing data) for validation.

Common Mistakes

• Forgetting add reducer (overwrites instead of accumulates)

• Mutating state in place (breaks checkpointing)

• Deeply nested state (hard to debug)

• No type hints (lose IDE support)

• Putting runtime config in state (use context_schema instead)

• Not caching expensive operations (repeated embedding calls)

Evaluations

See references/evaluations.md for test cases.

Related Skills

• langgraph-routing

• Using state fields for routing decisions

• langgraph-checkpoints

• Persist state for fault tolerance

• langgraph-parallel

• Accumulating state from parallel nodes

• langgraph-supervisor

• State tracking for agent completion

• langgraph-functional

• State in Functional API patterns

• type-safety-validation

• Pydantic model patterns

Capability Details

state-definition

Keywords: StateGraph, TypedDict, state schema, define state Solves:

• Define workflow state with TypedDict

• Create Pydantic state models

• Structure agent state properly

state-channels

Keywords: channel, Annotated, state channel, MessageChannel Solves:

• Configure state channels for data flow

• Implement message accumulation

• Handle channel-based state updates

state-reducers

Keywords: reducer, add_messages, operator.add, accumulate Solves:

• Implement state reducers with Annotated

• Accumulate messages across nodes

• Handle state merging strategies

subgraphs

Keywords: subgraph, nested graph, parent state, child graph Solves:

• Compose graphs with subgraphs

• Pass state between parent and child

• Implement modular workflow components

state-persistence

Keywords: persist, state persistence, durable state, save state Solves:

• Persist state across executions

• Implement durable workflows

• Handle state serialization