Module 10: Multi-Agent Systems

Pattern Supervisor

Capsule overview

Your Research Agent v4 does everything: it searches, analyzes, and writes. In the previous capsule you saw why that becomes a problem as complexity grows. Now you're going to solve it with the most common multi-agent pattern: the Supervisor.

The Supervisor pattern works like a project manager coordinating specialists. A central agent (the supervisor) receives the user's query, decides which specialist should work, delegates the task, receives the result, decides the next step, and repeats until it has a complete answer. The supervisor does NOT do the work — it coordinates the people who do.

In this capsule you'll build two versions: a manual supervisor with StateGraph (rule-based routing, no LLM for the routing decisions) and an LLM supervisor using create_supervisor (the model decides who to delegate to). You'll start simple and scale only if you need to — the same philosophy you apply to everything in engineering.


The supervisor's job: coordinate, don't execute

The most common mistake when designing a supervisor is turning it into a "super-agent" that understands everything. No. The supervisor is a router: it takes a task, decides who to hand it to, and collects results. It doesn't need to understand the content of the research — it just needs to know who to send it to.

A supervisor is NOT:                 A supervisor IS:
  An expert in everything              A coordinator
  An agent with 15 tools               A router with clear rules
  The one who does the work            The one who decides WHO does the work
  A genius who understands it all      A manager who knows who to ask

Think of a project manager: they don't need to know how to code in order to coordinate a team of developers. They need to know who's good at what, when to assign each task, and when a result is good enough.


Architecture

  User → Supervisor → Researcher → Supervisor → Analyst → Supervisor → Answer
             │                        │                       │
             └── decides who ─────────┴── decides next ──────┘

The flow: the user sends a query to the supervisor. The supervisor decides which worker needs to run, delegates, receives the result, and decides the next step. It repeats until it has enough information to answer.


Manual supervisor with StateGraph

We start with the simplest version: the supervisor is a Python function that decides the next step based on rules. No LLM for the routing.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class ResearchTeamState(TypedDict):
    query: str
    research: str
    analysis: str
    report: str
    next_worker: str
    iteration: int


def supervisor(state: ResearchTeamState) -> dict:
    iteration = state.get("iteration", 0)
    if iteration == 0:
        return {"next_worker": "researcher", "iteration": 1}
    if iteration == 1:
        return {"next_worker": "analyst", "iteration": 2}
    if iteration == 2:
        return {"next_worker": "reporter", "iteration": 3}
    return {"next_worker": "done", "iteration": iteration + 1}


def researcher(state: ResearchTeamState) -> dict:
    query = state["query"]
    results = (
        f"Research on '{query}':\n"
        f"  - Source 1: Stanford paper on RAG (2025)\n"
        f"  - Source 2: LangChain blog on hybrid search\n"
        f"  - Source 3: MTEB benchmark for embeddings"
    )
    return {"research": results}


def analyst(state: ResearchTeamState) -> dict:
    analysis = (
        f"Analysis of findings:\n"
        f"  - Pattern 1: Hybrid search beats dense retrieval by 15%\n"
        f"  - Pattern 2: Re-ranking improves precision significantly\n"
        f"  - Contradiction: Stanford paper vs MTEB benchmark on embeddings"
    )
    return {"analysis": analysis}


def reporter(state: ResearchTeamState) -> dict:
    report = (
        f"REPORT: {state['query']}\n"
        f"{'='*40}\n"
        f"{state['research']}\n\n"
        f"{state['analysis']}\n\n"
        f"Conclusion: hybrid search + re-ranking is the state of the art."
    )
    return {"report": report}


def route_to_worker(state: ResearchTeamState) -> str:
    return state["next_worker"]


builder = StateGraph(ResearchTeamState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_node("reporter", reporter)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_to_worker, {
    "researcher": "researcher",
    "analyst": "analyst",
    "reporter": "reporter",
    "done": END,
})
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")
builder.add_edge("reporter", "supervisor")

graph = builder.compile()

result = graph.invoke({
    "query": "State of the art in RAG 2025",
    "research": "", "analysis": "", "report": "",
    "next_worker": "", "iteration": 0,
})

