LangGraph Multi-Agent Supervisor: Building Agents That Delegate

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)
 

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()
 

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}
 

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.