From your first agent with create_agent, to a multi-agent supervisor in LangGraph with load balancing across workers.
0 / 10 completed
Level 1 · Foundations
Your first agent with create_agent
Goal: install LangChain and create your first agent with the create_agent function.
create_agent (from the langchain package) is today's recommended way to build a ReAct agent: it gives you a model capable of using tools and reasoning across multiple steps, with a flexible middleware system to extend it. Under the hood, it runs on LangGraph.
from langchain.agents import create_agent
agent = create_agent(
model="gpt-4o",
tools=[],
)
result = agent.invoke({
"messages": [{"role": "user", "content": "What is an AI agent, in one sentence?"}]
})
print(result["messages"][-1].content)
Every agent created with create_agent works with a message state (messages) — the same format used across LangGraph, which makes moving up to more complex graphs a natural step, not a rewrite.
Challenge
Swap model for a different provider (for example, an Anthropic model via langchain-anthropic) without changing anything else in the code.
Level 2 · Personality
System prompt and instructions
Goal: shape the agent's behavior with a system prompt and model parameters.
agent.py
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o", temperature=0.3, max_tokens=400)
agent = create_agent(
model=model,
tools=[],
system_prompt=(
"You are a technical support assistant. Reply in English, "
"briefly, with numbered steps. If you don't know something, say so."
),
)
result = agent.invoke({"messages": [{"role": "user", "content": "The app crashes when I open the camera"}]})
print(result["messages"][-1].content)
Passing a ChatOpenAI object instead of a string gives you fine-grained control over temperature, max_tokens, and other provider parameters.
Challenge
Write a system prompt that refuses to answer any question outside of a specific domain you choose.
Level 3 · Tools
Tools with @tool
Goal: create a custom tool with the @tool decorator from langchain_core.
agent.py
from langchain_core.tools import tool
from langchain.agents import create_agent
@tool
def current_weather(city: str) -> str:
"""Returns the current reported weather for a given city."""
data = {"Guatemala City": "72°F, partly cloudy"}
return data.get(city, "I don't have data for that city")
agent = create_agent(model="gpt-4o", tools=[current_weather])
result = agent.invoke({
"messages": [{"role": "user", "content": "What's the weather like in Guatemala City?"}]
})
print(result["messages"][-1].content)
The agent decides on its own when to call current_weather — the function's docstring is the description the model uses to make that decision, just like in most modern agent frameworks.
Challenge
Add a second tool, search_product(name: str), and check that the agent combines both in a single response when asked about both things at once.
Level 4 · Memory
Memory with a checkpointer
Goal: persist conversation history across calls using a checkpointer and a thread_id.
Since create_agent runs on LangGraph, memory is handled with a checkpointer: it saves the full graph state (including messages) tied to a thread_id, so you can pick the same conversation back up later.
agent_memory.py
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
agent = create_agent(
model="gpt-4o",
tools=[],
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "customer-42"}}
agent.invoke({"messages": [{"role": "user", "content": "I'm looking for a laptop for graphic design"}]}, config)
r = agent.invoke({"messages": [{"role": "user", "content": "Which of the ones you mentioned is the cheapest?"}]}, config)
print(r["messages"][-1].content)
InMemorySaver is great for prototypes — in production you'd use PostgresSaver or SqliteSaver so history survives a process restart.
Challenge
Change the thread_id mid-conversation and confirm that the agent "forgets" the previous context — that shows memory is tied to the thread, not the process.
Level 5 · Structured output
response_format with Pydantic
Goal: force the agent to return a validated object instead of free text.
agent.py
from pydantic import BaseModel
from langchain.agents import create_agent
class SupportTicket(BaseModel):
category: str
urgency: str
summary: str
agent = create_agent(model="gpt-4o", tools=[], response_format=SupportTicket)
result = agent.invoke({
"messages": [{"role": "user", "content": "The app crashes when uploading a profile photo, it's urgent"}]
})
ticket: SupportTicket = result["structured_response"]
print(ticket.category, ticket.urgency, ticket.summary)
LangChain automatically picks the right strategy: if the model supports native structured output (like OpenAI's JSON schemas) it uses that directly; otherwise, it wraps your schema as an internal tool the model must call.
Challenge
Define a Pydantic model to extract a rating, sentiment, and whether a specific issue is mentioned from a product review.
Level 6 · Streaming
Streaming with .stream()
Goal: display the agent's response in real time, token by token.
stream.py
for event, metadata in agent.stream(
{"messages": [{"role": "user", "content": "Explain what RAG is in 3 steps"}]},
stream_mode="messages",
):
if event.content:
print(event.content, end="", flush=True)
stream_mode="messages" gives you model tokens as they're generated. Other modes like "updates" show you each step of the graph (useful for debugging which tool ran and when). For async, use agent.astream() inside an async def function.
Challenge
Wrap the streaming in a FastAPI endpoint with StreamingResponse, using astream() instead of stream().
Level 7 · Multi-agent
Handoffs between agents
Goal: transfer control of a conversation from one specialized agent to another using the "handoff" pattern.
In LangGraph, a handoff is a special tool that, when run, doesn't just return a result — it tells the graph "from now on, another agent takes over." It's implemented by returning a Command object with goto (who to transfer to) and update (what to add to the shared state).
handoff.py
from langchain_core.tools import tool
from langgraph.types import Command
from langgraph.prebuilt import InjectedState
@tool
def transfer_to_support(state: InjectedState) -> Command:
"""Transfer the conversation to the technical support specialist."""
return Command(
goto="support_agent",
update={"messages": state["messages"]},
graph=Command.PARENT,
)
# The main agent receives this tool; when it decides to use it,
# control passes directly to the "support_agent" node in the graph
Unlike "agent as a tool" (where agent A calls agent B and waits for its response), a handoff transfers full control — agent B carries on the conversation directly with the user.
Challenge
Create a second handoff, transfer_to_sales, and test a conversation where the main agent decides which of the two to transfer to.
Level 8 · Orchestrator
Supervisor with StateGraph
Goal: build an explicit orchestrator that routes each message to the right specialist using a state graph.
When you need to see and control every routing decision explicitly (useful for debugging and tracing in production), you build the supervisor directly with StateGraph instead of letting an agent "decide on its own".
orchestrator.py
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
class State(TypedDict):
message: str
response: str
next: str
def supervisor_node(state: State) -> State:
# The supervisor uses the model with structured output to decide routing
decision = supervisor_agent.invoke({"messages": [{"role": "user", "content": state["message"]}]})
state["next"] = decision["structured_response"].agent # "sales" | "support" | "billing"
return state
def route(state: State) -> Literal["sales", "support", "billing"]:
return state["next"]
graph = StateGraph(State)
graph.add_node("supervisor", supervisor_node)
graph.add_node("sales", sales_node)
graph.add_node("support", support_node)
graph.add_node("billing", billing_node)
graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route, {"sales": "sales", "support": "support", "billing": "billing"})
graph.add_edge("sales", END)
graph.add_edge("support", END)
graph.add_edge("billing", END)
app = graph.compile()
print(app.invoke({"message": "I got charged twice for my subscription"}))
For this same problem, there's also the langgraph-supervisor package, which already ships this pattern pre-built if you don't need to customize the routing logic.
Challenge
Add an edge back from each specialist to the supervisor (instead of END) to support multi-turn conversations across several specialists in the same session.
Level 9 · Production
Traces, errors, and guardrails
Goal: instrument the graph with LangSmith and handle errors robustly.
TERMINAL
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-api-key"
export LANGSMITH_PROJECT="support-agent-prod"
python orchestrator.py
# Every graph node, every tool call, and every generated token
# is traced and visible at smith.langchain.com
Using a shared checkpointer (Postgres) instead of InMemorySaver is what lets you run multiple replicas of the same service without each one holding its own isolated, inconsistent memory.
2. Load balancer across replicas
nginx.conf
upstream support_agents {
least_conn;
server support-1:8001;
server support-2:8001;
server support-3:8001;
}
server {
listen 80;
location /support/ { proxy_pass http://support_agents/; }
}
With that, you've got the full path: from a one-line create_agent() call in Level 1, to a multi-agent supervisor in production, with persistent memory, horizontal scaling, and end-to-end traces.
Final challenge
Split the 3 specialists from Level 8 into independent FastAPI services with a shared Postgres checkpointer, and run 2 replicas of one of them behind nginx.