print(result["report"][:80] + "...")
print(f"Supervisor iterations: {result['iteration']}")
# Expected output:
# REPORT: State of the art in RAG 2025
# ============================================...
# Supervisor iterations: 4

The supervisor ran 4 times: it delegated to researcher (1), got the research back and delegated to analyst (2), got the analysis back and delegated to reporter (3), got the report back and finished (4). The hub-and-spoke pattern: everything flows through the supervisor.


The name parameter: identifying agents

When you use create_agent for workers, the name parameter is key. It identifies the agent in the system, in the logs, and in the handoffs:

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool


@tool
def web_search(query: str) -> str:
    """Search the web for information about a topic."""
    return f"Web results for '{query}': 5 articles found."


researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[web_search],
    name="researcher",
    prompt="You are an expert researcher. Your ONLY job is to search for information.",
)

print(f"Agent name: {researcher.name}")
# Expected output:
# Agent name: researcher

The name shows up in the logs, in the handoff tools (transfer_to_researcher), and in the messages tagged by agent.


The supervisor's system prompt

The supervisor's prompt defines the delegation rules. It has to be clear about which worker handles what:

SUPERVISOR_PROMPT = """You are the supervisor of a research team.

Your team:
- researcher: searches the web and academic papers
- analyst: analyzes data, finds patterns and contradictions
- writer: generates executive reports

Rules:
1. To search for information → delegate to the researcher
2. To analyze findings → delegate to the analyst
3. To generate the final report → delegate to the writer
4. When the report is complete → answer the user directly

Do NOT do the work yourself. ALWAYS delegate to the right specialist.
"""

A good supervisor prompt:

  • ✅ Lists the available workers with their specialties
  • ✅ Defines clear routing rules
  • ✅ Specifies when to stop
  • ❌ Doesn't try to understand the content — it just coordinates

LLM Supervisor with create_supervisor

The idiomatic way to build a supervisor in LangGraph: create_supervisor from the langgraph-supervisor package. It creates the handoff tools automatically and the LLM decides who to delegate to.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph_supervisor import create_supervisor


@tool
def web_search(query: str) -> str:
    """Search the web for information about a topic."""
    return f"Results for '{query}': RAG combines retrieval with generation to improve LLM answers."


@tool
def analyze_text(text: str) -> str:
    """Analyze a text and extract key patterns."""
    return f"Analysis: the main trend in '{text[:30]}...' is hybrid search + re-ranking."


researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[web_search],
    name="researcher",
    prompt="You are a researcher. Find relevant information and report your findings.",
)

analyst = create_agent(
    init_chat_model("openai:gpt-4.1"),
    tools=[analyze_text],
    name="analyst",
    prompt="You are an analyst. Synthesize the information and find patterns.",
)

workflow = create_supervisor(
    [researcher, analyst],
    model=init_chat_model("openai:gpt-4.1"),
    prompt=(
        "You are a research supervisor. "
        "Delegate information searches to the researcher. "
        "Delegate pattern analysis to the analyst. "
        "When you have enough information, produce the final answer."
    ),
)

app = workflow.compile()

result = app.invoke({
    "messages": [("user", "What's the current state of RAG in 2025?")]
})

print(result["messages"][-1].content)
# Expected output (varies by LLM):
# Based on the research and analysis, the current state of RAG in 2025
# centers on hybrid search combined with re-ranking...

create_supervisor does three things automatically:

  1. Creates handoff tools (transfer_to_researcher, transfer_to_analyst) the LLM can call
  2. Wires the supervisor together with the workers in a graph
  3. Handles the message flow between supervisor and workers

How the handoff works

The handoff isn't magic — it's a tool call. create_supervisor generates tools like transfer_to_researcher and transfer_to_analyst. When the supervisor's LLM decides to call one, the framework transfers control to the matching worker. The worker runs its tools, produces its answer, and control goes back to the supervisor. The supervisor evaluates: need more? → delegate to someone else. Enough? → produce the final answer.


Stop conditions: when to finish

The supervisor needs to know when to stop. There are three strategies:

1. The supervisor decides (default in create_supervisor)

The supervisor's LLM decides when it has enough information and answers the user directly instead of delegating to another worker.

2. Iteration limit

Add a max_iterations to the state and check it in the supervisor:

