How to implement the supervisor pattern in LangGraph — routing, subgraphs, and shared state
When a single agent is not enough
A single ReAct agent with many tools works well up to a point. Beyond that point — too many tools, too many responsibilities, too many steps — it starts making mistakes. The supervisor pattern is the most practical solution: a lightweight routing agent coordinates a team of specialist agents, each focused on a narrow task.
Two ways to implement supervisor in LangGraph
LangGraph supports two approaches:
- langgraph-supervisor package: a thin convenience wrapper that builds the graph for you — best for straightforward setups
- Manual implementation: build the supervisor as a regular node with conditional edges — necessary for custom routing logic
This guide covers both.
Option 1: langgraph-supervisor package
pip install langgraph-supervisor
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
llm = ChatOpenAI(model='gpt-4o')
# --- Specialist agents ---
@tool
def web_search(query: str) -> str:
"""Search the web for current information."""
return f'Search results for: {query}'
@tool
def run_python(code: str) -> str:
"""Execute Python code and return the output."""
try:
exec_globals = {}
exec(code, exec_globals)
return str(exec_globals.get('result', 'Code executed.'))
except Exception as e:
return f'Error: {e}'
research_agent = create_react_agent(
model=llm,
tools=[web_search],
name='research_agent',
prompt='You are a research specialist. Use web_search to find information.'
)
code_agent = create_react_agent(
model=llm,
tools=[run_python],
name='code_agent',
prompt='You are a Python specialist. Write and execute code to solve problems.'
)
# --- Supervisor ---
supervisor = create_supervisor(
agents=[research_agent, code_agent],
model=llm,
prompt=(
'You are a supervisor managing a research agent and a code agent. '
'Route tasks to the appropriate specialist. '
'For information retrieval, use research_agent. '
'For calculations or data processing, use code_agent.'
),
).compile()
# Run it
result = supervisor.invoke({
'messages': [{'role': 'user', 'content': 'What is the population of Tokyo, and what is that squared?'}]
})
print(result['messages'][-1].content)
Option 2: Manual supervisor graph
When you need custom routing logic (e.g. confidence-based routing, priority queues, or complex state), build the supervisor manually.
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model='gpt-4o')
class SupervisorState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
next_agent: str
research_output: str
code_output: str
# Supervisor node: decides who to call next
def supervisor_node(state: SupervisorState) -> SupervisorState:
system = (
'You are a supervisor. Given the conversation, decide which agent should act next.\n'
'Reply with ONLY ONE of: research_agent, code_agent, FINISH'
)
msgs = [{'role': 'system', 'content': system}] + state['messages']
response = llm.invoke(msgs)
next_agent = response.content.strip()
return {'next_agent': next_agent}
# Route based on supervisor decision
def route(state: SupervisorState) -> Literal['research_agent', 'code_agent', END]:
return state['next_agent'] if state['next_agent'] != 'FINISH' else END
# Specialist nodes
def research_node(state: SupervisorState) -> SupervisorState:
result = research_agent.invoke(state) # subgraph call
return {'messages': result['messages'], 'research_output': result['messages'][-1].content}
def code_node(state: SupervisorState) -> SupervisorState:
result = code_agent.invoke(state)
return {'messages': result['messages'], 'code_output': result['messages'][-1].content}
# Build the graph
builder = StateGraph(SupervisorState)
builder.add_node('supervisor', supervisor_node)
builder.add_node('research_agent', research_node)
builder.add_node('code_agent', code_node)
builder.add_edge(START, 'supervisor')
builder.add_conditional_edges('supervisor', route)
builder.add_edge('research_agent', 'supervisor')
builder.add_edge('code_agent', 'supervisor')
graph = builder.compile()
Shared state vs message passing
| Approach | How it works | Best for |
|---|---|---|
| Shared TypedDict state | All agents read/write the same state object | Pipelines where agents build on each other's outputs |
| Message passing (add_messages) | Agents append to a messages list | Conversational multi-agent flows |
| Subgraph with private state | Each subgraph has its own state, results surface via output mapping | Truly independent agents with different state shapes |
Preventing infinite loops
Without a termination condition, a supervisor can loop indefinitely — agent A outputs to supervisor, supervisor routes back to agent A, repeat.from langgraph.graph import StateGraph
# Add a step counter to prevent infinite loops
class SafeSupervisorState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
next_agent: str
steps: int # track iterations
MAX_STEPS = 10
def safe_route(state: SafeSupervisorState):
if state['steps'] >= MAX_STEPS:
return END # force termination
return state['next_agent'] if state['next_agent'] != 'FINISH' else END
def supervisor_with_counter(state: SafeSupervisorState):
# ... supervisor logic ...
return {'next_agent': 'research_agent', 'steps': state.get('steps', 0) + 1}
When to use supervisor vs other patterns
| Pattern | Use when |
|---|---|
| Supervisor | Tasks require dynamic routing to 2-5 specialists |
| Sequential (pipeline) | Steps are always in the same fixed order |
| Parallel (map-reduce) | Independent subtasks can run concurrently and merge |
| Single ReAct agent | One agent with 3-5 tools handles everything fine |
| Hierarchical (supervisor of supervisors) | Very large systems with 10+ agents across domains |
Start with a single ReAct agent. Add a supervisor layer only when you have clear evidence that the agent is struggling because of tool overload or responsibility conflicts, not preemptively.