The Supervisor Pattern: How to Orchestrate Multi-Agent Teams with LangGraph

Why Single God Agents Fail at Scale

The god agent pattern — one agent, all tools, all tasks — seems elegant. In practice, it fails at scale for three structural reasons:

1. Context Window Overload

Every tool you give an agent consumes context window space at every reasoning step. An agent with 20 tools might burn 6,000–10,000 tokens on tool schema definitions alone. For a 128K context model, that is 5–8% of the window consumed before the task even begins. For complex, multi-step tasks that accumulate history, this constraint becomes severe.

2. Unclear Failure Attribution

When a god agent produces a poor result, what went wrong? Did the security scan miss the SQL injection? Did the performance analysis produce incorrect metrics? Did the documentation scorer fail to run at all? You cannot tell — everything happened inside one opaque reasoning chain. With the supervisor pattern, each worker's output is a discrete artefact. When the security worker misses something, you know exactly where to investigate and improve.

3. Inability to Parallelise

A sequential god agent processes tasks one at a time. A supervisor with parallel workers can run security scanning, performance analysis, logic review, and test coverage measurement simultaneously — reducing total execution time from the sum of all analyses to the maximum of the longest one. For code review, this typically means a 60–70% reduction in wall-clock time.

Supervisor Topologies: Hub-and-Spoke vs Hierarchical

The supervisor pattern has two main topologies, and choosing the right one matters for systems at scale:

Hub-and-Spoke: One supervisor routes to N worker agents. All workers report back to the supervisor. Simple, easy to reason about, works well for up to 6 workers. The supervisor's context grows with the number of worker result summaries, so 6+ workers can create a synthesis bottleneck.

Hierarchical: A top-level supervisor routes to mid-level supervisors, each of which routes to their own worker pool. Used when workers naturally cluster into functional groups (e.g., a "security team" supervisor managing 3 security workers, and a "performance team" supervisor managing 3 performance workers). Scales to 15+ workers. Adds complexity — two levels of orchestration logic to maintain.

For most enterprise use cases, start with hub-and-spoke. Move to hierarchical when worker count exceeds 6 or when workers in the same functional domain need shared context that would pollute the top-level supervisor's state.

Full LangGraph Implementation: Supervisor + 3 Workers

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

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# --- Pydantic schemas for worker outputs ---
# WHY Pydantic: forces workers to return structured output.
# Empty dict {} is valid Python but invalid WorkerResult.
# This is the fix for worker cascade failure (see failure mode section).

class WorkerResult(BaseModel):
    worker: str = Field(description="Name of the worker that produced this result")
    findings: list[str] = Field(description="List of findings — must have at least 1 item")
    severity: Literal["low", "medium", "high", "critical"] = Field(default="low")
    error: Optional[str] = Field(default=None, description="Set if worker failed")

class SupervisorState(TypedDict):
    task: str                                              # initial task
    security_result: Optional[WorkerResult]
    performance_result: Optional[WorkerResult]
    logic_result: Optional[WorkerResult]
    synthesis: Optional[str]                               # final output
    errors: Annotated[list[str], operator.add]             # accumulated worker errors

# --- Supervisor: decides which workers to call and synthesises ---
def supervisor_node(state: SupervisorState) -> dict:
    """
    Supervisor receives the task and decides worker routing.
    In this implementation all workers run (parallel fan-out).
    In dynamic routing: use LLM with tool-calling to select workers.
    """
    # For this example: route to all workers simultaneously
    # In a dynamic supervisor: call LLM to decide routing
    return {}  # State unchanged — supervisor just fans out

