You have a single AI agent that can answer questions, use tools, and call APIs. It works. But then the requests get harder — you need it to search a database, analyze code in a repo, and write a summary, all in one session. One agent trying to do everything gets tangled. Its context fills up with irrelevant tool calls. It forgets what it was doing halfway through.
That is where multi-agent orchestration comes in. Instead of one agent that tries to do everything, you split the work across specialists with a supervisor that delegates tasks. LangGraph, the low-level orchestration framework from LangChain, is built for exactly this pattern.
You will build a supervisor agent that delegates to research, code, and review agents, using both the Graph API and the newer Functional API. All code is copy-paste ready.
Prerequisites
- Python 3.10+
- An OpenAI API key (or any LLM provider supported by LangChain)
- Basic familiarity with Python async
pip install -U langgraph langchain-openai
Version used in this tutorial: LangGraph 1.2.10 (requires Python 3.10+).
Step 1: The Agent Building Blocks
Every LangGraph agent follows the same loop: call an LLM, check if it wants to use a tool, execute the tool if needed, and repeat. Let us build that as a reusable component.
Create a new file called agent.py:
from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
# A simple tool
def get_weather(location: str) -> str:
"""Get the current weather in a given location."""
return f"Sunny, 72°F in {location}"
tools = [get_weather]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
# Node: call the LLM
def call_model(state: MessagesState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# Node: execute tool calls
def call_tool(state: MessagesState):
last_msg = state["messages"][-1]
results = []
for tc in last_msg.tool_calls:
if tc["name"] == "get_weather":
result = get_weather(**tc["args"])
results.append({
"role": "tool",
"content": result,
"tool_call_id": tc["id"],
})
return {"messages": results}
# Conditional edge: continue if tool call, end if not
def should_continue(state: MessagesState) -> Literal["tools", END]:
last_msg = state["messages"][-1]
if last_msg.tool_calls:
return "tools"
return END
# Build the graph
builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", call_tool)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")
graph = builder.compile()
# Run it
result = graph.invoke({
"messages": [HumanMessage(content="What is the weather in Jakarta?")]
})
print(result["messages"][-1].content)
Run this:
python agent.py
You should see a response about the weather in Jakarta. The agent loop is working.
Step 2: The Supervisor Pattern
One agent is fine for simple queries. But real workflows need multiple specialists. The supervisor pattern solves this: a supervisor agent examines each user request and routes it to the right specialist.
Create supervisor.py:
from typing import Literal, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
# Shared state
class AgentState(TypedDict):
messages: Sequence[BaseMessage]
next: str # which agent should handle next
members = ["researcher", "coder", "reviewer"]
options = members + ["FINISH"]
system_prompt = f"""
You are a supervisor managing the following workers: {members}.
Given the user request, decide which worker should act next.
Each worker will do their task and report back.
Respond with the worker name or "FINISH" when the task is complete.
"""
llm = ChatOpenAI(model="gpt-4o-mini")
# Supervisor node
def supervisor(state: AgentState):
response = llm.invoke([
SystemMessage(content=system_prompt),
] + state["messages"])
return {"next": response.content.strip()}
# Worker nodes (simplified)
def researcher(state: AgentState):
return {"messages": [
HumanMessage(content="I am researching: " + state["messages"][-1].content)
]}
def coder(state: AgentState):
return {"messages": [
HumanMessage(content="I am coding: " + state["messages"][-1].content)
]}
def reviewer(state: AgentState):
return {"messages": [
HumanMessage(content="I am reviewing: " + state["messages"][-1].content)
]}
# Build the workflow
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_node("reviewer", reviewer)
# All workers report back to supervisor
for member in members:
builder.add_edge(member, "supervisor")
# Supervisor decides next step
builder.add_conditional_edges(
"supervisor",
lambda state: state["next"],
{m: m for m in members} | {END: END}
)
builder.add_edge(START, "supervisor")
# Compile with memory so the supervisor remembers the conversation
graph = builder.compile(checkpointer=MemorySaver())
# Run
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({
"messages": [HumanMessage(content="Build a REST API that returns weather data from a SQLite database")],
"next": "",
}, config)
for msg in result["messages"]:
print(f"{msg.type}: {msg.content[:80]}...")
Step 3: Why the Supervisor Pattern Works
The supervisor pattern solves three problems that single-agent setups struggle with:
Context management. Each specialist only sees the parts of the conversation relevant to its job. The researcher does not need the coder's import statements in its context window. The supervisor keeps the big picture.
Tool isolation. Each specialist carries only the tools it needs. The coder does not need a web search tool. The researcher does not need a code execution sandbox. This reduces hallucinations, because the LLM cannot pick the wrong tool for the wrong job.
Escalation path. When a task hits something the specialist cannot handle, the supervisor can route to a different worker or loop back for clarification. A single agent with all tools would try to handle everything and produce a mediocre result.
Step 4: Specialized Sub-Agents with Real Tools
Let us give each specialist actual tools. The researcher will use web search (we use a mock for this example), the coder will write and review files, and the reviewer will check code quality.
Create specialized_agents.py:
from typing import Literal, Sequence, Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage
from langchain_openai import ChatOpenAI
# --- Tools ---
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Search results for '{query}': Found 3 relevant articles about best practices."
def write_file(filename: str, content: str) -> str:
"""Write content to a file."""
with open(f"/tmp/{filename}", "w") as f:
f.write(content)
return f"Written {len(content)} bytes to {filename}"
def read_file(filename: str) -> str:
"""Read content from a file."""
try:
with open(f"/tmp/{filename}") as f:
return f.read()
except FileNotFoundError:
return f"File {filename} not found."
def lint_code(code: str) -> str:
"""Check code for common issues."""
issues = []
if "import" not in code:
issues.append("No import statements found")
if len(code.split("\n")) < 5:
issues.append("Code is very short")
if not issues:
issues.append("No obvious issues found")
return "\n".join(issues)
# --- Agents ---
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
next: str
members = ["researcher", "coder", "reviewer"]
system_prompt = f"Route to one of: {members}. Say FINISH when done."
# Build each specialist agent
def make_agent(tools, system_message):
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: AgentState):
sys_msg = SystemMessage(content=system_message)
response = llm_with_tools.invoke([sys_msg] + list(state["messages"]))
tool_results = []
if hasattr(response, "tool_calls") and response.tool_calls:
for tc in response.tool_calls:
for tool in tools:
if tool.__name__ == tc["name"]:
result = tool(**tc["args"])
tool_results.append({
"role": "tool",
"content": str(result),
"tool_call_id": tc["id"],
})
return {"messages": [response] + tool_results}
return agent_node
researcher_agent = make_agent(
[search_web],
"You are a researcher. Search the web to find information. Report findings clearly."
)
coder_agent = make_agent(
[write_file, read_file],
"You are a Python developer. Write clean, working code. Read files when needed."
)
reviewer_agent = make_agent(
[lint_code],
"You are a code reviewer. Check code quality and suggest improvements."
)
# --- Supervisor with structured output ---
from langchain_core.pydantic_v1 import BaseModel, Field
class Router(BaseModel):
next: str = Field(description=f"One of: {options}")
llm_supervisor = ChatOpenAI(model="gpt-4o-mini")
supervisor_llm = llm_supervisor.with_structured_output(Router)
def supervisor(state: AgentState):
response = supervisor_llm.invoke([
SystemMessage(content=system_prompt),
] + list(state["messages"]))
return {"next": response.next}
# Build graph
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_node("reviewer", reviewer_agent)
for member in members:
builder.add_edge(member, "supervisor")
builder.add_conditional_edges(
"supervisor",
lambda s: s["next"],
{m: m for m in members} | {END: END}
)
builder.add_edge(START, "supervisor")
graph = builder.compile(checkpointer=MemorySaver())
This version uses structured output (Router model) for the supervisor decision instead of parsing raw text. This is more reliable in production because the output format is guaranteed by the LLM provider.
Step 5: The Functional API (Simpler Alternative)
LangGraph also offers a Functional API using @task and @entrypoint decorators. This is simpler for linear workflows or star-shaped orchestration where the supervisor runs before and after workers, not interleaved.
from langgraph.func import entrypoint, task
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
@task
def research(query: str) -> str:
"""Find relevant information."""
# In reality you would call a search API here
return f"Research findings for: {query}"
@task
def generate_code(spec: str) -> str:
"""Write code based on the spec and research."""
response = llm.invoke([
HumanMessage(content=f"Write Python code for: {spec}")
])
return response.content
@task
def review(code: str) -> str:
"""Review the generated code."""
response = llm.invoke([
HumanMessage(content=f"Review this code and suggest improvements: {code}")
])
return response.content
@entrypoint()
def build_feature(user_request: str):
findings = research(user_request).result()
code = generate_code(f"{user_request}\n\nResearch: {findings}").result()
feedback = review(code).result()
return {
"findings": findings,
"code": code,
"review": feedback,
}
# Run
result = build_feature.invoke("Build a REST API endpoint for user authentication")
print(result["code"])
The Functional API is cleaner when your workflow is a known sequence, not an open-ended loop. Use it when you know the execution order ahead of time. Use the Graph API when the agent needs to decide the next step based on results.
When to Use LangGraph vs Alternatives
| Tool | Best For | Less Good For |
|---|---|---|
| LangGraph | Stateful multi-agent; long-running workflows with human-in-the-loop; complex branching | Simple single-agent chat (overkill); quick prototypes where CrewAI's simplicity helps |
| CrewAI | Quick multi-agent prototypes; role-based agents with simple delegation | Fine-grained state control; complex conditional routing; production durability |
| AutoGen | Conversational agent-to-agent patterns; teams that negotiate and debate | Supervised orchestration; workflows that need persistence and checkpointing |
| OpenAI Assistants | Single-agent with built-in retrieval and code interpreter | Multi-agent orchestration; custom state management; framework-agnostic setups |
LangGraph's strength is control — every edge, every state transition, every persistence decision is explicit. The tradeoff is more code compared to CrewAI. What you get is the ability to trace exactly why an agent made a wrong call.
Common Issues
The supervisor always picks the same worker. Your system prompt needs clearer criteria. Add rules like "if the user asks about code, route to coder. If they ask about information, route to researcher. If they ask about quality, route to reviewer."
Agents lose context between turns. Use MemorySaver() as the checkpointer and pass the same thread_id in the config. This preserves the conversation across invocations.
Tool execution blocks the graph. Tools run synchronously by default. For long-running tasks (API calls, file processing), wrap them in asyncio or use the @task decorator to run them as concurrent tasks.
Structured output fails. Some models handle structured output better than others. gpt-4o-mini works well. gpt-3.5-turbo sometimes returns malformed JSON. Test your router with the model you plan to use in production.
Where to Go Next
- Add human-in-the-loop with LangGraph's
interruptmechanism to pause execution for approval - Use subgraphs to encapsulate each specialist agent as its own graph — this lets you test and deploy them independently
- Add persistence with PostgreSQL or SQLite checkpointer so agents survive server restarts
- Instrument with LangSmith to trace every decision and tool call across agents
References
- LangGraph Documentation — Official docs including quickstart, tutorials, and API reference
- LangGraph GitHub Repository — Source code, examples, and issue tracker
- LangGraph Quickstart Guide — Step-by-step guide with both Graph API and Functional API