Back to Details

langgraph-supervisor

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangGraph Supervisor Pattern

LangGraph 管理者模式

Coordinate multiple specialized agents with a central supervisor.
通过中央管理者协调多个专业Agent。

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
  • 构建可将任务分派给工作者的中央协调Agent
  • 实现轮询或基于优先级的任务路由
  • 跟踪Agent任务完成情况与工作流进度
  • 使用Command API实现状态更新与路由的结合操作

Quick Start

快速开始

python
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"]}
python
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

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": []})
undefined
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": []})
undefined

Basic Supervisor

基础管理者模式

python
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
python
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

Build graph

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

Supervisor routes dynamically

Supervisor routes dynamically

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

Workers return to supervisor

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()
undefined
workflow.add_edge("analyzer", "supervisor") workflow.add_edge("validator", "supervisor")
workflow.add_edge(START, "supervisor") # Use START, not set_entry_point() app = workflow.compile()
undefined

Command API (2026 Best Practice)

Command API（2026最佳实践）

Use Command when you need to update state AND route in the same node:
python
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
    )
当你需要在同一个节点中同时更新状态和路由时，使用Command：
python
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

Build graph with Command

workflow = StateGraph(WorkflowState) workflow.add_node("supervisor", supervisor_with_command) workflow.add_node("analyzer", analyzer) workflow.add_node("validator", validator)
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

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)
workflow.add_edge(START, "supervisor") workflow.add_edge("analyzer", "supervisor") workflow.add_edge("validator", "supervisor")
app = workflow.compile()

**何时使用Command vs 条件路由：**
- **Command**：需要同时更新状态和路由时
- **条件路由**：仅需路由（无需更新状态）时

Round-Robin Supervisor

轮询式管理者

python
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
python
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

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")
undefined
for agent_name in ALL_AGENTS: workflow.add_node(agent_name, create_agent_node(agent_name)) workflow.add_edge(agent_name, "supervisor")
undefined

Priority-Based Routing

基于优先级的路由

python
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
python
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)

基于LLM的管理者（2026最佳实践）

python
from pydantic import BaseModel, Field
from typing import Literal
python
from pydantic import BaseModel, Field
from typing import Literal

Define structured output schema

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
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

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
undefined
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
undefined

Tracking Progress

进度跟踪

python
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
python
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

关键决策

DecisionRecommendation
Routing strategyRound-robin for uniform, priority for critical-first
Max agents3-8 specialists (avoid overhead)
Failure handlingSkip failed agent, continue with others
CoordinationCentralized supervisor (simpler debugging)
Command vs ConditionalUse Command when updating state + routing together
Entry pointUse 
add_edge(START, node)
not 
set_entry_point()
决策项推荐方案
路由策略轮询适用于均衡分配任务，优先级调度适用于优先处理关键任务
最大Agent数量3-8个专业Agent（避免额外开销）
故障处理跳过故障Agent，继续执行其他Agent任务
协调方式集中式管理者（便于调试）
Command与条件路由对比当需要同时更新状态和路由时，使用Command
入口点使用
add_edge(START, node)
而非
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
  • 未跟踪任务完成情况（导致Agent无限循环执行）
  • 遗漏工作者到管理者的连接边
  • 缺少结束条件
  • 管理者逻辑过于复杂（应保持轻量）
  • 使用
    set_entry_point()
    （已弃用，应使用
    add_edge(START, ...)
    ）
  • 在Command更适用的场景下使用条件路由

Evaluations

评估

See references/evaluations.md for test cases.
查看references/evaluations.md获取测试用例。

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
  • langgraph-routing
    - 用于动态路由的条件边模式
  • langgraph-parallel
    - 用于并行工作者执行的扇出/扇入模式
  • langgraph-state
    - 带完成跟踪的状态模式
  • langgraph-checkpoints
    - 持久化管理者进度以实现容错
  • langgraph-streaming
    - 工作流执行期间的实时进度更新
  • langgraph-human-in-loop
    - 在管理者决策中添加人工审批环节

Capability Details

能力详情

supervisor-design

supervisor-design

Keywords: supervisor, orchestration, routing, delegation Solves:
  • Design supervisor agent patterns
  • Route tasks to specialized workers
  • Coordinate multi-agent workflows
关键词: supervisor, orchestration, routing, delegation 解决问题:
  • 设计管理者Agent模式
  • 将任务路由至专业工作者
  • 协调多Agent工作流

worker-delegation

worker-delegation

Keywords: worker, delegation, specialized, agent Solves:
  • Create specialized worker agents
  • Define worker capabilities
  • Implement delegation logic
关键词: worker, delegation, specialized, agent 解决问题:
  • 创建专业工作者Agent
  • 定义工作者能力
  • 实现委托逻辑

orchestkit-workflow

orchestkit-workflow

Keywords: orchestkit, analysis, content, workflow Solves:
  • OrchestKit analysis workflow example
  • Production supervisor implementation
  • Real-world orchestration pattern
关键词: orchestkit, analysis, content, workflow 解决问题:
  • OrchestKit分析工作流示例
  • 生产级管理者实现
  • 真实世界编排模式

supervisor-template

supervisor-template

Keywords: template, implementation, code, starter Solves:
  • Supervisor workflow template
  • Production-ready code
  • Copy-paste implementation
关键词: template, implementation, code, starter 解决问题:
  • 管理者工作流模板
  • 生产就绪代码
  • 可直接复制使用的实现

content-analysis

content-analysis

Keywords: content, analysis, graph, multi-agent Solves:
  • Content analysis graph template
  • OrchestKit-specific workflow
  • Multi-agent content processing
关键词: content, analysis, graph, multi-agent 解决问题:
  • 内容分析图模板
  • OrchestKit专属工作流
  • 多Agent内容处理