2. Pattern: Supervisor

Overview

In the previous capsule you saw why a single agent that does everything — researches, analyzes, writes, evaluates — hits a point of diminishing returns. The context window saturates, the tools compete for the LLM's attention, and the agent loses focus. The solution: specialization. Multiple agents, each optimized for a specific task. But if you have 3 or 4 specialized agents, someone has to coordinate them. Someone has to decide who works, in what order, and when it's done.

That "someone" is the Supervisor. It's the first multi-agent orchestration pattern you'll implement, and probably the most intuitive one. Think of a team manager: they receive a project, break it into tasks, assign each task to the right specialist, review the results, and decide whether more work is needed or the project is complete. The Supervisor doesn't do the work — it coordinates the people who do.

Connection with the module: This capsule gives you the most fundamental orchestration pattern. In capsule 03 you'll see Handoffs (direct transfer between agents, with no central coordinator). In 04, Subagents (delegation with isolated context). In 05, Router (pure classification). By capsule 07, you'll combine these patterns. But everything starts here: the Supervisor is the base pattern the others are built on.

The Supervisor pattern

The core idea

The Supervisor pattern splits a multi-agent system into two roles:

1. Supervisor: An agent (LLM node) that makes routing decisions. It doesn't execute tasks — it decides who executes them.
2. Workers: Specialized agents that execute tasks. Each worker has its own tools, its own prompt, and its own area of expertise.

The flow is always the same: the user sends a task to the Supervisor → the Supervisor analyzes it and picks a worker → the worker executes and returns results → the Supervisor reviews → decides whether another worker should work or whether it's done → when everything is ready, it synthesizes the final answer.

Visual architecture

                    ┌─────────────────┐
                    │   SUPERVISOR    │
                    │                 │
    Task ─────────► │  1. Analyze     │
                    │  2. Decide next │
                    │  3. Review      │
                    │  4. Synthesize  │
                    └──┬──────┬───┬───┘
                       │      │   │
              ┌────────┘      │   └────────┐
              ▼               ▼            ▼
        ┌───────────┐  ┌───────────┐ ┌───────────┐
        │ RESEARCHER│  │  ANALYST  │ │  WRITER   │
        │           │  │           │ │           │
        │ Tools:    │  │ Tools:    │ │ Tools:    │
        │ - search  │  │ - analyze │ │ - write   │
        │ - fetch   │  │ - compare │ │ - format  │
        └───────────┘  └───────────┘ └───────────┘

Flow: Supervisor → Worker → Supervisor → Worker → ... → Supervisor → END

Control always returns to the Supervisor after each worker. This is what separates the Supervisor from a sequential pipeline:

PIPELINE (fixed sequence)
─────────────────────────
Researcher → Analyst → Writer → END
  Always the same order. No runtime decisions.

SUPERVISOR (dynamic decision)
─────────────────────────────
Supervisor → Researcher → Supervisor → Writer → Supervisor → END
  The Supervisor decides who's next. The order changes with the task.

If the task is "write a summary of X", the Supervisor can go straight to the Writer. If it's "research X, analyze the data, and write a report", it orchestrates all three. If the analysis reveals missing data, it sends it back to the Researcher. The routing is dynamic.

Implementation with LangGraph

The shared state

All the nodes share a state that defines what information flows between agents:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage

class SupervisorState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_worker: str
    task_complete: bool

The Supervisor node

The Supervisor invokes an LLM with routing instructions. It uses no tools — its job is to decide:

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

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

SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of specialized agents.

Available agents:
- researcher: Searches for information and sources. Use it when you need data.
- analyst: Analyzes data, compares sources, extracts insights.
- writer: Writes coherent text, creates reports.

Your job:
1. Analyze the user's task
2. Decide which agent should work NOW
3. If all the necessary steps are complete, respond FINISH

