LangGraph Supervisor Pattern

Coordinate multiple specialized agents with a central supervisor.

Overview

Building central coordinator agents that dispatch to workers
Implementing round-robin or priority-based task routing
Tracking agent completion and workflow progress
Using Command API for combined state update + routing

Quick Start

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import Literal, TypedDict

class WorkflowState(TypedDict):
    input: str
    results: list[str]
    agents_completed: list[str]

def supervisor(state) -> Command[Literal["worker_a", "worker_b", END]]:
    if "worker_a" not in state["agents_completed"]:
        return Command(goto="worker_a")
    elif "worker_b" not in state["agents_completed"]:
        return Command(goto="worker_b")
    return Command(goto=END)

def worker_a(state):
    return {"results": ["A done"], "agents_completed": ["worker_a"]}

def worker_b(state):
    return {"results": ["B done"], "agents_completed": ["worker_b"]}

# Build graph
graph = StateGraph(WorkflowState)
graph.add_node("supervisor", supervisor)
graph.add_node("worker_a", worker_a)
graph.add_node("worker_b", worker_b)
graph.add_edge(START, "supervisor")
graph.add_edge("worker_a", "supervisor")
graph.add_edge("worker_b", "supervisor")

app = graph.compile()
result = app.invoke({"input": "task", "results": [], "agents_completed": []})

Basic Supervisor

from langgraph.graph import StateGraph, START, END

def supervisor(state: WorkflowState) -> WorkflowState:
    """Route to next worker based on state."""
    if state["needs_analysis"]:
        state["next"] = "analyzer"
    elif state["needs_validation"]:
        state["next"] = "validator"
    else:
        state["next"] = END
    return state

def analyzer(state: WorkflowState) -> WorkflowState:
    """Specialized analysis worker."""
    result = analyze(state["input"])
    state["results"].append(result)
    return state

