langgraph-state（语言状态）

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

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