State Management Patterns for Multi-Agent Systems

State Management Patterns for Multi-Agent Systems

State schemas must support the coordination requirements of each multi-agent pattern. This reference covers state design for supervisor, router, orchestrator-worker, and handoff patterns.

Supervisor Pattern State

Basic Supervisor State

from typing import TypedDict, Literal, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class SupervisorState(TypedDict):
    """State for supervisor pattern."""
    # Message history with reducer
    messages: Annotated[list[BaseMessage], add_messages]

    # Routing decision
    next: Literal["agent1", "agent2", "agent3", "FINISH"]

    # Track current agent
    current_agent: str

    # Completion flag
    task_complete: bool

TypeScript:

import { Annotation, messagesStateReducer } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";

const SupervisorState = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: messagesStateReducer,
    default: () => [],
  }),
  next: Annotation<"agent1" | "agent2" | "agent3" | "FINISH">(),
  currentAgent: Annotation<string>(),
  taskComplete: Annotation<boolean>(),
});

Enhanced Supervisor State

Add metadata and context tracking:

class EnhancedSupervisorState(TypedDict):
    """Supervisor state with enhanced tracking."""
    messages: Annotated[list[BaseMessage], add_messages]
    next: Literal["researcher", "writer", "reviewer", "FINISH"]
    current_agent: str

    # Agent history
    agent_history: list[str]

    # Shared context
    context: dict

    # Metadata
    start_time: float
    iteration_count: int
    cost_estimate: float

Router Pattern State

Basic Router State

class RouterState(TypedDict):
    """State for router pattern."""
    # Input message
    messages: list[BaseMessage]

    # One-time routing decision
    route: Literal["sales", "support", "billing"]

    # Optional: routing confidence
    confidence: float

Multi-Label Router State

Support routing to multiple agents:

class MultiLabelRouterState(TypedDict):
    """Router state for multi-label routing."""
    messages: list[BaseMessage]

    # Multiple routes possible
    routes: list[str]

    # Confidence per route
    route_confidence: dict[str, float]

    # Primary route
    primary_route: str

Orchestrator-Worker Pattern State

Orchestrator State

class OrchestratorState(TypedDict):
    """State for orchestrator-worker pattern."""
    # Original task
    task: str

    # Decomposed subtasks
    subtasks: list[dict]

    # Worker results (accumulated)
    results: Annotated[list[str], lambda x, y: x + y]

    # Aggregated final result
    final_result: str

    # Progress tracking
    completed_count: int
    total_count: int

Worker State

class WorkerState(TypedDict):
    """State for individual worker."""
    # Assigned subtask
    subtask: dict

    # Worker result
    result: str

    # Status tracking
    status: Literal["pending", "processing", "complete", "failed"]
    error: str | None

TypeScript:

const OrchestratorState = Annotation.Root({
  task: Annotation<string>(),
  subtasks: Annotation<Array<any>>(),
  results: Annotation<string[]>({
    reducer: (left, right) => left.concat(right),
  }),
  finalResult: Annotation<string>(),
  completedCount: Annotation<number>(),
  totalCount: Annotation<number>(),
});

Handoff Pattern State

Basic Handoff State

class HandoffState(TypedDict):
    """State for handoff pattern."""
    # Message history
    messages: Annotated[list[BaseMessage], add_messages]

    # Current agent
    current_agent: str

    # Next agent to hand off to
    next_agent: Literal["researcher", "writer", "editor", "FINISH"]

    # Shared context across handoffs
    context: dict

Structured Handoff State

Use structured context:

from pydantic import BaseModel

class HandoffContext(BaseModel):
    """Structured context for handoffs."""
    research_data: list[str] = []
    draft_versions: list[str] = []
    review_notes: list[str] = []
    metadata: dict = {}

class StructuredHandoffState(TypedDict):
    """Handoff state with structured context."""
    messages: Annotated[list[BaseMessage], add_messages]
    current_agent: str
    next_agent: str

    # Structured context
    context: HandoffContext

    # Handoff history
    handoff_history: list[str]

Common State Patterns

1. Message Accumulation

Use add_messages reducer for chat history:

from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    """State with message accumulation."""
    messages: Annotated[list[BaseMessage], add_messages]

The add_messages reducer:

• Appends new messages to existing list
• Handles message updates by ID
• Supports RemoveMessage-based deletions when you need to prune history

2. List Accumulation

Accumulate results from multiple agents:

def extend_list(left: list, right: list) -> list:
    """Reducer that extends lists."""
    return left + right

class AccumulatorState(TypedDict):
    """State with list accumulation."""
    results: Annotated[list[str], extend_list]

3. Dictionary Merging

Merge context dictionaries:

def merge_dicts(left: dict, right: dict) -> dict:
    """Reducer that merges dicts."""
    return {**left, **right}

class MergeState(TypedDict):
    """State with dict merging."""
    context: Annotated[dict, merge_dicts]

4. Conditional Updates

Only update if value is present:

def conditional_update(left: str, right: str | None) -> str:
    """Update only if right is not None."""
    return right if right is not None else left

class ConditionalState(TypedDict):
    """State with conditional updates."""
    status: Annotated[str, conditional_update]

State Validation