def supervisor_with_limit(state):
    if state["iteration"] >= state["max_iterations"]:
        return {"next_worker": "done"}
    # ... normal routing ...

Always include an iteration limit as a safety net, even with an LLM supervisor.

3. Quality check

The supervisor evaluates whether the result meets a quality bar before finishing. Useful with an LLM supervisor.


Simple vs LLM supervisor: when to use each

CriterionSimple supervisor (rules)LLM supervisor
RoutingKeyword matching, if/elseThe LLM decides based on context
CostNo extra LLM costAt least 1 extra LLM call per decision
FlexibilityPredictable, fixed flowAdapts to unexpected queries
DebuggingTrivial (clear rules)Hard (the LLM can make unexpected calls)
LatencyMinimal (just Python)+500ms-2s per routing decision
Use casePredictable flows, fixed pipelinesVaried queries, complex routing

The rule: start with a simple supervisor. If rule-based routing doesn't cover your cases, move up to an LLM. Most production systems use a hybrid: rules for the common cases, an LLM for the edge cases.


Model optimization per worker

One of the most concrete advantages of multi-agent: each worker uses the optimal model for its task.

researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),  # cheap: $0.40/M input
    tools=[web_search],
    name="researcher",
    prompt="Find relevant information.",
)

analyst = create_agent(
    init_chat_model("openai:gpt-4.1"),  # powerful: $2.00/M input
    tools=[analyze_data],
    name="analyst",
    prompt="Analyze data and find complex patterns.",
)

The researcher doesn't need a powerful model — searching the web is simple. The analyst does need complex reasoning. The result: roughly 60% savings on the search side with no loss of quality in the analysis.


Collecting results: output_mode

create_supervisor has an output_mode parameter that controls how much of the worker's information reaches the supervisor:

  • "last_message" (default): only each worker's last message. Fewer tokens, more efficient.
  • "full_history": the worker's entire history. More context, useful if the supervisor needs to see the full reasoning.

Start with "last_message". Only switch to "full_history" if the supervisor makes bad decisions for lack of context.


Troubleshooting

Problem 1: "The supervisor gets stuck in an infinite loop between two workers"

Symptom: The supervisor delegates to the researcher, gets a result, delegates to the analyst, gets a result, and goes back to the researcher indefinitely.

Cause: The supervisor's prompt has no clear instructions about when to stop. The LLM doesn't know that "enough" means answering directly.

Fix: Add an explicit instruction to the prompt: "When you have enough information to answer the user's query, answer directly instead of delegating to another agent." Also consider an iteration limit as a safety net.

Problem 2: "The supervisor always delegates to the same worker"

Symptom: No matter the query, the supervisor always calls transfer_to_researcher.

Cause: The supervisor's prompt doesn't clearly differentiate each worker's specialty. Or the worker names aren't descriptive.

Fix: Use descriptive names (researcher, analyst, not agent_1, agent_2). Describe each worker's specialty in the supervisor's prompt: "the researcher searches for information, the analyst analyzes data."

Problem 3: "create_supervisor can't find the agents"

Symptom: An error when compiling the workflow with create_supervisor.

Cause: The agents passed to create_supervisor don't have a name attribute, or they aren't compiled Pregel objects.

Fix: Check that each agent was created with create_agent (which returns a compiled graph) and that it has a name set. create_agent(model, tools=[...], name="my_agent") — the name is mandatory for multi-agent.

Problem 4: "The workers can't see other workers' context"

Symptom: The analyst can't see what the researcher found. Cause: With output_mode="last_message", only the worker's last message propagates. Fix: Switch to output_mode="full_history", or design your workers to consolidate all the information into a single final message.


Exercises

Exercise 1: Manual supervisor with 2 workers (Easy)

Build a manual supervisor that coordinates translator (translates to English) and summarizer (summarizes text). The supervisor delegates in order: translator → summarizer.

See solution
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class TranslateState(TypedDict):
    original_text: str
    translated: str
    summary: str
    next_worker: str
    step: int


def supervisor(state: TranslateState) -> dict:
    if state["step"] == 0:
        return {"next_worker": "translator", "step": 1}
    elif state["step"] == 1:
        return {"next_worker": "summarizer", "step": 2}
    return {"next_worker": "done", "step": 3}


