- Break Into Data
- Posts
- Build effective agents with OpenAI Agents SDK
Build effective agents with OpenAI Agents SDK
Learn how to build 5 common patterns for agentic systems introduced by Anthropic with OpenAI Agent SDK
Join our highly rated AI Engineer Bootcamp in June 2025 and take your AI Engineering skills to the next level in only 6 weeks!
We’re entering a new phase of GenAI in 2025, one where systems don’t only generate answers, but can reason, make decisions, and act autonomously. Anthropic’s article on Building Effective Agents introduced 5 common patterns in agentic systems that capture this shift well.
But here’s the thing: knowing the theory isn’t enough.
In this article, you will learn how to actually build each of these five patterns using OpenAI’s Agent SDK. The Agent SDK lets engineers connect LLMs with tools and coordinate different agents to handle multi-step complex tasks together.

Overview of agentic workflows
We will cover:
Whether you're building a coding tutor, a RAG-based assistant, or a multi-agent system, this guide will help you think modularly and implement robust agentic systems.
If you’re new to working with Agents, sign up for a free 14-Day AI Product Engineering Course to get up to speed.
Now let’s begin!
Building Blocks
The OpenAI Agent SDK introduces some old and some new concepts to define and manage agent workflows:
Agents: LLMs configured with instructions, tools, guardrails, and handoffs
Tools: Functions that agents can use to take actions beyond just generating text
Handoffs: A specialized tool used for transferring control between agents
Context: Memory of past actions and custom context passed to
Guardrails: Configurable safety checks for input and output validation
*The Agent SDK also has a tracing module that allows you to view, debug, and optimize your workflows inside OpenAI’s developer dashboard.

