--- title: "Building a Multi-Agent System with LangGraph: Supervisor and Worker Patterns" description: "Build multi-agent systems in LangGraph using subgraph composition, supervisor routing, and parallel worker execution to create specialized agent teams that collaborate on complex tasks." canonical: https://callsphere.ai/blog/langgraph-multi-agent-system-supervisor-worker-patterns category: "Learn Agentic AI" tags: ["LangGraph", "Multi-Agent", "Supervisor Pattern", "Subgraphs", "Python"] author: "CallSphere Team" published: 2026-03-17T00:00:00.000Z updated: 2026-05-06T01:02:43.503Z --- # Building a Multi-Agent System with LangGraph: Supervisor and Worker Patterns > Build multi-agent systems in LangGraph using subgraph composition, supervisor routing, and parallel worker execution to create specialized agent teams that collaborate on complex tasks. ## 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: ```mermaid flowchart TD USER(["User input"]) SUPER["Supervisor node routes by state"] A["Specialist node A research"] B["Specialist node B writing"] TOOL{"Tool call needed?"} EXEC["Tool executor ToolNode"] CHK[("Postgres checkpointer")] INT{"interrupt for human approval?"} HUMAN(["Human reviewer"]) OUT(["Final response"]) USER --> SUPER SUPER --> A SUPER --> B A --> TOOL B --> TOOL TOOL -->|Yes| EXEC --> SUPER TOOL -->|No| INT INT -->|Yes| HUMAN --> SUPER INT -->|No| OUT SUPER CHK style SUPER fill:#4f46e5,stroke:#4338ca,color:#fff style CHK fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b style OUT fill:#059669,stroke:#047857,color:#fff style HUMAN fill:#f59e0b,stroke:#d97706,color:#1f2937 ``` ```python 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: ```python 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: ```python 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: ```python 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: ```python 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 ```python 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 --- Source: https://callsphere.ai/blog/langgraph-multi-agent-system-supervisor-worker-patterns