def translator(state: TranslateState) -> dict:
    return {"translated": f"[FR] Translation of: {state['original_text'][:50]}..."}


def summarizer(state: TranslateState) -> dict:
    return {"summary": f"Summary: {state['translated'][:30]}... (3 key points)"}


def route(state: TranslateState) -> str:
    return state["next_worker"]


builder = StateGraph(TranslateState)
builder.add_node("supervisor", supervisor)
builder.add_node("translator", translator)
builder.add_node("summarizer", summarizer)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "translator": "translator", "summarizer": "summarizer", "done": END,
})
builder.add_edge("translator", "supervisor")
builder.add_edge("summarizer", "supervisor")

graph = builder.compile()
result = graph.invoke({
    "original_text": "Artificial intelligence is transforming the financial industry.",
    "translated": "", "summary": "", "next_worker": "", "step": 0,
})

print(f"Translated: {result['translated']}")
print(f"Summary: {result['summary']}")
# Expected output:
# Translated: [FR] Translation of: Artificial intelligence is transforming the fina...
# Summary: Summary: [FR] Translation of: Artif... (3 key points)

Explanation: The supervisor runs 3 times: it delegates to translator (1), delegates to summarizer (2), finishes (3). Each worker only does its own specific task.

Exercise 2: Add an iteration limit to the supervisor (Easy)

Modify the manual supervisor from the main example so it has a configurable max_iterations. If the supervisor goes past the limit, it should finish with whatever it has. Try it with max_iterations=2 (research only, no analysis, no report).

See solution
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class LimitedState(TypedDict):
    query: str
    research: str
    analysis: str
    next_worker: str
    iteration: int
    max_iterations: int


def supervisor(state: LimitedState) -> dict:
    iteration = state.get("iteration", 0)

    if iteration >= state["max_iterations"]:
        return {"next_worker": "done", "iteration": iteration + 1}

    if iteration == 0:
        return {"next_worker": "researcher", "iteration": 1}
    elif iteration == 1:
        return {"next_worker": "analyst", "iteration": 2}
    return {"next_worker": "done", "iteration": iteration + 1}


def researcher(state: LimitedState) -> dict:
    return {"research": f"Research on '{state['query']}': 5 sources"}


def analyst(state: LimitedState) -> dict:
    return {"analysis": f"Analysis of: {state['research'][:30]}..."}


def route(state: LimitedState) -> str:
    return state["next_worker"]


builder = StateGraph(LimitedState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "researcher": "researcher",
    "analyst": "analyst",
    "done": END,
})
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")

graph = builder.compile()

result = graph.invoke({
    "query": "RAG 2025", "research": "", "analysis": "",
    "next_worker": "", "iteration": 0, "max_iterations": 2,
})

assert result["research"] != "" and result["analysis"] == ""
print(f"Research: {result['research']}")
print(f"Analysis empty: {result['analysis'] == ''}")
print("✅ Iteration limit: only research completed, the analyst never ran")
# Expected output:
# Research: Research on 'RAG 2025': 5 sources
# Analysis empty: True
# ✅ Iteration limit: only research completed, the analyst never ran

Explanation: With max_iterations=2, the supervisor runs the researcher (iteration 1) and then, on reaching iteration 2, hits the limit and stops. The analyst never runs. This prevents infinite loops and keeps costs under control.

Exercise 3: Supervisor with collect → validate → format and retry (Medium)

Build a system with 3 workers: data_collector, validator, and formatter. The supervisor delegates in order. If the validator rejects the data (the first call returns partial data, the second returns complete data), the supervisor goes back to data_collector. Maximum 2 retries.

See solution
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class PipelineState(TypedDict):
    source: str
    raw_data: str
    is_valid: bool
    formatted: str
    next_worker: str
    phase: str
    retries: int


def supervisor(state: PipelineState) -> dict:
    phase = state.get("phase", "start")
    retries = state.get("retries", 0)
    if phase == "start":
        return {"next_worker": "data_collector", "phase": "collecting"}
    if phase == "collecting":
        return {"next_worker": "validator", "phase": "validating"}
    if phase == "validating":
        if state.get("is_valid"):
            return {"next_worker": "formatter", "phase": "formatting"}
        if retries < 2:
            return {"next_worker": "data_collector", "phase": "collecting", "retries": retries + 1}
        return {"next_worker": "formatter", "phase": "formatting"}
    return {"next_worker": "done", "phase": "complete"}


