Multi-Agent Swarms: Supervisor Pattern, Subgraphs, and Handoff Protocols

When an engineer tries to make a single agent do everything—conduct research, write Python code, execute database migrations, verify security compliance, and generate marketing copy—the system inevitably deteriorates.

A single model context cluttered with 40 tool definitions and 10 pages of disparate instructions suffers from instruction dilution and severe tool hallucination.

The solution is Multi-Agent Decomposition: dividing complex domains among specialized, narrowly focused agents that collaborate through structured communication protocols.

In this capstone post of Module 3: State Machines & Orchestration, we architect a production Multi-Agent Supervisor System in LangGraph.

1. Multi-Agent Topologies: Centralized vs. Decentralized

TOPOLOGY A: CENTRALIZED SUPERVISOR (Recommended for Enterprise)
                     [User Request]
                           │
                           ▼
                  ┌─────────────────┐
                  │ SUPERVISOR NODE │ ◄──────────────────┐
                  └────────┬────────┘                    │
           ┌───────────────┼───────────────┐             │
           ▼               ▼               ▼             │
     [Researcher]       [Coder]        [Reviewer]        │
           │               │               │             │
           └───────────────┴───────────────┴─────────────┘
                  (Handoff Results Back to Supervisor)

TOPOLOGY B: PEER-TO-PEER HANDOFF (Choreography)
  [Researcher] ──(Handoff)──> [Coder] ──(Handoff)──> [Reviewer]

Why the Supervisor Pattern Dominates:

• Centralized Accountability: One executive agent maintains the overarching goal and decides who works next.
• Context Isolation: The Coder agent does not need to see all 15 raw search pages parsed by the Researcher; the Supervisor passes only the curated research brief.
• Deterministic Termination: Prevents infinite conversational loops between peer agents agreeing with each other endlessly.

2. Defining Specialized Agent Nodes

In LangGraph, each worker agent is represented as a node (or nested subgraph) that processes the shared message stream and returns its domain contribution:

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

# Central Multi-Agent State
class TeamState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_worker: str

# Supervisor Router Schema
class RouteResponse(BaseModel):
    next_worker: Literal["Researcher", "Coder", "Reviewer", "FINISH"] = Field(
        ...,
        description="The next specialized agent to route to, or FINISH if the task is complete."
    )

3. Implementing the Supervisor Node

The supervisor examines the ongoing conversation and routes dynamically:

supervisor_llm = ChatOpenAI(model="gpt-4o", temperature=0.0)

SUPERVISOR_PROMPT = """
You are the Technical Lead Supervisor overseeing a specialized engineering team:
1. 'Researcher': Gathers technical documentation and architectural specifications.
2. 'Coder': Implements Python code solutions based on research.
3. 'Reviewer': Audits code for security vulnerabilities, syntax bugs, and performance flaws.

Given the current conversation, decide who acts next. If the solution is verified and complete, emit 'FINISH'.
"""

def supervisor_node(state: TeamState) -> dict:
    messages = [{"role": "system", "content": SUPERVISOR_PROMPT}] + list(state["messages"])

    response = supervisor_llm.with_structured_output(RouteResponse).invoke(messages)
    print(f"\n👑 [SUPERVISOR DECISION]: Routing task to -> {response.next_worker}")
    return {"next_worker": response.next_worker}

4. Worker Node Implementation

Each worker has a distinct system prompt and scoped toolset:

worker_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

def researcher_node(state: TeamState) -> dict:
    print("🔬 [RESEARCHER]: Investigating library specs...")
    prompt = "You are a Research Specialist. Synthesize factual specs for the query."
    response = worker_llm.invoke([{"role": "system", "content": prompt}] + list(state["messages"]))
    return {"messages": [AIMessage(content=f"[Researcher]: {response.content}")]}

def coder_node(state: TeamState) -> dict:
    print("💻 [CODER]: Implementing Python component...")
    prompt = "You are a Senior Python Coder. Generate clean, typed, idiomatic code."
    response = worker_llm.invoke([{"role": "system", "content": prompt}] + list(state["messages"]))
    return {"messages": [AIMessage(content=f"[Coder]: {response.content}")]}

def reviewer_node(state: TeamState) -> dict:
    print("🛡️ [REVIEWER]: Performing static code audit...")
    prompt = "You are a Security Reviewer. Audit the code for flaws. State 'VERIFIED' if passing."
    response = worker_llm.invoke([{"role": "system", "content": prompt}] + list(state["messages"]))
    return {"messages": [AIMessage(content=f"[Reviewer]: {response.content}")]}

5. Wiring the Multi-Agent Graph

Now we connect the team into a coordinated LangGraph workflow:

workflow = StateGraph(TeamState)

# Add all nodes
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Coder", coder_node)
workflow.add_node("Reviewer", reviewer_node)

# Set entry point
workflow.set_entry_point("supervisor")

# Conditional routing from supervisor to workers
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_worker"],
    {
        "Researcher": "Researcher",
        "Coder": "Coder",
        "Reviewer": "Reviewer",
        "FINISH": END
    }
)

# Every worker reports back to the supervisor upon finishing
workflow.add_edge("Researcher", "supervisor")
workflow.add_edge("Coder", "supervisor")
workflow.add_edge("Reviewer", "supervisor")

# Compile multi-agent app
team_app = workflow.compile()

6. Execution Trace

When given: "Build an asynchronous rate-limiter in Python using token bucket algorithm":
1. Supervisor routes to Researcher $\rightarrow$ Researcher returns mathematical token-bucket formulas.
2. Supervisor routes to Coder $\rightarrow$ Coder drafts AsyncTokenBucket class.
3. Supervisor routes to Reviewer $\rightarrow$ Reviewer identifies a race condition in locking mechanism.
4. Supervisor routes back to Coder $\rightarrow$ Coder applies asyncio.Lock() fix.
5. Supervisor routes to Reviewer $\rightarrow$ Reviewer emits VERIFIED.
6. Supervisor routes to FINISH $\rightarrow$ Workflow terminates successfully.

Summary of Module 3: State Machines & Orchestration

We have completed our deep dive into LangGraph orchestration:
- Post 08: The fundamental breakdown of linear chains and the necessity of cyclic state graphs.
- Post 09: LangGraph core primitives: state channels, custom reducers, and conditional routers.
- Post 10: Durable persistence and session time-travel with SQLite and PostgreSQL checkpointers.
- Post 11: Human-in-the-Loop breakpoints, manual approval queues, and mid-flight state editing.
- Post 12: Multi-agent swarms orchestrated via centralized Supervisor routing and subgraphs.

Now that we can architect sophisticated multi-agent graphs, the ultimate engineering question arises: How do we observe, test, evaluate, and secure them in production?

Welcome to Module 4: Productionization, Evals & Reliability [LangSmith], starting with [Post 13: Observability & Tracing with LangSmith]!