Pydantic Validation

Use Pydantic for state validation:

from pydantic import BaseModel, validator

class ValidatedState(BaseModel):
    """State with validation."""
    messages: list[BaseMessage]
    current_agent: str
    iteration_count: int

    @validator("iteration_count")
    def check_iterations(cls, v):
        """Prevent infinite loops."""
        if v > 10:
            raise ValueError("Too many iterations")
        return v

    @validator("current_agent")
    def check_agent(cls, v):
        """Validate agent name."""
        valid_agents = {"researcher", "writer", "editor"}
        if v not in valid_agents:
            raise ValueError(f"Invalid agent: {v}")
        return v

Note: For StateGraph schemas, Pydantic validation is applied to the initial input state only. The graph output is not returned as a Pydantic instance, and subsequent node updates are not re-validated.

Custom Validation

Add validation in nodes:

def validate_state(state: SupervisorState) -> None:
    """Validate state before processing."""
    if not state.get("messages"):
        raise ValueError("Messages cannot be empty")

    if state.get("iteration_count", 0) > 20:
        raise ValueError("Maximum iterations exceeded")

def supervisor_node(state: SupervisorState) -> dict:
    """Supervisor with state validation."""
    validate_state(state)

    # Continue processing...

State Size Management

Context Summarization

Prevent state from growing too large:

def summarize_messages(messages: list[BaseMessage], max_length: int = 10) -> list[BaseMessage]:
    """Summarize messages if too long."""
    if len(messages) <= max_length:
        return messages

    # Keep first and last messages, summarize middle
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import SystemMessage

    model = ChatOpenAI(model="gpt-4o-mini")
    summary_prompt = f"Summarize these messages: {messages[1:-5]}"
    summary = model.invoke(summary_prompt).content

    return [
        messages[0],
        SystemMessage(content=f"Summary: {summary}"),
        *messages[-5:]
    ]

def supervisor_with_summarization(state: SupervisorState) -> dict:
    """Supervisor that manages state size."""
    messages = state["messages"]

    if len(messages) > 15:
        messages = summarize_messages(messages)

    # Continue with summarized messages...

Selective State Updates

Only update necessary fields:

def minimal_update_agent(state: SupervisorState) -> dict:
    """Agent that updates only necessary fields."""
    # Don't re-send entire state
    return {
        "next": "next_agent",  # Only routing decision
        # No need to resend messages, context, etc.
    }

Pattern-Specific Considerations

Supervisor Pattern

Key requirements:

• Message accumulation with add_messages
• Routing decision field (next)
• Agent tracking (current_agent)
• Loop prevention (iteration counter)

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next: str
    current_agent: str
    iteration_count: int  # For loop detection

Router Pattern

Key requirements:

• Input messages
• One-time routing decision
• Optional confidence scores

class RouterState(TypedDict):
    messages: list[BaseMessage]
    route: str
    confidence: float  # Optional

Orchestrator-Worker

Key requirements:

• Task and subtasks
• Result accumulation
• Progress tracking

class OrchestratorState(TypedDict):
    task: str
    subtasks: list[dict]
    results: Annotated[list[str], extend_list]
    completed: int
    total: int

Handoff Pattern

Key requirements:

• Message accumulation
• Handoff target (next_agent)
• Shared context
• Handoff history (for loop prevention)

class HandoffState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_agent: str
    context: dict
    handoff_history: list[str]

TypeScript State Patterns

Annotation API

import { Annotation, messagesStateReducer } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";

// Message accumulation
const StateWithMessages = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: messagesStateReducer,
    default: () => [],
  }),
});

// Dictionary merging
const StateWithContext = Annotation.Root({
  context: Annotation<Record<string, any>>({
    reducer: (left, right) => ({ ...left, ...right }),
    default: () => ({}),
  }),
});

// Simple replacement
const StateWithRoute = Annotation.Root({
  route: Annotation<string>({
    reducer: (_, right) => right,
  }),
});

Testing State Management

Test Reducers

def test_message_reducer():
    """Test message accumulation."""
    from langgraph.graph.message import add_messages
    from langchain_core.messages import HumanMessage

    existing = [HumanMessage(content="Hello")]
    new_msgs = [HumanMessage(content="World")]

    result = add_messages(existing, new_msgs)

    assert len(result) == 2
    assert result[1].content == "World"

def test_custom_reducer():
    """Test custom reducer."""
    def extend_list(left, right):
        return left + right

    existing = ["a", "b"]
    new = ["c", "d"]

    result = extend_list(existing, new)

    assert result == ["a", "b", "c", "d"]

Test State Updates

def test_state_updates():
    """Test state updates work correctly."""
    initial_state = {
        "messages": [],
        "next": "",
        "current_agent": ""
    }

    update = supervisor_node(initial_state)

    assert "next" in update
    assert update["next"] in ["agent1", "agent2", "FINISH"]

Additional Resources

• LangGraph Graph API (Python): https://docs.langchain.com/oss/python/langgraph/graph-api
• LangGraph Graph API (JavaScript): https://docs.langchain.com/oss/javascript/langgraph/graph-api
• Reducers in LangGraph: https://docs.langchain.com/oss/python/langgraph/use-graph-api