Respond ONLY with: researcher, analyst, writer, or FINISH."""

def supervisor_node(state: SupervisorState) -> dict:
    response = model.invoke(
        [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    )
    decision = response.content.strip().lower()

    return {
        "next_worker": decision,
        "task_complete": decision == "finish",
        "messages": [response]
    }

The Worker nodes

Each worker is an agent with a specialized prompt and its own tools:

from langgraph.prebuilt import create_react_agent

researcher = create_react_agent(
    model, tools=[search_web, fetch_url],
    prompt="You are an expert researcher. Find relevant information and report findings with sources."
)

analyst = create_react_agent(
    model, tools=[calculate, compare_data],
    prompt="You are a data analyst. Evaluate the information and present conclusions."
)

writer = create_react_agent(
    ChatOpenAI(model="gpt-4o-mini"),  # cheaper model for writing
    tools=[format_text],
    prompt="You are a technical writer. Synthesize the information into coherent text."
)

To wire these agents in as graph nodes, use wrappers:

def make_worker_node(agent, name: str):
    """Wrapper that runs a sub-agent and returns its labeled result."""
    def worker_node(state: SupervisorState) -> dict:
        last_message = state["messages"][-1]
        result = agent.invoke({"messages": [HumanMessage(content=last_message.content)]})
        output = result["messages"][-1]
        return {
            "messages": [HumanMessage(content=f"[{name}] {output.content}", name=name)]
        }
    return worker_node

researcher_node = make_worker_node(researcher, "researcher")
analyst_node = make_worker_node(analyst, "analyst")
writer_node = make_worker_node(writer, "writer")

The complete graph

from langgraph.graph import StateGraph, START, END

builder = StateGraph(SupervisorState)

builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("analyst", analyst_node)
builder.add_node("writer", writer_node)

builder.add_edge(START, "supervisor")

def route_from_supervisor(state: SupervisorState) -> str:
    if state.get("task_complete"):
        return "end"
    return state["next_worker"]

builder.add_conditional_edges("supervisor", route_from_supervisor, {
    "researcher": "researcher", "analyst": "analyst",
    "writer": "writer", "end": END,
})

builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")
builder.add_edge("writer", "supervisor")

graph = builder.compile()

Every worker always returns to the Supervisor. The Supervisor always decides what comes next. This cycle is the heart of the pattern.

Designing the Supervisor

The free-text problem

The Supervisor in the previous example replies with free text: "researcher". That has problems: the LLM might reply "Researcher" (capitalized), "the researcher", or invent workers that don't exist. In production, you need structured output.

Structured output with Pydantic

from pydantic import BaseModel, Field

class SupervisorDecision(BaseModel):
    next_worker: Literal["researcher", "analyst", "writer", "FINISH"] = Field(
        description="The next agent, or FINISH if the task is complete"
    )
    reasoning: str = Field(
        description="Brief justification for the decision"
    )

structured_model = model.with_structured_output(SupervisorDecision)

def supervisor_node(state: SupervisorState) -> dict:
    decision: SupervisorDecision = structured_model.invoke(
        [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    )
    is_done = decision.next_worker == "FINISH"

    return {
        "next_worker": decision.next_worker.lower(),
        "task_complete": is_done,
        "messages": [HumanMessage(
            content=f"[supervisor] → {decision.next_worker} ({decision.reasoning})",
            name="supervisor"
        )]
    }

Supervisor with progress tracking

A more sophisticated Supervisor tracks which workers have already run:

class SupervisorState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_worker: str
    task_complete: bool
    workers_called: list[str]

SUPERVISOR_PROMPT_V2 = """You are a supervisor coordinating: researcher, analyst, writer.

Workers that ALREADY worked: {workers_called}

Rules:
- Do not repeat a worker unless its results are insufficient
- When all the necessary steps are complete, respond FINISH