def data_collector(state: PipelineState) -> dict:
    if state.get("retries", 0) == 0:
        # First attempt: the data comes back half-baked, on purpose
        return {"raw_data": "partial_data_missing"}
    return {"raw_data": f"complete_data_from_{state['source']}"}


def validator(state: PipelineState) -> dict:
    # The sentinel: only the complete value carries "complete" in its name.
    # Watch out here — if the partial value contained "complete" as a substring,
    # the check would pass by accident and the retry would never happen.
    return {"is_valid": "complete" in state.get("raw_data", "")}


def formatter(state: PipelineState) -> dict:
    prefix = "✅ VALID" if state.get("is_valid") else "⚠️ PARTIAL"
    return {"formatted": f"{prefix}: {state['raw_data']}"}


def route(state: PipelineState) -> str:
    return state["next_worker"]


builder = StateGraph(PipelineState)
builder.add_node("supervisor", supervisor)
builder.add_node("data_collector", data_collector)
builder.add_node("validator", validator)
builder.add_node("formatter", formatter)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "data_collector": "data_collector", "validator": "validator",
    "formatter": "formatter", "done": END,
})
for w in ["data_collector", "validator", "formatter"]:
    builder.add_edge(w, "supervisor")

graph = builder.compile()
result = graph.invoke({
    "source": "sales_API", "raw_data": "", "is_valid": False,
    "formatted": "", "next_worker": "", "phase": "start", "retries": 0,
})

print(f"Valid: {result['is_valid']} | Retries: {result['retries']}")
print(f"Result: {result['formatted']}")
# Expected output:
# Valid: True | Retries: 1
# Result: ✅ VALID: complete_data_from_sales_API

Explanation: First attempt: incomplete data → the validator rejects it → the supervisor retries. Second attempt: complete data → the validator approves → the formatter produces the output. The supervisor handled the retry on its own.

Exercise 4: Supervisor with create_supervisor (Medium)

Use create_supervisor to build a system with 2 workers: a researcher that searches for information and a critic that evaluates quality. The supervisor should coordinate them automatically. You'll need an API key.

See solution
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph_supervisor import create_supervisor


@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return (
        f"Results for '{query}': "
        "1) RAG improves precision by 40%, "
        "2) Hybrid search beats dense retrieval, "
        "3) Fine-tuning complements RAG in specific domains."
    )


@tool
def evaluate_quality(text: str) -> str:
    """Evaluate the quality and completeness of a text."""
    word_count = len(text.split())
    if word_count < 10:
        return "QUALITY: Low. Needs more detail and sources."
    return "QUALITY: Acceptable. Enough information for a summary."


researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[search_web],
    name="researcher",
    prompt="You are a researcher. Find complete information about the requested topic.",
)

critic = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[evaluate_quality],
    name="critic",
    prompt="You are a quality critic. Evaluate whether the information gathered is sufficient.",
)

workflow = create_supervisor(
    [researcher, critic],
    model=init_chat_model("openai:gpt-4.1"),
    prompt=(
        "You are a research supervisor. "
        "First delegate to the researcher to find information. "
        "Then delegate to the critic to evaluate the quality. "
        "If the quality is low, ask the researcher to search for more. "
        "If the quality is acceptable, produce a final summary."
    ),
)

app = workflow.compile()

result = app.invoke({
    "messages": [("user", "What's the impact of RAG on AI applications?")]
})

print(result["messages"][-1].content[:120] + "...")
# Expected output (varies by LLM):
# Based on the research and the quality review: RAG improves precision by 40%...

Explanation: create_supervisor automatically creates the handoff tools (transfer_to_researcher, transfer_to_critic). The supervisor's LLM decides who to delegate to based on context. The flow is automatic: you never write routing by hand.

Exercise 5: Supervisor with conditional retry (Advanced)

Build a manual supervisor that coordinates fetcher and validator. The fetcher returns incomplete data the first 2 times (simulate it with a global counter). The validator checks whether the data is complete. If it isn't, the supervisor re-delegates to the fetcher. Maximum 3 attempts.