Agent as a feedback loop with access to your custom environment
Understanding these building blocks will help you design more effective workflows and agent-based applications.
*For this article, we will use a coding agent tutor as an example throughout the tutorial for simplicity.
Agents
How to define agents?
from agents import Agent
basic_agent = Agent(
name="My First Agent",
instructions="You are a helpful coding tutor.",
model="gpt-4o" # Optional: defaults to "gpt-4o" if not specified
)
At the center of the OpenAI Agent SDK is the Agent class. It has 3 main components: name, instructions, and model.
Additionally, you can select and define more attributes, like tools, output_type, and handoffs. See the documentation for more details.
How to run agents?
To run agents, you need to use the Runner class. There are 3 different ways to run agents:
Runner.run()
, which runs async and returns aRunResult
. This is the most popular use case.
# Use asyncio for Python scripts:
import asyncio
result = await Runner.run(agent, "What are Python Decorators? ")
print(result.final_output)
Runner.run_sync()
, which is a sync method and just runs.run()
. If you want a simple approach, use this example.
result = Runner.run_sync(agent, "What are Python Decorators?")
print(result.final_output)
Runner.run_streamed()
, which runs async and returns aRunResultStreaming
. It calls the LLM in streaming mode and streams those events to you as they are received.
import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner
result = Runner.run_streamed(agent, input="What are Python Decorators?")
async for event in result.stream_events():
if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
print(event.data.delta, end="", flush=True)
Prompt Chaining
Sometimes, solving a problem takes more than one step, especially when each step builds on the last. That’s where prompt chaining comes in. This workflow creates a linear chain of agents, where each one takes the output of the previous agent and pushes the task further.
When to use prompt chaining:
Prompt chaining is ideal when a task has clear sequential steps that depend on one another. It’s especially useful for workflows that require validation or review before moving forward (see "gate” in the diagram below).

Prompt Chaining Workflow
In our example below, we build a personalized coding tutor using three agents:
The first agent generates a curriculum outline based on a learning goal.
We feed the outline into the second agent to assess if it matches the goal and is high quality.
If the curriculum is poor or off-topic, we stop.
If it’s a good fit, we pass it to the third agent.
The third agent writes detailed lessons for each section.
import asyncio
from pydantic import BaseModel
from agents import Agent, Runner, trace
# Agent 1: Generate learning plan
curriculum_agent = Agent(
name="curriculum_agent",
instructions="Generate a structured curriculum outline to help someone achieve the programming learning goal they provide.",
)
# Output schema for the curriculum checker
class CurriculumCheckOutput(BaseModel):
good_quality: bool
matches_goal: bool
# Agent 2: Evaluate learning plan
curriculum_checker_agent = Agent(
name="curriculum_checker_agent",
instructions="Evaluate the provided curriculum outline. Determine if it is high quality and if it matches the user’s learning goal.",
output_type=CurriculumCheckOutput,
)
# Agent 3: Generate full lessons for each section
lesson_writer_agent = Agent(
name="lesson_writer_agent",
instructions="Write a detailed coding lesson for each section in the curriculum outline. Include explanations, code examples, and one practice question per section.",
output_type=str,
)
async def main():
learning_goal = input("What do you want to learn? ")
with trace("Deterministic tutor flow"):
# 1. Generate curriculum
curriculum_result = await Runner.run(
curriculum_agent,
learning_goal,
)
print("\nCurriculum generated.")
# 2. Check the curriculum
check_result = await Runner.run(
curriculum_checker_agent,
curriculum_result.final_output,
)
assert isinstance(check_result.final_output, CurriculumCheckOutput)
if not check_result.final_output.good_quality:
print("Curriculum is low quality. Stopping.")
exit(0)
if not check_result.final_output.matches_goal:
print("Curriculum doesn't match the learning goal. Stopping.")
exit(0)
print("Curriculum looks good. Proceeding to generate lessons...")
# 3. Generate lessons
lessons_result = await Runner.run(
lesson_writer_agent,
curriculum_result.final_output,
)
print(f"\nGenerated Lessons:\n{lessons_result.final_output}")
if __name__ == "__main__":
asyncio.run(main())
Parallelization
Agentic workflows can handle tasks simultaneously, and their outputs can be combined or evaluated afterward. This approach is known as parallelization, and it comes in 2 main forms:
Sectioning: Split a task into sub-tasks that are independent from one another and run them in parallel.
Voting: Run the same task multiple times in parallel to generate diverse outputs, then select the best one.
When to use parallelization:
The task can be broken down into parts that can be done faster in parallel.
You want multiple perspectives or repeated attempts to boost the quality or confidence of the final result.
For complex problems with several factors to consider, it's often better to let each LLM call focus on just one factor. This leads to clearer, more thoughtful outputs and improves overall performance.

Parallelization Workflow
In our example below, we run the coding_explainer
agent three times in parallel, and pick the best result with explanation_picker
agent.
import asyncio
from agents import Agent, ItemHelpers, Runner, trace
# Agent that gives Python explanations
coding_explainer = Agent(
name="coding_explainer",
instructions="You are a coding tutor. You explain the given programming concept snippet in simple terms for a beginner learning to code.",
)
# Agent that selects the best explanation
explanation_picker = Agent(
name="explanation_picker",
instructions="You choose the most beginner-friendly and technically correct explanation from the provided options.",
)
async def main():
code_snippet = input("What's your programming question?:\n\n")
# Run three parallel explanation attempts
with trace("Parallel code explanations"):
res_1, res_2, res_3 = await asyncio.gather(
Runner.run(coding_explainer, code_snippet),
Runner.run(coding_explainer, code_snippet),
Runner.run(coding_explainer, code_snippet),
)
explanations = [
ItemHelpers.text_message_outputs(res_1.new_items),
ItemHelpers.text_message_outputs(res_2.new_items),
ItemHelpers.text_message_outputs(res_3.new_items),
]
all_explanations = "\n\n".join(explanations)
print(f"\n\nCandidate Explanations:\n\n{all_explanations}")
# Select the best explanation
best_explanation = await Runner.run(
explanation_picker,
f"Code:\n{code_snippet}\n\nExplanations:\n{all_explanations}",
)
print("\n\n-----")
print(f"Best explanation:\n{best_explanation.final_output}")
if __name__ == "__main__":
asyncio.run(main())
Routing
Not every problem can be solved with the same kind of agent. That’s where routing comes in. Instead of relying on one generalist agent, this workflow assigns incoming requests to the right specialist based on the user’s query.
When to use routing:
Routing is powerful when your system supports multiple domains or expertise areas, and you want to give users the best possible answer from the most relevant source. It’s perfect for coding help, customer support, or language-based queries.
By giving each agent a focused role and letting a coordinator handle direction, this pattern keeps your AI system organized.

In this example, we’ve built a smart, multi-language coding tutor, that has 4 agents:
Python tutor agent
JavaScript tutor agent
SQL tutor agent
Routing agent
The routing agent receives a user's question and routes it to the appropriate specialist tutor: Python, JavaScript, or SQL.
import asyncio
import uuid
from agents import Agent, RawResponsesStreamEvent, Runner, TResponseInputItem, trace
python_tutor = Agent(
name="python_tutor",
instructions="You're a Python expert. Help the user understand or debug Python code.",
)
js_tutor = Agent(
name="js_tutor",
instructions="You're a JavaScript expert. Help the user with JS questions or problems.",
)
sql_tutor = Agent(
name="sql_tutor",
instructions="You're an SQL expert. Help the user write or optimize SQL queries.",
)
routing_agent = Agent(
name="triage_agent",
instructions="You're a smart tutor coordinator. Based on the user's message, route them to the correct specialist (Python, JavaScript, or SQL).",
handoffs=[python_tutor, js_tutor, sql_tutor],
)
async def main():
conversation_id = str(uuid.uuid4().hex[:16])
print("Welcome to Code Tutor! Ask me anything about Python, JavaScript, or SQL.")
msg = input("What would you like help with? ")
agent = routing_agent
while True:
with trace("Coding tutor routing"):
result = await Runner.run(
agent,
input=msg,
)
print("\nAssistant:")
print(result.final_output)
inputs = result.to_input_list()
print("\n")
if __name__ == "__main__":
asyncio.run(main())
Orchestration-workers
Not every project can be mapped out step‑by‑step in advance; that’s where orchestration comes in. A single “orchestrator” agent inspects the request, breaks it into dynamic subtasks, spins up worker agents to handle each piece, then merges their outputs into one coherent answer.
When to use orchestration:
Use it for jobs where the steps emerge only after the user inputs a query. The orchestrator keeps everyone in sync while the worker agents work in parallel, giving you the agility of multiple specialists without losing overall control.
Examples could be:
cleaning up a large codebase
compiling a multi‑source research brief
drafting a report that keeps changing as new data drops.

In this example, we’ve built a multi-agent personalized syllabus creator agentic system, which has 3 agents:
Planner agent (Orchestrator)
Search agent
Writer agent
The planner agent receives a user's topic and generates search queries, which are executed by search agents. And finally, our writer agent will write the final syllabus.
import asyncio
from pydantic import BaseModel
from agents import Agent, WebSearchTool
from agents.model_settings import ModelSettings
class WebSearchItem(BaseModel):
reason: str # why we’re running this search
query: str # the exact search term
class WebSearchPlan(BaseModel):
searches: list[WebSearchItem]
class SyllabusData(BaseModel):
short_summary: str
markdown_report: str
follow_up_questions: list[str]
planner_agent = Agent(
name="PlannerAgent",
instructions=(
"You are a programming syllabus creator agent. Given a programming topic, propose 5-20 web searches that, together, will answer it comprehensively. Return them as JSON in the schema provided."
),
model="gpt-4o",
output_type=WebSearchPlan,
)
search_agent = Agent(
name="SearchAgent",
instructions=(
"You are an internet researcher. Use the WebSearch tool to gather the most relevant information for the given query. Summarize your findings in clear, markdown bullets (≤300 words)."
),
tools=[WebSearchTool()],
# force the model to choose the tool; no stray text-only answers
model_settings=ModelSettings(tool_choice="required"),
)
writer_agent = Agent(
name="WriterAgent",
instructions=(
"You are a senior technical syllabus writer agent. Combine the provided research summaries into a coherent, well-structured report (5-10 pages of markdown). Begin with a one-paragraph outline of the syllabus, then the full syllabus."
),
model="o3-mini",
output_type=SyllabusData,
)
class ResearchManager:
async def run(self, query: str) -> None:
trace_id = gen_trace_id()
with trace("Research trace", trace_id=trace_id):
plan = await self._plan_searches(query)
results = await self._perform_searches(plan)
report = await self._write_report(query, results)
print(report.markdown_report)
async def _plan_searches(self, query: str):
result = await Runner.run(planner_agent, f"Query: {query}")
return result.final_output.searches
async def _perform_searches(self, searches):
tasks = [asyncio.create_task(self._search(item)) for item in searches]
return [await t for t in asyncio.as_completed(tasks)]
async def _search(self, item):
res = await Runner.run(search_agent, item.query)
return res.final_output
async def _write_report(self, query: str, summaries):
joined = "\n".join(summaries)
output = await Runner.run(writer_agent,
f"Original query: {query}\nResearch:\n{joined}"
)
return output.final_output
Evaluator - Optimizer
Some jobs deserve an internal editor. In the evaluator‑optimizer loop, one agent drafts an answer while a second agent plays reviewer, grading the draft against a rubric, pointing out gaps, and asking for another pass if needed. The pair iterates until the reviewer is satisfied, producing work that’s better than any agent could manage alone.
When an evaluator is worth adding:
Reach for this pattern when you can spell out what “good” looks like (style rules, accuracy checks, test cases) and the draft truly improves with targeted feedback.

In this example, we are using a common LLM as a judge pattern.
The first agent generates a coding exercise based on the user’s query.
The second agent judges the exercise and provides feedback on how to improve it as well as whether it passes or not.
from __future__ import annotations
import asyncio
from dataclasses import dataclass
from typing import Literal
from agents import Agent, ItemHelpers, Runner, TResponseInputItem, trace
@dataclass
class EvaluationFeedback:
feedback: str
score: Literal["pass", "needs_improvement", "fail"]
exercise_generator = Agent(
name="story_outline_generator",
instructions=(
"You generate a coding exercise based on the user's topic of interest.If there is any feedback provided, use it to improve the exercise."
),
)
evaluator = Agent[None](
name="evaluator",
instructions=(
"You evaluate the exercise and decide if it's good enough. If it's not good enough, you provide feedback on what needs to be improved. Never give it a pass on the first try."
),
output_type=EvaluationFeedback,
)
async def main() -> None:
msg = input("What kind of coding exercise would you like? ")
input_items: list[TResponseInputItem] = [{"content": msg, "role": "user"}]
latest_outline: str | None = None
# We'll run the entire workflow in a single trace
with trace("LLM as a judge"):
while True:
exercise_result = await Runner.run(
exercise_generator,
input_items,
)
input_items = story_outline_result.to_input_list()
latest_outline = ItemHelpers.text_message_outputs(exercise_result.new_items)
print("Exercise generated")
evaluator_result = await Runner.run(evaluator, input_items)
result: EvaluationFeedback = evaluator_result.final_output
print(f"Evaluator score: {result.score}")
if result.score == "pass":
print("Coding exercise is good enough, exiting.")
break
print("Re-running with feedback")
input_items.append({"content": f"Feedback: {result.feedback}", "role": "user"})
print(f"Final exercise: {latest_outline}")
if __name__ == "__main__":
asyncio.run(main())
Congratulations on finishing this article!
You’ve just taken the first step into the world of AI agents,and things are only going to get more exciting from here.
We are launching our next AI Product Engineering Bootcamp this June 2025!
In this bootcamp, you will:
Design a custom multi-agent workflow tailored to a real-world problem.
Implement and test agents using the OpenAI Agents SDK or AI SDK from Vercel and tracing tools.
Deploy your multi-agent system and monitor performance with real-time traces.
AND SO MUCH MORE!