langgraph-supervisor（语言图主管）

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

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