def security_worker(state: SupervisorState) -> dict:
    """
    Security specialist: analyses code for OWASP vulnerabilities,
    injection risks, authentication flaws, and secrets exposure.
    Context: only the task + security-specific tools. No performance data.
    """
    response = llm.invoke([
        SystemMessage(content="""You are a security code reviewer specialising in OWASP Top 10.
        Analyse the provided code for: SQL injection, XSS, authentication weaknesses,
        hardcoded secrets, insecure dependencies, and data exposure risks.
        Return a JSON object with fields: worker, findings (list), severity."""),
        HumanMessage(content=f"Review for security vulnerabilities:\n{state['task']}")
    ])

    try:
        import json
        data = json.loads(response.content)
        result = WorkerResult(worker="security", **data)
    except Exception as e:
        result = WorkerResult(
            worker="security",
            findings=["Security analysis failed — manual review required"],
            severity="high",
            error=str(e)
        )

    return {"security_result": result}

def performance_worker(state: SupervisorState) -> dict:
    """
    Performance specialist: analyses algorithmic complexity, N+1 queries,
    memory leaks, and unnecessary blocking operations.
    Context: only the task + performance-specific patterns. No security data.
    """
    response = llm.invoke([
        SystemMessage(content="""You are a performance code reviewer.
        Analyse for: O(n²) or worse algorithms, N+1 database queries,
        blocking I/O in async contexts, memory leaks, and missing indexes.
        Return JSON: worker, findings (list), severity."""),
        HumanMessage(content=f"Review for performance issues:\n{state['task']}")
    ])

    try:
        import json
        data = json.loads(response.content)
        result = WorkerResult(worker="performance", **data)
    except Exception as e:
        result = WorkerResult(
            worker="performance",
            findings=["Performance analysis failed — manual review required"],
            severity="medium",
            error=str(e)
        )

    return {"performance_result": result}

def logic_worker(state: SupervisorState) -> dict:
    """
    Logic specialist: analyses correctness, edge case handling,
    error propagation, and business rule compliance.
    """
    response = llm.invoke([
        SystemMessage(content="""You are a logic and correctness code reviewer.
        Analyse for: null pointer dereferences, off-by-one errors, missing error handling,
        incorrect business logic, and untested edge cases.
        Return JSON: worker, findings (list), severity."""),
        HumanMessage(content=f"Review for logic correctness:\n{state['task']}")
    ])

    try:
        import json
        data = json.loads(response.content)
        result = WorkerResult(worker="logic", **data)
    except Exception as e:
        result = WorkerResult(
            worker="logic",
            findings=["Logic analysis failed — manual review required"],
            severity="medium",
            error=str(e)
        )

    return {"logic_result": result}

def synthesise_results(state: SupervisorState) -> dict:
    """
    Supervisor synthesises all worker outputs into a unified review.
    WHY separate synthesis node: gives the supervisor fresh context for
    synthesis rather than accumulating all worker reasoning in one chain.
    """
    # Validate all workers completed with results
    workers = {
        "security": state.get("security_result"),
        "performance": state.get("performance_result"),
        "logic": state.get("logic_result"),
    }

    errors = []
    summary_parts = []

    for worker_name, result in workers.items():
        if result is None:
            errors.append(f"Worker '{worker_name}' returned no result — validation gate blocked it")
            summary_parts.append(f"## {worker_name.title()} Review\nFailed — manual review required")
        elif result.error:
            errors.append(f"Worker '{worker_name}' reported error: {result.error}")
            summary_parts.append(f"## {worker_name.title()} Review\nPartial: {'; '.join(result.findings)}")
        else:
            findings_text = "\n".join(f"- [{result.severity.upper()}] {f}" for f in result.findings)
            summary_parts.append(f"## {worker_name.title()} Review\n{findings_text}")

    synthesis = "# Code Review Report\n\n" + "\n\n".join(summary_parts)

    if errors:
        synthesis += f"\n\n## ⚠️ Worker Errors\n" + "\n".join(f"- {e}" for e in errors)

    return {"synthesis": synthesis, "errors": errors}

# --- Build the supervisor graph ---
builder = StateGraph(SupervisorState)

# Nodes
builder.add_node("supervisor", supervisor_node)
builder.add_node("security", security_worker)
builder.add_node("performance", performance_worker)
builder.add_node("logic", logic_worker)
builder.add_node("synthesise", synthesise_results)