Who should work now?"""

def supervisor_node(state: SupervisorState) -> dict:
    workers_called = state.get("workers_called", [])
    prompt = SUPERVISOR_PROMPT_V2.format(
        workers_called=", ".join(workers_called) or "none"
    )
    decision: SupervisorDecision = structured_model.invoke(
        [SystemMessage(content=prompt)] + state["messages"]
    )
    is_done = decision.next_worker == "FINISH"

    return {
        "next_worker": decision.next_worker.lower(),
        "task_complete": is_done,
        "workers_called": workers_called + ([decision.next_worker] if not is_done else []),
        "messages": [HumanMessage(
            content=f"[supervisor] → {decision.next_worker} ({decision.reasoning})",
            name="supervisor"
        )]
    }

workers_called prevents infinite loops. If ["researcher", "analyst"] already worked and the results are good, the Supervisor knows only writer or FINISH is left.

Workers as specialized agents

A common mistake is making "dumb" workers — simple Python functions that run a fixed operation. That's a pipeline. The workers in the Supervisor pattern are full agents with their own LLM, tools, prompt, and reasoning cycle.

Worker with specialized tools

from langchain_core.tools import tool

@tool
def extract_key_findings(text: str) -> str:
    """Extract the main findings from a research text."""
    response = model.invoke(f"Extract the 3-5 most important findings:\n\n{text}")
    return response.content

researcher = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[TavilySearchResults(max_results=5), extract_key_findings],
    prompt="You are an expert researcher. Find information and extract key findings. Cite sources."
)

Workers with MCP tools (M7)

If you completed module 7, connect workers to MCP servers:

from langchain_mcp_adapters.client import MultiServerMCPClient

async def create_researcher_with_mcp():
    client = MultiServerMCPClient({
        "research": {"command": "python", "args": ["research_server.py"], "transport": "stdio"}
    })
    tools = await client.get_tools()
    return create_react_agent(
        ChatOpenAI(model="gpt-4o"), tools=tools,
        prompt="You are an expert researcher. Use the available tools."
    )

Each worker discovers its own tools. The Supervisor doesn't know which tools each worker has — it only knows what kind of task it can assign to them.

Different models per worker

In multi-agent, optimize cost by assigning models according to complexity:

Aggregating results

With three workers producing independent results, the Supervisor needs to synthesize — concatenating isn't enough.

Final synthesis node

def synthesize_node(state: SupervisorState) -> dict:
    worker_results = [
        msg.content for msg in state["messages"]
        if getattr(msg, "name", None) in ["researcher", "analyst", "writer"]
    ]
    if not worker_results:
        return {"messages": [HumanMessage(content="There are no results to synthesize.")]}

    synthesis_prompt = f"""Integrate these agent results into a coherent final answer:

{chr(10).join(f'--- Result ---{chr(10)}{r}' for r in worker_results)}

