The Supervisor Pattern: Coordinating Specialist Agents in LangGraph

A single agent trying to do everything becomes a jack of all trades and master of none. The supervisor pattern solves this by placing a coordinator at the top of the hierarchy — a supervisor agent that analyzes incoming tasks, routes them to specialist agents, and aggregates results. It’s how you build agent systems that scale without becoming unmaintainable.

Why the Supervisor Pattern

As agent systems grow, you face a fundamental tension:

• Single agent — Simple but limited. One prompt, one set of tools, one context window.
• Multiple agents — Powerful but chaotic. Without coordination, agents duplicate work, conflict, and waste tokens.

The supervisor pattern resolves this by separating concerns:

1. Supervisor — Understands the task, decides which specialist handles it, and coordinates the workflow
2. Specialists — Deep expertise in specific domains (research, coding, analysis, writing)
3. State management — Shared context that flows between agents

User Request
    ↓
Supervisor (Router)
    ├── Research Specialist
    ├── Code Specialist
    ├── Analysis Specialist
    └── Writing Specialist
    ↓
Aggregated Result

Implementation in LangGraph

Define the Specialists

Each specialist is an independent agent with its own tools and system prompt:

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o")

# Research Specialist
@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    # Implementation here
    return f"Search results for: {query}"

@tool
def read_document(url: str) -> str:
    """Read and extract content from a URL."""
    # Implementation here
    return f"Content from: {url}"

research_agent = create_react_agent(
    llm,
    tools=[web_search, read_document],
    prompt="You are a research specialist. Find accurate, relevant information."
)

# Code Specialist
@tool
def run_code(code: str) -> str:
    """Execute Python code safely."""
    # Implementation here
    return f"Code output for: {code[:50]}..."

@tool
def analyze_code(code: str) -> str:
    """Analyze code for bugs and improvements."""
    # Implementation here
    return f"Analysis of: {code[:50]}..."

code_agent = create_react_agent(
    llm,
    tools=[run_code, analyze_code],
    prompt="You are a code specialist. Write clean, efficient code."
)

# Analysis Specialist
@tool
def数据分析(data: str) -> str:
    """Analyze data and generate insights."""
    # Implementation here
    return f"Analysis of: {data[:50]}..."

analysis_agent = create_react_agent(
    llm,
    tools=[数据分析],
    prompt="You are an analytics specialist. Provide data-driven insights."
)

Build the Supervisor

The supervisor decides which specialist handles each task:

from typing import Literal

class SupervisorState(MessagesState):
    next: str  # Which specialist to route to
    task_type: str  # Classification of the current task

def supervisor_node(state: SupervisorState):
    """Classify the task and route to the appropriate specialist."""
    last_message = state["messages"][-1].content
    
    classification_prompt = f"""Classify this task into exactly one category:
    
Task: {last_message}

Categories:
- research: Finding information, gathering data, fact-checking
- code: Writing, reviewing, debugging, or analyzing code
- analysis: Data analysis, pattern recognition, recommendations
- response: Final synthesis, formatting, or direct answer

Respond with just the category name."""
    
    response = llm.invoke(classification_prompt)
    task_type = response.content.strip().lower()
    
    return {"task_type": task_type, "next": task_type}

def router(state: SupervisorState):
    """Route to the appropriate specialist."""
    return state.get("next", "end")

Assemble the Graph

Connect all components in a LangGraph:

graph = StateGraph(SupervisorState)

# Add nodes
graph.add_node("supervisor", supervisor_node)
graph.add_node("research", research_agent)
graph.add_node("code", code_agent)
graph.add_node("analysis", analysis_agent)

# Add edges
graph.add_edge(START, "supervisor")

# Conditional routing from supervisor
graph.add_conditional_edges(
    "supervisor",
    router,
    {
        "research": "research",
        "code": "code",
        "analysis": "analysis",
        "response": END,
    }
)

# All specialists return to supervisor for next steps
graph.add_edge("research", "supervisor")
graph.add_edge("code", "supervisor")
graph.add_edge("analysis", "supervisor")

# Compile
app = graph.compile()