# Supervisor fans out to all workers simultaneously
builder.add_edge("__start__", "supervisor")
builder.add_edge("supervisor", "security")
builder.add_edge("supervisor", "performance")
builder.add_edge("supervisor", "logic")

# All workers converge to synthesis
builder.add_edge("security", "synthesise")
builder.add_edge("performance", "synthesise")
builder.add_edge("logic", "synthesise")
builder.add_edge("synthesise", END)

graph = builder.compile()

# Run a code review
result = graph.invoke({
    "task": "def get_user(user_id):\n    return db.execute(f'SELECT * FROM users WHERE id={user_id}')",
    "security_result": None, "performance_result": None, "logic_result": None,
    "synthesis": None, "errors": []
})

print(result["synthesis"])

Production Failure: Worker Cascade Failure

The worker cascade failure is the most common production bug I see in supervisor pattern implementations. Here is exactly how it happens and why Pydantic validation is the fix.

The bug: A worker agent encounters an LLM API timeout or produces malformed JSON. Rather than raising an exception, it catches the error and returns an empty dict {}. In Python, returning {} from a LangGraph node is valid — it means "no state updates." The supervisor's state field for that worker remains None. The synthesis node runs, finds the field is None, and either crashes or — worse — silently skips that worker's findings. The final report says "no security issues found" because the security worker's result was None, not because the code is actually secure.

# BROKEN: The naive worker implementation that causes cascade failure
def broken_security_worker(state: SupervisorState) -> dict:
    try:
        response = llm.invoke([...])
        data = json.loads(response.content)
        return {"security_result": data}  # What if data is {} or malformed?
    except Exception:
        return {}  # BUG: empty dict means "no state update"
                   # supervisor.security_result stays None
                   # synthesis sees None and skips security findings
                   # final report says "no security issues" — WRONG

# FIXED: Always return a structured WorkerResult, even on failure
def fixed_security_worker(state: SupervisorState) -> dict:
    try:
        response = llm.invoke([...])
        data = json.loads(response.content)
        # Pydantic validation: if data is missing required fields, raises ValidationError
        result = WorkerResult(worker="security", **data)
    except json.JSONDecodeError as e:
        # Structured failure result — supervisor knows this worker errored
        result = WorkerResult(
            worker="security",
            findings=["JSON parse failed — security review incomplete"],
            severity="high",
            error=f"JSON decode error: {str(e)}"
        )
    except Exception as e:
        result = WorkerResult(
            worker="security",
            findings=["Security worker exception — manual review required"],
            severity="critical",
            error=str(e)
        )

    # Always return a result — never return {}
    return {"security_result": result}

# The synthesis node can now distinguish:
# result is None → worker never ran (graph routing bug)
# result.error is set → worker ran but failed (handled gracefully)
# result.findings is populated → worker succeeded

After implementing Pydantic validation at every worker boundary, the client's code review system went from 3 silent synthesis failures per week (where missing security findings were treated as "no issues found") to zero. The fix is not complex — it is discipline: every worker boundary must have a schema, and failure must produce a structured error result, never an empty return.

Pattern Comparison: When to Use Each

Conclusion: The Supervisor Pattern Is Not Optional at Scale

If you are building AI agents for tasks that require multiple types of expertise — security + performance + logic review; research + drafting + fact-checking; data extraction + analysis + narrative — you will hit the limits of the god agent pattern. Not in the demo. In production, at scale, under adversarial conditions.

The supervisor pattern, implemented correctly with Pydantic output validation at every worker boundary, is the architecture that makes multi-domain AI reliable. It adds a day of implementation work upfront and saves weeks of debugging production incidents down the line. In every client engagement where I've introduced it, the engineering team's reaction at the three-month mark is the same: "I can't believe we tried to do this with a single agent."

In Day 3 of the Agentic AI Workshop, we build a complete supervisor + workers system from scratch — including deliberate worker failure injection so participants can see the cascade failure in action and implement the Pydantic fix themselves. Understanding the failure mode is what makes the pattern stick.