# Build graph
workflow = StateGraph(WorkflowState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("analyzer", analyzer)
workflow.add_node("validator", validator)

# Supervisor routes dynamically
workflow.add_conditional_edges(
    "supervisor",
    lambda s: s["next"],
    {
        "analyzer": "analyzer",
        "validator": "validator",
        END: END
    }
)

# Workers return to supervisor
workflow.add_edge("analyzer", "supervisor")
workflow.add_edge("validator", "supervisor")

workflow.add_edge(START, "supervisor")  # Use START, not set_entry_point()
app = workflow.compile()

Command API (2026 Best Practice)

Use Command when you need to update state AND route in the same node:

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import Literal

def supervisor_with_command(state: WorkflowState) -> Command[Literal["analyzer", "validator", END]]:
    """Use Command for combined state update + routing."""
    if state["needs_analysis"]:
        return Command(
            update={"current_agent": "analyzer", "routing_reason": "needs analysis"},
            goto="analyzer"
        )
    elif state["needs_validation"]:
        return Command(
            update={"current_agent": "validator", "routing_reason": "needs validation"},
            goto="validator"
        )
    return Command(
        update={"status": "complete"},
        goto=END
    )

# Build graph with Command
workflow = StateGraph(WorkflowState)
workflow.add_node("supervisor", supervisor_with_command)
workflow.add_node("analyzer", analyzer)
workflow.add_node("validator", validator)

# No conditional edges needed - Command handles routing
workflow.add_edge(START, "supervisor")
workflow.add_edge("analyzer", "supervisor")
workflow.add_edge("validator", "supervisor")

app = workflow.compile()

When to use Command vs Conditional Edges:

Command: When updating state AND routing together
Conditional edges: When routing only (no state updates needed)

Round-Robin Supervisor

ALL_AGENTS = ["security", "tech", "implementation", "tutorial"]

def supervisor_node(state: AnalysisState) -> AnalysisState:
    """Route to next available agent."""
    completed = set(state["agents_completed"])
    available = [a for a in ALL_AGENTS if a not in completed]

    if not available:
        state["next"] = "quality_gate"
    else:
        state["next"] = available[0]

    return state

# Register all agent nodes
for agent_name in ALL_AGENTS:
    workflow.add_node(agent_name, create_agent_node(agent_name))
    workflow.add_edge(agent_name, "supervisor")

Priority-Based Routing

AGENT_PRIORITIES = {
    "security": 1,    # Run first
    "tech": 2,
    "implementation": 3,
    "tutorial": 4     # Run last
}

def priority_supervisor(state: WorkflowState) -> WorkflowState:
    """Route by priority, not round-robin."""
    completed = set(state["agents_completed"])
    available = [a for a in AGENT_PRIORITIES if a not in completed]

    if not available:
        state["next"] = "finalize"
    else:
        # Sort by priority
        next_agent = min(available, key=lambda a: AGENT_PRIORITIES[a])
        state["next"] = next_agent

    return state

LLM-Based Supervisor (2026 Best Practice)

from pydantic import BaseModel, Field
from typing import Literal

# Define structured output schema
class SupervisorDecision(BaseModel):
    """Validated supervisor routing decision."""
    next_agent: Literal["security", "tech", "implementation", "tutorial", "DONE"]
    reasoning: str = Field(description="Brief explanation for routing decision")

async def llm_supervisor(state: WorkflowState) -> WorkflowState:
    """Use LLM with structured output for reliable routing."""
    available = [a for a in AGENTS if a not in state["agents_completed"]]

    # Use structured output (2026 best practice)
    decision = await llm.with_structured_output(SupervisorDecision).ainvoke(
        f"""Task: {state['input']}

Completed: {state['agents_completed']}
Available: {available}

Select the next agent or 'DONE' if all work is complete."""
    )

    # Validated response - no string parsing needed
    state["next"] = END if decision.next_agent == "DONE" else decision.next_agent
    state["routing_reasoning"] = decision.reasoning  # Track decision rationale
    return state

# Alternative: OpenAI structured output
async def llm_supervisor_openai(state: WorkflowState) -> WorkflowState:
    """OpenAI with strict structured output."""
    response = await client.beta.chat.completions.parse(
        model="gpt-5.2",
        messages=[{"role": "user", "content": prompt}],
        response_format=SupervisorDecision
    )
    decision = response.choices[0].message.parsed
    state["next"] = END if decision.next_agent == "DONE" else decision.next_agent
    return state

Tracking Progress

def agent_node_factory(agent_name: str):
    """Create agent node that tracks completion."""
    async def node(state: WorkflowState) -> WorkflowState:
        result = await agents[agent_name].run(state["input"])

        return {
            **state,
            "results": state["results"] + [result],
            "agents_completed": state["agents_completed"] + [agent_name],
            "current_agent": None
        }
    return node

Key Decisions

Decision	Recommendation
Routing strategy	Round-robin for uniform, priority for critical-first
Max agents	3-8 specialists (avoid overhead)
Failure handling	Skip failed agent, continue with others
Coordination	Centralized supervisor (simpler debugging)
Command vs Conditional	Use Command when updating state + routing together
Entry point	Use `add_edge(START, node)` not `set_entry_point()`

Common Mistakes

No completion tracking (runs agents forever)
Forgetting worker → supervisor edge
Missing END condition
Heavy supervisor logic (should be lightweight)
Using set_entry_point() (deprecated, use add_edge(START, ...))
Using conditional edges when Command would be cleaner

Evaluations

See references/evaluations.md for test cases.

Related Skills

langgraph-routing - Conditional edge patterns for dynamic routing
langgraph-parallel - Fan-out/fan-in for parallel worker execution
langgraph-state - State schemas with completion tracking
langgraph-checkpoints - Persist supervisor progress for fault tolerance
langgraph-streaming - Real-time progress updates during workflow
langgraph-human-in-loop - Add human approval gates to supervisor decisions

Capability Details

supervisor-design

Keywords: supervisor, orchestration, routing, delegation Solves:

Design supervisor agent patterns
Route tasks to specialized workers
Coordinate multi-agent workflows

worker-delegation

Keywords: worker, delegation, specialized, agent Solves:

Create specialized worker agents
Define worker capabilities
Implement delegation logic

orchestkit-workflow

Keywords: orchestkit, analysis, content, workflow Solves:

OrchestKit analysis workflow example
Production supervisor implementation
Real-world orchestration pattern

supervisor-template

Keywords: template, implementation, code, starter Solves:

Supervisor workflow template
Production-ready code
Copy-paste implementation

content-analysis

Keywords: content, analysis, graph, multi-agent Solves:

Content analysis graph template
OrchestKit-specific workflow
Multi-agent content processing

langgraph-supervisor