Running the Supervisor

result = app.invoke({
    "messages": [("user", "Research the latest trends in AI agents, then write a Python script to visualize them")]
})

# The supervisor will:
# 1. Classify as "research" → routes to research specialist
# 2. Research specialist finds trends
# 3. Returns to supervisor → classifies as "code"
# 4. Code specialist writes visualization
# 5. Returns to supervisor → classifies as "response"
# 6. Final answer synthesized

Advanced Patterns

Hierarchical Supervision

For complex systems, nest supervisors:

Main Supervisor
    ├── Research Supervisor
    │   ├── Web Search Agent
    │   ├── Academic Paper Agent
    │   └── News Agent
    ├── Code Supervisor
    │   ├── Python Agent
    │   ├── JavaScript Agent
    │   └── DevOps Agent
    └── Analysis Supervisor
        ├── Data Agent
        ├── Visualization Agent
        └── Report Agent

# Build sub-supervisors
research_supervisor = build_supervisor(
    name="research",
    specialists=[web_agent, academic_agent, news_agent]
)

# Main supervisor includes sub-supervisors as specialists
main_supervisor = build_supervisor(
    name="main",
    specialists=[research_supervisor, code_supervisor, analysis_supervisor]
)

State Accumulation

Specialists contribute to shared state as they work:

from typing import Annotated
from operator import add

class AccumulatorState(MessagesState):
    research_findings: Annotated[list[str], add] = []
    code_outputs: Annotated[list[str], add] = []
    analysis_results: Annotated[list[str], add] = []

def research_node(state: AccumulatorState):
    result = research_agent.invoke(state)
    return {
        "research_findings": [result["messages"][-1].content]
    }

def code_node(state: AccumulatorState):
    # Access accumulated research
    research_context = "\n".join(state.get("research_findings", []))
    
    result = code_agent.invoke({
        "messages": [("user", f"Based on this research, write code:\n{research_context}")]
    })
    
    return {
        "code_outputs": [result["messages"][-1].content]
    }

Confidence-Based Routing

Route based on specialist confidence:

def supervised_routing(state: SupervisorState):
    last_message = state["messages"][-1].content
    
    # Get confidence scores from each specialist
    scores = {}
    for name, agent in specialists.items():
        confidence_prompt = f"""Rate your confidence (0-100) in handling this task:
        
Task: {last_message}

Respond with just a number."""
        response = llm.invoke(confidence_prompt)
        try:
            scores[name] = int(response.content.strip())
        except ValueError:
            scores[name] = 0
    
    # Route to highest confidence specialist
    best_specialist = max(scores, key=scores.get)
    
    if scores[best_specialist] < 50:
        return {"next": "supervisor"}  # Supervisor handles it directly
    
    return {"next": best_specialist}

Error Handling and Fallbacks

Specialist Failure Recovery

def resilient_supervisor(state: SupervisorState):
    try:
        return supervisor_node(state)
    except Exception as e:
        # Log the error
        logger.error(f"Supervisor classification failed: {e}")
        
        # Fallback: route to a general-purpose specialist
        return {"next": "general", "error": str(e)}

Timeout Protection

import asyncio

async def supervisor_with_timeout(state: SupervisorState, timeout: int = 30):
    try:
        result = await asyncio.wait_for(
            supervisor_node(state),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        logger.warning("Supervisor timed out, using default routing")
        return {"next": "general"}

When to Use the Supervisor Pattern

Use the supervisor pattern when:

• Tasks span multiple domains (research + code + analysis)
• You need different tools for different task types
• Specialists have distinct system prompts and behaviors
• You want to track which specialist handled which task
• You need to scale specialists independently

Avoid it when:

• Tasks are simple and don’t need routing
• The overhead of classification isn’t worth it
• A single well-prompted agent can handle everything

Conclusion

The supervisor pattern brings order to multi-agent systems. By separating task classification from execution, you get clean boundaries between specialists, clear routing logic, and the ability to scale each specialist independently. Start with a simple classifier, add specialists as needed, and graduate to hierarchical supervision as your system grows.