Building a Multi-Agent System with LangGraph: Supervisor and Worker Patterns

When Single Agents Are Not Enough

A single agent with many tools quickly hits a ceiling. As you add more tools, the LLM becomes less reliable at selecting the right one. The system prompt grows unwieldy. Different tasks require different model configurations or temperature settings. Multi-agent systems solve this by decomposing complex workflows into specialized agents, each focused on a narrow domain, coordinated by a supervisor.

The Supervisor Pattern

In the supervisor pattern, one agent acts as a router that decides which specialized worker agent should handle each step:

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

class TeamState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str

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

def supervisor(state: TeamState) -> dict:
    system = SystemMessage(content="""You are a supervisor routing tasks.
    Based on the user request, decide which worker to invoke:
    - 'researcher' for information gathering
    - 'writer' for content creation
    - 'coder' for code generation
    - 'FINISH' if the task is complete
    Respond with ONLY the worker name.""")

    response = supervisor_llm.invoke(
        [system] + state["messages"]
    )
    return {"next_agent": response.content.strip().lower()}

Worker Agents

Each worker is a focused agent with its own system prompt and tools:

researcher_llm = ChatOpenAI(model="gpt-4o-mini")
writer_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
coder_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def researcher(state: TeamState) -> dict:
    system = SystemMessage(
        content="You are a research assistant. Find and summarize information."
    )
    response = researcher_llm.invoke(
        [system] + state["messages"]
    )
    return {"messages": [response]}

def writer(state: TeamState) -> dict:
    system = SystemMessage(
        content="You are a content writer. Create polished, well-structured text."
    )
    response = writer_llm.invoke(
        [system] + state["messages"]
    )
    return {"messages": [response]}

def coder(state: TeamState) -> dict:
    system = SystemMessage(
        content="You are a Python developer. Write clean, tested code."
    )
    response = coder_llm.invoke(
        [system] + state["messages"]
    )
    return {"messages": [response]}

Assembling the Supervisor Graph

Connect the supervisor to workers with conditional routing:

def route_to_worker(state: TeamState) -> Literal[
    "researcher", "writer", "coder", "__end__"
]:
    next_agent = state["next_agent"]
    if next_agent == "finish":
        return "__end__"
    return next_agent

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_node("coder", coder)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_to_worker)

# All workers route back to supervisor after completing
builder.add_edge("researcher", "supervisor")
builder.add_edge("writer", "supervisor")
builder.add_edge("coder", "supervisor")

graph = builder.compile()

The supervisor evaluates each response and decides whether to hand off to another worker or finish. This creates a loop where the supervisor orchestrates a multi-step collaboration.

Subgraph Composition

For complex workers that are themselves multi-step graphs, use subgraph composition:

def build_research_subgraph() -> StateGraph:
    """Build a research agent with search and analysis steps."""

    class ResearchState(TypedDict):
        messages: Annotated[list, add_messages]

    def search(state: ResearchState) -> dict:
        # Perform web search
        return {"messages": [{"role": "assistant", "content": "Search results..."}]}

    def analyze(state: ResearchState) -> dict:
        # Analyze search results
        return {"messages": [{"role": "assistant", "content": "Analysis..."}]}

    sub = StateGraph(ResearchState)
    sub.add_node("search", search)
    sub.add_node("analyze", analyze)
    sub.add_edge(START, "search")
    sub.add_edge("search", "analyze")
    sub.add_edge("analyze", END)
    return sub.compile()

research_graph = build_research_subgraph()

# Use the subgraph as a node in the parent graph
builder.add_node("researcher", research_graph)

The parent graph treats the subgraph as a single node. State flows in, the subgraph processes it through its own internal nodes, and the final state flows back to the parent.

Parallel Worker Execution

LangGraph supports sending work to multiple nodes simultaneously:

from langgraph.graph import Send

def fan_out(state: TeamState) -> list[Send]:
    """Send the task to multiple workers in parallel."""
    return [
        Send("researcher", state),
        Send("writer", state),
    ]

builder.add_conditional_edges("supervisor", fan_out)

The Send object directs execution to a specific node with a given state. Returning multiple Send objects causes parallel execution, and the results are merged using the state reducers.

Putting It All Together

result = graph.invoke({
    "messages": [HumanMessage(
        content="Research the latest trends in AI agents, "
        "then write a blog post about the findings."
    )],
    "next_agent": "",
})

# The supervisor coordinates: researcher gathers info, writer creates content
for msg in result["messages"]:
    print(f"{msg.__class__.__name__}: {msg.content[:80]}...")

The supervisor first routes to the researcher, then after receiving the research results, routes to the writer to produce the final output.

FAQ

How many worker agents can a supervisor manage?

There is no hard limit, but LLM-based routers become less reliable with more than 8-10 options. For larger systems, use a hierarchical pattern with multiple supervisors, each managing a team of 3-5 specialists.

Can worker agents communicate directly with each other?

In the standard supervisor pattern, workers communicate through the shared state — they read each other's outputs from the message history. Direct agent-to-agent communication is possible by having workers write to specific state channels that other workers read from.

How do I handle a worker that gets stuck in a loop?

Add loop counters to state and check them in the supervisor. If a worker has been called more than N times without progress, the supervisor should either try a different worker or terminate with a partial result.

---

#LangGraph #MultiAgent #SupervisorPattern #Subgraphs #Python #AgenticAI #LearnAI #AIEngineering