Introduction: Why Migrate from OpenAI Agents SDK to LangGraph?
The OpenAI Agents SDK (formerly known as Swarm) provides a lightweight, opinionated framework for building multi-agent systems with handoffs, guardrails, and tool calling. It's a great starting point for prototyping agent workflows. However, as your agent applications grow in complexity, you may find yourself needing more granular control over state management, conditional routing, human-in-the-loop interactions, and observability.
LangGraph, built by LangChain, is a framework designed specifically for creating stateful, multi-actor applications with LLMs. It models agent workflows as graphs — with nodes representing computation steps and edges representing the flow of control and state. This graph-based approach gives developers explicit control over execution, making it easier to build robust, production-grade agentic systems.
In this tutorial, we'll walk through a complete migration from the OpenAI Agents SDK to LangGraph, covering the conceptual mapping between the two frameworks, practical code examples, and best practices to ensure a smooth transition.
Understanding the Conceptual Differences
Before diving into code, it's important to understand how the two frameworks differ in their mental models.
OpenAI Agents SDK Mental Model
The OpenAI Agents SDK is built around a few core primitives:
- Agent: An LLM configured with instructions, tools, and a model.
- Handoff: A mechanism for one agent to transfer control to another.
- Guardrail: Input/output validation that runs in parallel with the agent.
- Runner: The execution engine that orchestrates the agent loop.
The SDK abstracts away much of the control flow. You define agents and handoffs, and the Runner handles the loop of calling the LLM, executing tools, and transferring between agents.
LangGraph Mental Model
LangGraph takes a different approach, modeling everything explicitly as a graph:
- State: A typed schema (typically a TypedDict or Pydantic model) that flows through the graph.
- Node: A function that receives state and returns updates to it.
- Edge: A connection between nodes, which can be conditional.
- Graph (StateGraph): The compiled, executable workflow.
This means concepts like "agents" and "handoffs" are implemented as nodes and conditional edges. You have full visibility into — and control over — the execution flow.
Project Setup
Let's start by setting up a project with both SDKs installed so we can compare implementations side by side. Create a new directory and install the required packages:
pip install openai-agents langgraph langchain-openai python-dotenv
Create a .env file with your API key:
OPENAI_API_KEY=sk-your-api-key-here
Now let's build a simple multi-agent system using the OpenAI Agents SDK, and then migrate it to LangGraph.
Building a Multi-Agent System with OpenAI Agents SDK
Our example will be a customer support system with two agents: a Triage Agent that routes requests, and a Billing Agent that handles billing-related questions. The Triage Agent can hand off to the Billing Agent when appropriate.
Defining Agents with the OpenAI Agents SDK
import os
from dotenv import load_dotenv
from agents import Agent, Runner, function_tool
load_dotenv()
# Define a tool for the billing agent
@function_tool
def get_billing_info(account_id: str) -> str:
"""Retrieve billing information for a given account."""
# Simulated data
database = {
"ACC001": {"balance": 125.50, "due_date": "2025-02-15", "plan": "Pro"},
"ACC002": {"balance": 0.00, "due_date": "2025-02-20", "plan": "Free"},
}
info = database.get(account_id, {"error": "Account not found"})
return str(info)
# Define the billing agent
billing_agent = Agent(
name="Billing Agent",
instructions="""You are a billing support specialist.
Help customers with billing questions using the get_billing_info tool.
Always be polite and reference the customer's account details in your response.""",
tools=[get_billing_info],
)
# Define the triage agent with a handoff to billing
triage_agent = Agent(
name="Triage Agent",
instructions="""You are a customer support triage agent.
For billing-related questions, hand off to the Billing Agent.
For general questions, answer directly.
Always greet the customer first.""",
handoffs=[billing_agent],
)
# Run the agent
result = Runner.run_sync(
triage_agent,
"Hi, I need to check my billing info for account ACC001."
)
print(result.final_output)
This is clean and concise. The OpenAI Agents SDK handles the handoff automatically — when the Triage Agent decides the query is billing-related, it transfers control to the Billing Agent, which then uses the tool to retrieve information and respond.
Migrating to LangGraph: Step by Step
Now let's migrate this same system to LangGraph. The migration involves several steps: defining state, creating nodes for each agent, setting up conditional edges for handoffs, and compiling the graph.
Step 1: Define the State Schema
In LangGraph, state is explicit and typed. We need to define what information flows between nodes. For our customer support system, we'll track the conversation messages and the current agent.
from typing import Annotated, TypedDict, 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, ToolMessage
from langchain_core.tools import tool
# Define the state schema
class SupportState(TypedDict):
messages: Annotated[list, add_messages]
current_agent: str
The add_messages reducer function ensures that new messages are appended to the list rather than replacing it. The current_agent field tracks which agent should handle the next step.
Step 2: Define Tools
LangGraph uses LangChain's tool definitions. The syntax is similar but uses @tool from langchain_core.tools instead of @function_tool from the Agents SDK.
@tool
def get_billing_info(account_id: str) -> str:
"""Retrieve billing information for a given account."""
database = {
"ACC001": {"balance": 125.50, "due_date": "2025-02-15", "plan": "Pro"},
"ACC002": {"balance": 0.00, "due_date": "2025-02-20", "plan": "Free"},
}
info = database.get(account_id, {"error": "Account not found"})
return str(info)
Step 3: Create Agent Nodes
Each agent becomes a node function. The node receives the current state, calls the LLM with the appropriate system prompt and tools, and returns updated state.
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# Bind tools to the billing LLM
billing_llm = llm.bind_tools([get_billing_info])
TRIAGE_SYSTEM_PROMPT = """You are a customer support triage agent.
For billing-related questions, indicate that you are transferring to the Billing Agent
by responding with exactly: TRANSFER_TO_BILLING
For general questions, answer directly.
Always greet the customer first."""
BILLING_SYSTEM_PROMPT = """You are a billing support specialist.
Help customers with billing questions using the get_billing_info tool.
Always be polite and reference the customer's account details in your response."""
def triage_node(state: SupportState) -> dict:
"""Triage agent node that routes requests."""
messages = [SystemMessage(content=TRIAGE_SYSTEM_PROMPT)] + state["messages"]
response = llm.invoke(messages)
return {"messages": [response], "current_agent": "triage"}
def billing_node(state: SupportState) -> dict:
"""Billing agent node that handles billing questions."""
messages = [SystemMessage(content=BILLING_SYSTEM_PROMPT)] + state["messages"]
response = billing_llm.invoke(messages)
return {"messages": [response], "current_agent": "billing"}
def tool_executor_node(state: SupportState) -> dict:
"""Execute any pending tool calls."""
last_message = state["messages"][-1]
tool_messages = []
for tool_call in last_message.tool_calls:
if tool_call["name"] == "get_billing_info":
result = get_billing_info.invoke(tool_call["args"])
tool_messages.append(
ToolMessage(content=result, tool_call_id=tool_call["id"])
)
return {"messages": tool_messages}
Step 4: Define Routing Logic (Handoffs)
In the OpenAI Agents SDK, handoffs are declarative — you list them on the agent and the SDK handles the rest. In LangGraph, we implement handoffs as conditional edges. We need routing functions that examine the state and determine which node to go to next.
def route_from_triage(state: SupportState) -> str:
"""Determine if triage should hand off to billing or end."""
last_message = state["messages"][-1]
# Check if the triage agent wants to transfer
if "TRANSFER_TO_BILLING" in last_message.content:
return "billing"
return "end"
def route_from_billing(state: SupportState) -> str:
"""Determine if billing needs to call tools or end."""
last_message = state["messages"][-1]
# If the LLM made tool calls, route to tool execution
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"
def route_from_tools(state: SupportState) -> str:
"""After tool execution, return to the billing agent."""
return "billing"
Step 5: Build and Compile the Graph
Now we wire everything together into a StateGraph. This is where the explicit control flow of LangGraph shines — you can see exactly how data moves through the system.
# Build the graph
graph = StateGraph(SupportState)
# Add nodes
graph.add_node("triage", triage_node)
graph.add_node("billing", billing_node)
graph.add_node("tools", tool_executor_node)
# Set entry point
graph.add_edge(START, "triage")
# Add conditional edges for routing
graph.add_conditional_edges(
"triage",
route_from_triage,
{
"billing": "billing",
"end": END,
}
)
graph.add_conditional_edges(
"billing",
route_from_billing,
{
"tools": "tools",
"end": END,
}
)
# After tools execute, go back to billing
graph.add_edge("tools", "billing")
# Compile the graph
app = graph.compile()
Step 6: Run the Graph
Now we can invoke the compiled graph with a user message, just as we did with the Runner in the Agents SDK.
# Run the graph
initial_state = {
"messages": [HumanMessage(content="Hi, I need to check my billing info for account ACC001.")],
"current_agent": "",
}
result = app.invoke(initial_state)
# Print the final response
print(result["messages"][-1].content)
Handling More Complex Scenarios
The basic migration above covers the fundamentals. Let's now look at some more advanced scenarios that demonstrate LangGraph's additional capabilities.
Adding a Third Agent with Bidirectional Handoffs
In a real support system, you might have a Technical Support agent in addition to Billing. Let's add this agent and enable bidirectional handoffs — the Billing agent should be able to transfer to Technical Support and vice versa.
TECH_SYSTEM_PROMPT = """You are a technical support specialist.
Help customers with technical issues like bugs, outages, and configuration.
If the question is about billing, respond with exactly: TRANSFER_TO_BILLING
If the question is general, respond with exactly: TRANSFER_TO_TRIAGE"""
# Update billing prompt to support handoffs
BILLING_SYSTEM_PROMPT_V2 = """You are a billing support specialist.
Help customers with billing questions using the get_billing_info tool.
If the question is about technical issues, respond with exactly: TRANSFER_TO_TECH
If the question is general, respond with exactly: TRANSFER_TO_TRIAGE
Always be polite and reference the customer's account details in your response."""
@tool
def search_kb(query: str) -> str:
"""Search the knowledge base for technical solutions."""
kb = {
"login": "Try clearing your browser cache and cookies, then attempt login again.",
"slow": "Check your internet connection. If the issue persists, try restarting the app.",
}
for key, value in kb.items():
if key in query.lower():
return value
return "No matching solution found. Please contact support."
tech_llm = llm.bind_tools([search_kb])
billing_llm_v2 = llm.bind_tools([get_billing_info])
def tech_node(state: SupportState) -> dict:
messages = [SystemMessage(content=TECH_SYSTEM_PROMPT)] + state["messages"]
response = tech_llm.invoke(messages)
return {"messages": [response], "current_agent": "tech"}
def billing_node_v2(state: SupportState) -> dict:
messages = [SystemMessage(content=BILLING_SYSTEM_PROMPT_V2)] + state["messages"]
response = billing_llm_v2.invoke(messages)
return {"messages": [response], "current_agent": "billing"}
def route_from_billing_v2(state: SupportState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
if "TRANSFER_TO_TECH" in last_message.content:
return "tech"
if "TRANSFER_TO_TRIAGE" in last_message.content:
return "triage"
return "end"
def route_from_tech(state: SupportState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
if "TRANSFER_TO_BILLING" in last_message.content:
return "billing"
if "TRANSFER_TO_TRIAGE" in last_message.content:
return "triage"
return "end"
def tool_executor_node_v2(state: SupportState) -> dict:
last_message = state["messages"][-1]
tool_messages = []
for tool_call in last_message.tool_calls:
if tool_call["name"] == "get_billing_info":
result = get_billing_info.invoke(tool_call["args"])
elif tool_call["name"] == "search_kb":
result = search_kb.invoke(tool_call["args"])
else:
result = "Unknown tool"
tool_messages.append(
ToolMessage(content=result, tool_call_id=call["id"])
)
return {"messages": tool_messages}
def route_from_tools_v2(state: SupportState) -> str:
return state.get("current_agent", "triage")
# Build the expanded graph
graph_v2 = StateGraph(SupportState)
graph_v2.add_node("triage", triage_node)
graph_v2.add_node("billing", billing_node_v2)
graph_v2.add_node("tech", tech_node)
graph_v2.add_node("tools", tool_executor_node_v2)
graph_v2.add_edge(START, "triage")
graph_v2.add_conditional_edges("triage", route_from_triage,
{"billing": "billing", "end": END})
graph_v2.add_conditional_edges("billing", route_from_billing_v2,
{"tools": "tools", "tech": "tech", "triage": "triage", "end": END})
graph_v2.add_conditional_edges("tech", route_from_tech,
{"tools": "tools", "billing": "billing", "triage": "triage", "end": END})
graph_v2.add_conditional_edges("tools", route_from_tools_v2,
{"billing": "billing", "tech": "tech", "triage": "triage"})
app_v2 = graph_v2.compile()
Adding Human-in-the-Loop Checkpoints
One of LangGraph's most powerful features is the ability to pause execution for human approval. This is difficult to achieve in the OpenAI Agents SDK but straightforward in LangGraph using checkpointers and interrupts.
from langgraph.checkpoint.memory import MemorySaver
# Add a checkpointer for state persistence
checkpointer = MemorySaver()
# Compile with interrupt before tool execution
app_with_interrupt = graph.compile(
checkpointer=checkpointer,
interrupt_before=["tools"]
)
# Run with a thread ID for state tracking
config = {"configurable": {"thread_id": "thread-1"}}
initial_state = {
"messages": [HumanMessage(content="Check billing for ACC001")],
"current_agent": "",
}
# First invocation — will pause before tools
result = app_with_interrupt.invoke(initial_state, config=config)
# Review the pending tool call
last_msg = result["messages"][-1]
print(f"Pending tool call: {last_msg.tool_calls}")
# Resume execution after human approval
result = app_with_interrupt.invoke(None, config=config)
print(result["messages"][-1].content)
This pattern is invaluable for production systems where tool calls — especially those with side effects like issuing refunds or modifying accounts — require human oversight.
Migrating Guardrails
The OpenAI Agents SDK has a built-in guardrail system for input and output validation. In LangGraph, you implement guardrails as additional nodes in the graph. Here's how to migrate a simple input guardrail:
OpenAI Agents SDK Guardrail
from agents import Agent, Runner, InputGuardrail, GuardrailResult
async def profanity_guardrail(ctx, agent, input_data):
banned_words = ["spam", "scam"]
if any(word in input_data.lower() for word in banned_words):
return GuardrailResult(triggered=True, output="Inappropriate content detected.")
return GuardrailResult(triggered=False)
agent = Agent(
name="Safe Agent",
instructions="You are a helpful assistant.",
input_guardrails=[profanity_guardrail],
)
LangGraph Equivalent
def guardrail_node(state: SupportState) -> dict:
"""Input guardrail that checks for inappropriate content."""
last_message = state["messages"][-1]
banned_words = ["spam", "scam"]
if any(word in last_message.content.lower() for word in banned_words):
return {
"messages": [SystemMessage(content="Inappropriate content detected. Request blocked.")],
"current_agent": "blocked"
}
return {"current_agent": "triage"}
def route_from_guardrail(state: SupportState) -> str:
if state.get("current_agent") == "blocked":
return "end"
return "triage"
# Add to graph
graph.add_node("guardrail", guardrail_node)
graph.add_edge(START, "guardrail")
graph.add_conditional_edges("guardrail", route_from_guardrail,
{"triage": "triage", "end": END})
Best Practices for Migration
1. Map Your Agents to Nodes First
Before writing any code, create a diagram of your agent system. Each agent becomes a node, each handoff becomes a conditional edge, and each tool execution becomes a tool node. Having this visual map makes the migration much smoother.
2. Design Your State Schema Carefully
The state schema is the backbone of your LangGraph application. Think about what data each node needs to read and what it needs to write. Use Annotated types with reducers like add_messages for accumulating data. Avoid putting large, unnecessary data in state — keep it focused on what the graph needs to make routing decisions.
3. Use Structured Output for Routing Decisions
Instead of parsing free-text responses for transfer signals (like "TRANSFER_TO_BILLING"), consider using structured outputs for more reliable routing:
from pydantic import BaseModel, Field
class RoutingDecision(BaseModel):
next_agent: Literal["triage", "billing", "tech", "end"] = Field(
description="The next agent to route to"
)
reasoning: str = Field(description="Why this routing decision was made")
routing_llm = llm.with_structured_output(RoutingDecision)
def triage_node_structured(state: SupportState) -> dict:
messages = [SystemMessage(content=TRIAGE_SYSTEM_PROMPT)] + state["messages"]
decision = routing_llm.invoke(messages)
return {"current_agent": decision.next_agent}
4. Leverage Checkpointing for Production
Always use a persistent checkpointer (like SqliteSaver or PostgresSaver) in production. This enables state recovery after crashes, human-in-the-loop workflows, and conversation history persistence.
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)
5. Add Recursion Limits
LangGraph allows you to set a recursion limit to prevent infinite loops — something that's especially important when agents can hand off to each other bidirectionally.
result = app.invoke(
initial_state,
config={"recursion_limit": 25}
)
6. Use Streaming for Better UX
LangGraph supports streaming both tokens and state updates. This provides a much better user experience than waiting for the entire agent loop to complete:
for event in app.stream(initial_state, stream_mode="values"):
last_message = event["messages"][-1]
if hasattr(last_message, "content") and last_message.content:
print(f"[{event.get('current_agent', '?')}] {last_message.content}")
7. Test Nodes in Isolation
One advantage of LangGraph's explicit node structure is that each node is just a function. You can unit test nodes independently by passing in mock state:
def test_billing_node_with_tool_call():
state = {
"messages": [HumanMessage(content="Check ACC001")],
"current_agent": "billing",
}
result = billing_node_v2(state)
assert result["current_agent"] == "billing"
assert len(result["messages"]) == 1
Common Pitfalls and How to Avoid Them
Pitfall 1: Forgetting to Return State Updates
In the OpenAI Agents SDK, the Runner manages state implicitly. In LangGraph, nodes must explicitly return the state updates. A common mistake is modifying state in place without returning it. Always return a dictionary with the fields you want to update.
Pitfall 2: Infinite Handoff Loops
When agents can hand off to each other, it's possible to create cycles where agents bounce a request back and forth indefinitely. Always set a recursion limit and consider adding a "handoff count" to your state to break ties:
class SupportState(TypedDict):
messages: Annotated[list, add_messages]
current_agent: str
handoff_count: int
# In routing logic
def route_from_billing(state: SupportState) -> str:
if state.get("handoff_count", 0) >= 5:
return "end"
# ... normal routing
Pitfall 3: Not Handling Tool Errors
The OpenAI Agents SDK has built-in error handling for tools. In LangGraph, you need to handle errors explicitly in your tool executor node:
def tool_executor_node(state: SupportState) -> dict:
last_message = state["messages"][-1]
tool_messages = []
for tool_call in last_message.tool_calls:
try:
if tool_call["name"] == "get_billing_info":
result = get_billing_info.invoke(tool_call["args"])
elif tool_call["name"] == "search_kb":
result = search_kb.invoke(tool_call["args"])
else:
result = f"Error: Unknown tool {tool_call['name']}"
except Exception as e:
result = f"Error executing tool: {str(e)}"
tool_messages.append(
ToolMessage(content=result, tool_call_id=tool_call["id"])
)
return {"messages": tool_messages}
Feature Mapping Reference
Here's a quick reference table for mapping OpenAI Agents SDK concepts to LangGraph equivalents:
| OpenAI Agents SDK | LangGraph Equivalent |
|----------------------------|---------------------------------------------|
| Agent | Node function |
| Runner.run_sync() | graph.compile().invoke() |
| Handoff | Conditional edge |
| function_tool | @tool from langchain_core.tools |
| InputGuardrail | Guardrail node + conditional edge |
| OutputGuardrail | Post-processing node + conditional edge |
| Agent.instructions | SystemMessage in node function |
| Agent.model | ChatOpenAI instance in node function |
| Agent.tools | llm.bind_tools() in node function |
| Tracing (built-in) | LangSmith integration |
| Context (local state) | State schema fields |
| Session (conversation) | Thread ID + checkpointer |
Conclusion
Migrating from the OpenAI Agents SDK to LangGraph requires a shift in mental model — from declarative agent definitions to explicit graph-based control flow. While this means writing more code upfront, the payoff is significant: you gain full control over state management, routing logic, human-in-the-loop interactions, error handling, and observability. The graph structure makes your agent system more debuggable, testable, and production-ready. Start by mapping your existing agents to nodes and handoffs to edges, design your state schema thoughtfully, and incrementally add features like checkpointing, streaming, and structured routing. As your application grows in complexity, you'll find that LangGraph's explicit approach scales far better than the implicit abstractions of the Agents SDK, giving you the flexibility to build sophisticated multi-agent systems that are both powerful and maintainable.