Do not repeat information. Present conclusions in a structured way."""

    response = model.invoke([HumanMessage(content=synthesis_prompt)])
    return {"messages": [response]}

Add this node to the graph so it runs when task_complete = True:

builder.add_node("synthesize", synthesize_node)

builder.add_conditional_edges("supervisor", route_from_supervisor, {
    "researcher": "researcher", "analyst": "analyst",
    "writer": "writer", "synthesize": "synthesize",
})
builder.add_edge("synthesize", END)

Aggregation strategies

When to use Supervisor

Good use cases

• Centralized control. One entity decides everything: what gets done, in what order, with what priority.
• Progress monitoring. The Supervisor sees the whole state and can detect failures or insufficient results.
• Multi-phase tasks. "Research X, analyze the results, write a report" — a logical sequence with dynamic routing.
• Workers with very different expertise. Researcher and Writer need fundamentally different tools and prompts.

Bad use cases

• Simple task. If a single agent with 3 tools can solve it, Supervisor is unnecessary overhead.
• High autonomy required. If the agents need to talk directly to each other, the central coordinator is a bottleneck. Consider Handoffs (capsule 03).
• Always the same fixed flow. If it's always Researcher → Analyst → Writer with no variation, a sequential pipeline is simpler.
• Latency is critical. Every trip through the Supervisor is an LLM call. With 3 workers, that's at least 7 LLM calls.

Decision framework

Does the task have sub-tasks with different expertise?
  └─ No → A single agent is enough
  └─ Yes → Is the order of sub-tasks always fixed?
              └─ Yes → Sequential pipeline
              └─ No → Do you need centralized control?
                         └─ Yes → SUPERVISOR ✓
                         └─ No → Handoffs or Router

Protection against infinite loops

A real risk: the Supervisor sends work to the Researcher, reviews the results, decides they're not good enough, sends it to the Researcher again, and repeats forever.

MAX_SUPERVISOR_ITERATIONS = 10

def supervisor_node(state: SupervisorState) -> dict:
    iteration = state.get("iteration_count", 0) + 1

    if iteration > MAX_SUPERVISOR_ITERATIONS:
        return {
            "next_worker": "finish", "task_complete": True,
            "iteration_count": iteration,
            "messages": [HumanMessage(
                content="[supervisor] Iteration limit reached. Finishing.",
                name="supervisor"
            )]
        }

    decision: SupervisorDecision = structured_model.invoke(
        [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    )

    return {
        "next_worker": decision.next_worker.lower(),
        "task_complete": decision.next_worker == "FINISH",
        "iteration_count": iteration,
        "messages": [HumanMessage(
            content=f"[supervisor] Iter {iteration}: → {decision.next_worker}",
            name="supervisor"
        )]
    }

In production, add logging when the limit is reached so you can investigate why the Supervisor didn't converge.

Connection with the project

In the project of this module (capsule 08), the Research Agent expands into a multi-agent system with Supervisor as its base:

┌─────────────────────────────────────────────┐
│              SUPERVISOR AGENT               │
│  - Decomposes task (planning M5)            │
│  - Assigns to specialized workers           │
│  - Uses memory (M6) for tracking            │
│  - Synthesizes final result                 │
└──────┬──────────┬──────────────┬────────────┘
       ▼          ▼              ▼
┌────────────┐ ┌──────────┐ ┌──────────┐
│ RESEARCHER │ │ ANALYST  │ │  WRITER  │
│ MCP: search│ │ analyze  │ │ MCP: fs  │
│      papers│ │ compare  │ │   format │
└────────────┘ └──────────┘ └──────────┘

In the following capsules, you'll see alternative patterns that solve the same problem with different trade-offs. In capsule 07, you'll combine patterns: a Router that directs to a Supervisor, which coordinates workers that use Subagents.

Troubleshooting

Problem 1: The Supervisor falls into an infinite loop

Cause: It repeatedly decides the results aren't good enough.

Solution: Implement MAX_SUPERVISOR_ITERATIONS. Add workers_called to the prompt so it knows what already ran. If it persists, check whether the prompt is too demanding about its completeness criteria.

Problem 2: The Supervisor always picks the same worker

Cause: The worker descriptions don't clearly differentiate their capabilities.

Solution: Make the descriptions mutually exclusive. Instead of "researcher: searches for information" and "analyst: analyzes information", use "researcher: searches the web, returns sources with URLs" and "analyst: compares numeric data, identifies statistical trends".

Problem 3: The worker results never reach the Supervisor

Cause: The worker wrapper isn't returning messages correctly.

Solution: Verify that make_worker_node returns {"messages": [...]} and that each message has a name to identify the worker. Debug by printing len(state['messages']) inside the Supervisor.

Problem 4: "KeyError" in the conditional edges

Cause: The Supervisor returns a next_worker that isn't in the mapping.

Solution: Use structured output with Literal to guarantee valid values. If you use free text, normalize it and add a fallback: if decision not in valid_workers: decision = "finish".

Problem 5: The system is slow with many workers

Cause: Each worker runs its own ReAct cycle, and the Supervisor adds overhead.

Solution: Measure timings per node. Use cheaper models for workers that don't need complex reasoning. Limit max_steps in create_react_agent. Evaluate whether you really need a Supervisor or whether a pipeline is enough.

Exercises

Exercise 1: Basic Supervisor with two workers (Easy)

Implement a Supervisor system with two workers: summarizer (summarizes texts) and translator (translates to Spanish). The Supervisor receives "Summarize this text and translate it to Spanish" and coordinates both. The workers can be simple nodes that invoke an LLM.

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
from typing import Annotated, Literal
from typing_extensions import TypedDict
import operator

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

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_worker: str
    task_complete: bool

class Decision(BaseModel):
    next_worker: Literal["summarizer", "translator", "FINISH"] = Field(description="Next worker or FINISH")

structured = model.with_structured_output(Decision)

def supervisor(state):
    decision = structured.invoke(
        [SystemMessage(content="You coordinate summarizer and translator. First the summary, then the translation, then FINISH.")]
        + state["messages"]
    )
    return {"next_worker": decision.next_worker.lower(), "task_complete": decision.next_worker == "FINISH",
            "messages": [HumanMessage(content=f"[supervisor] → {decision.next_worker}", name="supervisor")]}

def summarizer(state):
    r = model.invoke([SystemMessage(content="Summarize in 2-3 sentences.")] + state["messages"])
    return {"messages": [HumanMessage(content=f"[summarizer] {r.content}", name="summarizer")]}

def translator(state):
    r = model.invoke([SystemMessage(content="Translate the most recent content to Spanish.")] + state["messages"])
    return {"messages": [HumanMessage(content=f"[translator] {r.content}", name="translator")]}

builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("summarizer", summarizer)
builder.add_node("translator", translator)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor",
    lambda s: "end" if s.get("task_complete") else s["next_worker"],
    {"summarizer": "summarizer", "translator": "translator", "end": END})
builder.add_edge("summarizer", "supervisor")
builder.add_edge("translator", "supervisor")
graph = builder.compile()

Exercise 2: Structured output with reasoning and confidence (Medium)

Modify the previous exercise so the Supervisor uses a Pydantic model with next_worker, reasoning (why it picked that worker), and confidence (float 0.0-1.0). Print these decisions during execution.

class DetailedDecision(BaseModel):
    next_worker: Literal["summarizer", "translator", "FINISH"] = Field(description="Next worker")
    reasoning: str = Field(description="Why you picked this worker")
    confidence: float = Field(description="Confidence 0.0 to 1.0", ge=0.0, le=1.0)

detailed = model.with_structured_output(DetailedDecision)

def supervisor(state):
    decision = detailed.invoke(
        [SystemMessage(content="You coordinate summarizer and translator. Evaluate your confidence.")]
        + state["messages"]
    )
    print(f"  → {decision.next_worker} | confidence: {decision.confidence:.1f} | {decision.reasoning}")
    return {"next_worker": decision.next_worker.lower(), "task_complete": decision.next_worker == "FINISH",
            "messages": [HumanMessage(
                content=f"[supervisor] → {decision.next_worker} (conf: {decision.confidence:.1f})",
                name="supervisor")]}

Exercise 3: Anti-loop protection with tracking (Medium)

Add to the Supervisor: iteration_count, workers_called, and MAX_ITERATIONS = 6. If it hits the limit, finish with a message that includes which workers ran and how many iterations were used.

MAX_ITERATIONS = 6

class SafeState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_worker: str
    task_complete: bool
    workers_called: list[str]
    iteration_count: int

def safe_supervisor(state):
    iteration = state.get("iteration_count", 0) + 1
    called = state.get("workers_called", [])

    if iteration > MAX_ITERATIONS:
        return {"next_worker": "finish", "task_complete": True, "iteration_count": iteration,
                "workers_called": called,
                "messages": [HumanMessage(
                    content=f"[supervisor] Limit ({MAX_ITERATIONS}). Workers: {called}", name="supervisor")]}

    prompt = f"Workers already run: {called or 'none'}. Iteration {iteration}/{MAX_ITERATIONS}."
    decision = structured.invoke([SystemMessage(content=prompt)] + state["messages"])
    is_done = decision.next_worker == "FINISH"

    return {"next_worker": decision.next_worker.lower(), "task_complete": is_done,
            "iteration_count": iteration,
            "workers_called": called + ([decision.next_worker] if not is_done else []),
            "messages": [HumanMessage(
                content=f"[supervisor] Iter {iteration}: → {decision.next_worker}", name="supervisor")]}

Exercise 4: Workers with create_react_agent and tools (Hard)

Create a system with 2 workers that use create_react_agent: calculator (tools: add, multiply) and formatter (tool: format_number). The Supervisor coordinates: "Compute 15 × 7 + 23, then format it with 2 decimals".

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

@tool
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

@tool
def format_number(n: float, decimals: int = 2) -> str:
    """Format a number with N decimals."""
    return f"{n:.{decimals}f}"

calc_agent = create_react_agent(model, tools=[add, multiply],
    prompt="You are a calculator. Use the tools to operate. Report the result.")
fmt_agent = create_react_agent(ChatOpenAI(model="gpt-4o-mini"), tools=[format_number],
    prompt="Format numbers. Extract the number from the message and apply format_number.")

def calc_node(state):
    result = calc_agent.invoke({"messages": state["messages"]})
    return {"messages": [HumanMessage(content=f"[calculator] {result['messages'][-1].content}", name="calculator")]}

def fmt_node(state):
    result = fmt_agent.invoke({"messages": state["messages"]})
    return {"messages": [HumanMessage(content=f"[formatter] {result['messages'][-1].content}", name="formatter")]}

class CalcDecision(BaseModel):
    next_worker: Literal["calculator", "formatter", "FINISH"] = Field(description="Next worker")

calc_structured = model.with_structured_output(CalcDecision)

def calc_supervisor(state):
    decision = calc_structured.invoke(
        [SystemMessage(content="You coordinate calculator and formatter. First compute, then format.")]
        + state["messages"])
    return {"next_worker": decision.next_worker.lower(),
            "task_complete": decision.next_worker == "FINISH",
            "messages": [HumanMessage(content=f"[supervisor] → {decision.next_worker}", name="supervisor")]}

builder = StateGraph(SupervisorState)
builder.add_node("supervisor", calc_supervisor)
builder.add_node("calculator", calc_node)
builder.add_node("formatter", fmt_node)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor",
    lambda s: "end" if s.get("task_complete") else s["next_worker"],
    {"calculator": "calculator", "formatter": "formatter", "end": END})
builder.add_edge("calculator", "supervisor")
builder.add_edge("formatter", "supervisor")
graph = builder.compile()

Exercise 5: Complete system with synthesis (Hard)

Build the complete system: Supervisor + 3 workers (researcher, analyst, writer) + a synthesize node + anti-loop protection + structured output with reasoning. When the Supervisor decides FINISH, the flow goes to synthesize, which integrates all the results.

MAX_ITERATIONS = 8

class FullState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_worker: str
    task_complete: bool
    workers_called: list[str]
    iteration_count: int

class FullDecision(BaseModel):
    next_worker: Literal["researcher", "analyst", "writer", "FINISH"] = Field(description="Next")
    reasoning: str = Field(description="Justification")

full_structured = model.with_structured_output(FullDecision)

def supervisor(state):
    iteration = state.get("iteration_count", 0) + 1
    called = state.get("workers_called", [])
    if iteration > MAX_ITERATIONS:
        return {"next_worker": "finish", "task_complete": True, "iteration_count": iteration,
                "workers_called": called,
                "messages": [HumanMessage(content=f"[supervisor] Limit reached", name="supervisor")]}

    decision = full_structured.invoke(
        [SystemMessage(content=f"You coordinate researcher, analyst, writer. Already run: {called}. Iter {iteration}/{MAX_ITERATIONS}.")]
        + state["messages"])
    is_done = decision.next_worker == "FINISH"
    return {"next_worker": decision.next_worker.lower(), "task_complete": is_done,
            "iteration_count": iteration,
            "workers_called": called + ([decision.next_worker] if not is_done else []),
            "messages": [HumanMessage(
                content=f"[supervisor] → {decision.next_worker} ({decision.reasoning})", name="supervisor")]}

def make_worker(name, instruction):
    def node(state):
        r = model.invoke([SystemMessage(content=instruction)] + state["messages"])
        return {"messages": [HumanMessage(content=f"[{name}] {r.content}", name=name)]}
    return node

def synthesize(state):
    results = [m.content for m in state["messages"] if getattr(m, "name", None) in ["researcher", "analyst", "writer"]]
    r = model.invoke([HumanMessage(content=f"Integrate these results:\n\n{chr(10).join(results)}")])
    return {"messages": [r]}

builder = StateGraph(FullState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", make_worker("researcher", "Research the topic in detail."))
builder.add_node("analyst", make_worker("analyst", "Analyze the information. Identify pros and cons."))
builder.add_node("writer", make_worker("writer", "Write a conclusion based on the information."))
builder.add_node("synthesize", synthesize)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor",
    lambda s: "synthesize" if s.get("task_complete") else s["next_worker"],
    {"researcher": "researcher", "analyst": "analyst", "writer": "writer", "synthesize": "synthesize"})
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")
builder.add_edge("writer", "supervisor")
builder.add_edge("synthesize", END)
graph = builder.compile()

Summary

In this capsule you implemented the Supervisor pattern — the first and most fundamental multi-agent orchestration pattern:

• The Supervisor is a coordinator, not an executor. It receives the task, decides which worker should work, reviews results, and decides the next step. The workers execute.
• Dynamic routing is what separates the Supervisor from a pipeline. The order isn't hardcoded — the Supervisor decides at runtime.
• Structured output with Pydantic eliminates fragile free-text parsing. next_worker validated against a Literal, reasoning for explainability.
• Workers are complete agents with their own LLM, tools, prompt, and ReAct cycle. They can use create_react_agent or MCP tools.
• Aggregating results requires a dedicated synthesis node that integrates the outputs of multiple workers.
• Anti-loop protection with iteration_count and MAX_ITERATIONS is mandatory in production.
• Supervisor isn't always the answer. If the task is simple, the flow is fixed, or the agents need high autonomy, consider other patterns.

Next capsule: Handoffs Pattern — direct transfer of control between agents with no central coordinator. What happens when agents talk to each other instead of going through a Supervisor?

Additional Resources

1. LangGraph — Multi-Agent Supervisor Tutorial — Official tutorial for implementing the Supervisor pattern
2. LangGraph — Multi-Agent Systems — Conceptual documentation on multi-agent systems
3. Structured Output — LangChain — Reference for with_structured_output for typed routing
4. create_react_agent — LangGraph — API reference for creating the ReAct agents used as workers
5. Multi-Agent Patterns — LangChain Blog — Post comparing multi-agent orchestration patterns