multi-agent-supervisor-pattern

Multi-Agent Supervisor Pattern

The supervisor pattern uses a central LLM-based router to coordinate multiple specialized agents, delegating tasks based on requirements.

Architecture

User Input -> Supervisor -> Route to Agent -> Agent Executes -> Back to Supervisor -> Repeat or Finish

Reference: Multi-Agent Supervisor Tutorial

See also: For designing supervisor state schemas and reducer functions, see the state-management skill.

Implementation Pattern

import logging
from typing import TypedDict, Annotated, List, Literal
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# Initialize logging
logger = logging.getLogger(__name__)

# Initialize LLM (choose one based on your provider)
from langchain_anthropic import ChatAnthropic
# from langchain_openai import ChatOpenAI

llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")
# llm = ChatOpenAI(model="gpt-4o")

# For specialized agents (can use same or different models)
research_llm = llm
code_llm = llm

# Supervisor system prompt
SUPERVISOR_PROMPT = """You are a supervisor coordinating specialized agents.
Available agents:
- research: Find information and answer questions
- code: Write and review code
- FINISH: Task is complete

Analyze the conversation and decide which agent to use next."""

# Type alias for routing destinations
AgentName = Literal["research", "code", "FINISH"]

class RouterDecision(BaseModel):
    """Supervisor's routing decision."""
    next_agent: str = Field(description="Name of next agent or FINISH")

class SupervisorState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]  # Preferred reducer
    next_agent: str

# Supervisor decides routing using SystemMessage with error handling
def supervisor_node(state: SupervisorState) -> dict:
    """Supervisor makes routing decisions with error handling."""
    try:
        messages = [
            SystemMessage(content=SUPERVISOR_PROMPT),
            *state["messages"]
        ]
        router_llm = llm.with_structured_output(RouterDecision)
        decision = router_llm.invoke(messages)
        return {"next_agent": decision.next_agent}
    except Exception as e:
        logger.error(f"Supervisor error: {e}")
        return {"next_agent": "FINISH"}  # Safe fallback

# Specialized agents
def research_agent(state: SupervisorState) -> dict:
    """Research-specific logic."""
    result = research_llm.invoke(state["messages"])
    return {"messages": [result]}

def code_agent(state: SupervisorState) -> dict:
    """Code-specific logic."""
    result = code_llm.invoke(state["messages"])
    return {"messages": [result]}

# Routing function with validation for unexpected LLM outputs
def route_to_agent(state: SupervisorState) -> AgentName:
    """Route to next agent, validating the LLM's routing decision."""
    next_agent = state["next_agent"]
    if next_agent not in ("research", "code", "FINISH"):
        logger.warning(f"Unexpected agent '{next_agent}', defaulting to FINISH")
        return "FINISH"
    return next_agent  # type: ignore - validated above

# Build graph
workflow = StateGraph(SupervisorState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("research", research_agent)
workflow.add_node("code", code_agent)

# Use add_edge with START constant (modern pattern)
workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {"research": "research", "code": "code", "FINISH": END}
)
workflow.add_edge("research", "supervisor")  # Back to supervisor
workflow.add_edge("code", "supervisor")

app = workflow.compile()