See solution
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

fetch_call_count = 0


class RetryState(TypedDict):
    data: str
    is_complete: bool
    attempt: int
    max_attempts: int
    next_worker: str
    phase: str


def supervisor(state: RetryState) -> dict:
    phase = state.get("phase", "init")
    attempt = state.get("attempt", 0)

    if phase == "init":
        return {"next_worker": "fetcher", "phase": "fetching", "attempt": 1}
    if phase == "fetching":
        return {"next_worker": "validator", "phase": "validating"}
    if phase == "validating":
        if state["is_complete"]:
            return {"next_worker": "done", "phase": "complete"}
        if attempt < state["max_attempts"]:
            return {"next_worker": "fetcher", "phase": "fetching", "attempt": attempt + 1}
        return {"next_worker": "done", "phase": "complete_partial"}
    return {"next_worker": "done", "phase": "error"}


def fetcher(state: RetryState) -> dict:
    global fetch_call_count
    fetch_call_count += 1
    if fetch_call_count < 3:
        # The first 2 calls come back with half the data
        return {"data": "partial_data"}
    return {"data": "complete_data_verified"}


def validator(state: RetryState) -> dict:
    # "partial_data" does not contain "complete" → the retry fires
    return {"is_complete": "complete" in state.get("data", "")}


def route(state: RetryState) -> str:
    return state["next_worker"]


builder = StateGraph(RetryState)
builder.add_node("supervisor", supervisor)
builder.add_node("fetcher", fetcher)
builder.add_node("validator", validator)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "fetcher": "fetcher", "validator": "validator", "done": END,
})
builder.add_edge("fetcher", "supervisor")
builder.add_edge("validator", "supervisor")

graph = builder.compile()
fetch_call_count = 0

result = graph.invoke({
    "data": "", "is_complete": False, "attempt": 0,
    "max_attempts": 3, "next_worker": "", "phase": "init",
})

print(f"Data: {result['data']}")
print(f"Complete?: {result['is_complete']}")
print(f"Attempts: {result['attempt']}")
print(f"Calls to fetcher: {fetch_call_count}")
# Expected output:
# Data: complete_data_verified
# Complete?: True
# Attempts: 3
# Calls to fetcher: 3

Explanation: The supervisor implements a retry loop: fetch → validate → (if incomplete) fetch → validate → ... until the data is complete or the attempts run out. Three calls to the fetcher: the first 2 return partial data, the third returns complete data.


Summary

  • The supervisor is a coordinator, not an executor. Its job is to decide which worker needs to run, not to do the work. Think of a project manager: they know who to assign each task to, but they don't need to know how to code
  • A manual supervisor with StateGraph is the starting point: rule-based routing (if/else), no extra LLM cost, trivial debugging, predictable flow. Ideal for pipelines with a fixed flow
  • create_supervisor is the idiomatic form for LLM routing: it creates handoff tools automatically, the model decides who to delegate to based on context, and it adapts to unexpected queries. Ideal for varied queries
  • Start simple, scale if you need to. Most production systems start with rule-based routing and only add LLM routing when the use cases demand it
  • Model optimization per worker is a concrete advantage: a cheap model for search, a powerful model for analysis. Same result, lower cost
  • Stop conditions prevent infinite loops: an iteration limit as a safety net, a quality check as a termination criterion, LLM decision as the default in create_supervisor
  • The name is mandatory in multi-agent. It identifies each worker in logs, handoffs, and shared state

Next capsule: Pattern Handoffs — you'll learn how one agent transfers control directly to another with no central supervisor, when to use handoffs vs a supervisor, and how to build chains of specialists.


Further reading

  1. LangGraph — Multi-Agent Supervisor — Official docs on the supervisor pattern with examples and variations
  2. create_supervisor API Reference — Complete reference for create_supervisor: parameters, options, and examples
  3. langgraph-supervisor GitHub — Official repo with source code and advanced examples
  4. How to build a multi-agent supervisor — A practical step-by-step guide
  5. Multi-Agent Architectures — LangChain Blog — Comparison of multi-agent patterns: supervisor, handoffs, hierarchical

Module 10 — LangChain & LangGraph: From Chains to Agents