What is a Data Analysis Agent?
A Data Analysis Agent is an AI-powered assistant that can autonomously perform end-to-end data analysis tasks — from loading and cleaning data to running statistical computations and generating visualizations. Unlike traditional scripts that follow a fixed path, an agent reasons about the data dynamically, chooses the right analytical approach, and iteratively refines its output based on results.
With the OpenAI Agents SDK, you can build such an agent by combining large language models (LLMs) with custom tools — Python functions that the agent can call to read files, compute statistics, create charts, and more. The agent becomes a collaborative partner: you describe what you want to learn from your data, and it figures out the how.
Why the OpenAI Agents SDK Matters for Data Analysis
Building a data analysis agent from scratch requires solving several hard problems: orchestrating multiple LLM calls, managing tool execution, handling errors gracefully, and keeping the agent on track. The OpenAI Agents SDK provides production-ready abstractions for all of these:
- Agents — encapsulate LLM configuration, instructions, and tools in a single reusable unit
- Tools — Python functions decorated for the LLM to discover and invoke autonomously
- Handoffs — delegate complex sub-tasks to specialized sub-agents (e.g., one agent for statistical analysis, another for visualization)
- Guardrails — validate inputs and outputs to prevent the agent from producing misleading or unsafe results
- Runner — manages the conversation loop, tool execution, and streaming with minimal boilerplate
This tutorial walks you through building a complete, working data analysis agent using the OpenAI Agents SDK. You'll learn how to set up the environment, define analysis tools, wire them into an agent, and run it against real datasets.
Prerequisites and Installation
Before diving in, make sure you have:
- Python 3.10 or later installed on your system
- An OpenAI API key (set as the environment variable
OPENAI_API_KEY) - Basic familiarity with pandas, matplotlib, and async Python (helpful but not required)
Install the required packages in a virtual environment:
# Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# Install the OpenAI Agents SDK and data science dependencies
pip install openai-agents pandas matplotlib seaborn numpy
Verify your API key is accessible:
import os
print("API Key set:", "OPENAI_API_KEY" in os.environ)
# Should print: API Key set: True
Core Concepts of the OpenAI Agents SDK
Before writing code, let's understand the four pillars of the SDK that you'll use to build your data analysis agent.
1. Agents
An Agent is the central abstraction. It bundles an LLM model, a system prompt (instructions), and a set of tools. You create an agent with Agent(), specifying its name, instructions, and which tools it can use. The agent does not execute itself — it's a configuration that the Runner uses to drive a conversation.
from agents import Agent
analysis_agent = Agent(
name="DataAnalyst",
instructions="You are a skilled data analyst. Help users explore datasets.",
tools=[], # We'll add tools soon
model="gpt-4o",
)
2. Tools (Functions)
Tools are Python functions that the LLM can invoke. You decorate a function with @function_tool (or use FunctionTool directly). The SDK automatically generates a JSON schema from the function's type hints and docstring, enabling the model to call it with the correct arguments. Tools can be synchronous or async.
from agents import function_tool
@function_tool
def compute_mean(values: list[float]) -> dict:
"""Calculate the arithmetic mean of a list of numbers.
Args:
values: A list of floating-point numbers.
Returns:
A dict with keys 'mean' and 'count'.
"""
if not values:
return {"mean": None, "count": 0, "error": "Empty list"}
return {"mean": sum(values) / len(values), "count": len(values)}
3. Handoffs
Handoffs allow one agent to delegate work to another specialized agent. For example, your main data analyst agent might hand off complex statistical modeling to a dedicated StatisticsAgent, or chart generation to a VisualizationAgent. The SDK manages the transition seamlessly, preserving conversation context.
from agents import Agent, handoff
stats_agent = Agent(
name="StatisticsExpert",
instructions="Perform advanced statistical tests and regression analysis.",
tools=[t_test_tool, regression_tool],
model="gpt-4o",
)
main_agent = Agent(
name="DataAnalyst",
instructions="You are a data analyst. For advanced statistics, hand off to StatisticsExpert.",
tools=[load_csv_tool, compute_describe_tool],
handoffs=[handoff(stats_agent)],
model="gpt-4o",
)
4. Guardrails
Guardrails validate agent inputs and outputs. An input guardrail can reject or transform user messages before they reach the agent. An output guardrail can check the agent's final response for correctness or safety. For data analysis, you might use guardrails to ensure the agent doesn't fabricate statistics or access forbidden file paths.
from agents import Guardrail, guardrail_check
@guardrail_check
def validate_file_path(request: str) -> bool:
"""Prevent access to sensitive system paths."""
forbidden = ["/etc/", "C:\\Windows", "~/.ssh"]
return not any(f in request for f in forbidden)
Building a Data Analysis Agent: Step-by-Step
Now let's build a fully functional data analysis agent. We'll create tools for loading data, computing statistics, and generating visualizations, then wire everything into a single agent that can handle real analytical queries.
Step 1: Project Structure
Create a new directory for your project with the following structure:
data_analyst_agent/
├── agent.py # Main agent definition and runner
├── tools/
│ ├── __init__.py
│ ├── data_loader.py
│ ├── statistics.py
│ └── visualization.py
└── requirements.txt
Initialize the tools package:
# tools/__init__.py
from .data_loader import load_csv, get_data_summary
from .statistics import compute_descriptive_stats, run_correlation
from .visualization import create_bar_chart, create_scatter_plot
__all__ = [
"load_csv",
"get_data_summary",
"compute_descriptive_stats",
"run_correlation",
"create_bar_chart",
"create_scatter_plot",
]
Step 2: Data Loading Tools
The agent needs the ability to load CSV files and inspect their structure. We'll use pandas for robust data handling. Create tools/data_loader.py:
# tools/data_loader.py
import pandas as pd
from typing import Optional
from agents import function_tool
# In-memory cache to avoid reloading the same file repeatedly
_data_cache: dict[str, pd.DataFrame] = {}
@function_tool
def load_csv(file_path: str, delimiter: str = ",", encoding: str = "utf-8") -> dict:
"""Load a CSV file and return a summary of its contents.
Use this tool whenever you need to read a dataset from disk.
After loading, the data is cached in memory so subsequent
operations on the same file are instantaneous.
Args:
file_path: Absolute or relative path to the CSV file.
delimiter: Column delimiter character (default comma).
encoding: File encoding (default utf-8).
Returns:
A dictionary with keys: 'file_path', 'row_count',
'column_count', 'columns', 'dtypes', 'first_rows' (list of dicts),
and 'missing_values_summary'.
"""
try:
df = pd.read_csv(file_path, delimiter=delimiter, encoding=encoding)
except FileNotFoundError:
return {"error": f"File not found: {file_path}"}
except Exception as e:
return {"error": f"Failed to read CSV: {str(e)}"}
# Cache the dataframe
_data_cache[file_path] = df
# Build summary
first_rows = df.head(5).fillna("NaN").to_dict(orient="records")
missing = df.isnull().sum().to_dict()
missing_summary = {col: int(count) for col, count in missing.items() if count > 0}
return {
"file_path": file_path,
"row_count": len(df),
"column_count": len(df.columns),
"columns": list(df.columns),
"dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
"first_rows": first_rows,
"missing_values_summary": missing_summary,
}
@function_tool
def get_data_summary(file_path: str) -> dict:
"""Get a comprehensive statistical summary of a previously loaded dataset.
Requires that load_csv has been called first for this file_path.
Args:
file_path: Path to the CSV file that was previously loaded.
Returns:
Dictionary with describe() statistics for numeric columns,
and value counts for categorical columns.
"""
if file_path not in _data_cache:
return {"error": f"File not loaded. Please call load_csv('{file_path}') first."}
df = _data_cache[file_path]
numeric_cols = df.select_dtypes(include=['number']).columns.tolist()
categorical_cols = df.select_dtypes(include=['object', 'category']).columns.tolist()
summary = {}
if numeric_cols:
desc = df[numeric_cols].describe().to_dict()
summary["numeric_summary"] = {
col: {stat: round(val, 2) if isinstance(val, float) else val
for stat, val in stats.items()}
for col, stats in desc.items()
}
if categorical_cols:
cat_summary = {}
for col in categorical_cols[:5]: # Limit to first 5 to control output size
vc = df[col].value_counts().head(10).to_dict()
cat_summary[col] = vc
summary["categorical_summary"] = cat_summary
summary["categorical_columns_count"] = len(categorical_cols)
summary["total_rows"] = len(df)
summary["total_columns"] = len(df.columns)
return summary
Step 3: Statistical Analysis Tools
Next, create tools for descriptive statistics and correlation analysis. These tools operate on the in-memory cache so the agent doesn't need to pass raw data back and forth. Create tools/statistics.py:
# tools/statistics.py
import pandas as pd
import numpy as np
from typing import Optional
from agents import function_tool
# Reference the cache from data_loader
from tools.data_loader import _data_cache
@function_tool
def compute_descriptive_stats(
file_path: str,
column: str,
include_distribution: bool = False
) -> dict:
"""Compute detailed descriptive statistics for a specific column.
Works with both numeric and categorical columns.
Args:
file_path: Path to the CSV file (must be loaded first).
column: Name of the column to analyze.
include_distribution: If True, include histogram/frequency data
for numeric columns or value counts for categorical columns.
Returns:
Dictionary with statistics specific to the column's data type.
"""
if file_path not in _data_cache:
return {"error": f"File not loaded. Call load_csv('{file_path}') first."}
df = _data_cache[file_path]
if column not in df.columns:
return {"error": f"Column '{column}' not found. Available columns: {list(df.columns)}"}
series = df[column]
if pd.api.types.is_numeric_dtype(series):
stats = {
"column": column,
"type": "numeric",
"count": int(series.count()),
"missing": int(series.isnull().sum()),
"mean": round(series.mean(), 4),
"median": round(series.median(), 4),
"std": round(series.std(), 4),
"min": round(series.min(), 4),
"max": round(series.max(), 4),
"q25": round(series.quantile(0.25), 4),
"q75": round(series.quantile(0.75), 4),
"skewness": round(series.skew(), 4),
}
if include_distribution:
hist_values, hist_bins = np.histogram(
series.dropna(), bins=10
)
stats["histogram"] = {
"counts": hist_values.tolist(),
"bin_edges": [round(b, 2) for b in hist_bins.tolist()],
}
else:
stats = {
"column": column,
"type": "categorical",
"count": int(series.count()),
"missing": int(series.isnull().sum()),
"unique_values": int(series.nunique()),
"most_common": series.mode().iloc[0] if not series.mode().empty else None,
"most_common_count": int(series.value_counts().iloc[0]) if not series.empty else 0,
}
if include_distribution:
stats["value_counts"] = series.value_counts().head(15).to_dict()
return stats
@function_tool
def run_correlation(
file_path: str,
columns: list[str],
method: str = "pearson"
) -> dict:
"""Calculate correlation coefficients between specified numeric columns.
Args:
file_path: Path to the CSV file (must be loaded first).
columns: List of numeric column names to correlate.
method: Correlation method - 'pearson', 'spearman', or 'kendall'.
Returns:
Dictionary with correlation matrix and interpretation hints.
"""
if file_path not in _data_cache:
return {"error": f"File not loaded. Call load_csv('{file_path}') first."}
df = _data_cache[file_path]
# Validate columns exist and are numeric
missing_cols = [c for c in columns if c not in df.columns]
if missing_cols:
return {"error": f"Columns not found: {missing_cols}. Available: {list(df.columns)}"}
non_numeric = [c for c in columns if not pd.api.types.is_numeric_dtype(df[c])]
if non_numeric:
return {"error": f"Non-numeric columns: {non_numeric}. Correlation requires numeric data."}
try:
corr_matrix = df[columns].corr(method=method)
# Round for readability
corr_dict = corr_matrix.to_dict()
for col in corr_dict:
for row in corr_dict[col]:
corr_dict[col][row] = round(corr_dict[col][row], 4)
# Find strongest correlations
pairs = []
for i, col1 in enumerate(columns):
for col2 in columns[i+1:]:
val = corr_dict[col1][col2]
pairs.append({
"pair": f"{col1} ↔ {col2}",
"coefficient": val,
"strength": "strong" if abs(val) > 0.7 else "moderate" if abs(val) > 0.4 else "weak",
})
pairs.sort(key=lambda x: abs(x["coefficient"]), reverse=True)
return {
"method": method,
"correlation_matrix": corr_dict,
"top_correlations": pairs[:5],
}
except Exception as e:
return {"error": f"Correlation computation failed: {str(e)}"}
Step 4: Visualization Tools
Now add tools that let the agent create charts and save them as image files. The agent can then describe what it created and point the user to the output file. Create tools/visualization.py:
# tools/visualization.py
import matplotlib
matplotlib.use("Agg") # Non-interactive backend for headless environments
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import os
from typing import Optional
from agents import function_tool
from tools.data_loader import _data_cache
# Output directory for generated charts
CHART_DIR = "charts"
os.makedirs(CHART_DIR, exist_ok=True)
@function_tool
def create_bar_chart(
file_path: str,
x_column: str,
y_column: Optional[str] = None,
title: Optional[str] = None,
top_n: int = 10,
horizontal: bool = False,
) -> dict:
"""Create a bar chart from a dataset and save it as a PNG file.
If y_column is provided, creates a bar chart of y values grouped by x.
If y_column is omitted, creates a frequency count chart of x_column.
Args:
file_path: Path to the loaded CSV file.
x_column: Column for the x-axis (categories).
y_column: Optional column for the y-axis (values). If None, counts are used.
title: Optional chart title (auto-generated if None).
top_n: Limit to top N categories (by frequency or sum).
horizontal: If True, creates a horizontal bar chart.
Returns:
Dictionary with 'output_path', 'chart_type', and 'data_summary'.
"""
if file_path not in _data_cache:
return {"error": f"File not loaded. Call load_csv('{file_path}') first."}
df = _data_cache[file_path]
if x_column not in df.columns:
return {"error": f"Column '{x_column}' not found."}
# Prepare data
if y_column and y_column in df.columns:
grouped = df.groupby(x_column)[y_column].sum().sort_values(ascending=False).head(top_n)
else:
grouped = df[x_column].value_counts().sort_values(ascending=False).head(top_n)
# Create the chart
fig, ax = plt.subplots(figsize=(10, 6))
if horizontal:
bars = ax.barh(grouped.index.astype(str)[::-1], grouped.values[::-1])
ax.set_xlabel(y_column or "Count")
else:
bars = ax.bar(grouped.index.astype(str), grouped.values)
ax.set_ylabel(y_column or "Count")
plt.xticks(rotation=45, ha="right")
chart_title = title or f"{y_column or 'Count'} by {x_column}"
ax.set_title(chart_title)
# Add value labels on bars
for bar in (bars if horizontal else bars):
if horizontal:
ax.text(bar.get_width() + 0.3, bar.get_y() + bar.get_height()/2,
str(int(bar.get_width())), va='center')
else:
ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.3,
str(int(bar.get_height())), ha='center')
plt.tight_layout()
# Save to file
filename = f"bar_{x_column}_{y_column or 'count'}.png"
filename = filename.replace("/", "_").replace("\\", "_")
output_path = os.path.join(CHART_DIR, filename)
plt.savefig(output_path, dpi=150, bbox_inches='tight')
plt.close(fig)
return {
"output_path": output_path,
"chart_type": "horizontal_bar" if horizontal else "bar_chart",
"categories_count": len(grouped),
"data_summary": {str(k): int(v) if pd.notna(v) else 0 for k, v in grouped.items()},
}
@function_tool
def create_scatter_plot(
file_path: str,
x_column: str,
y_column: str,
color_column: Optional[str] = None,
title: Optional[str] = None,
sample_size: int = 5000,
) -> dict:
"""Create a scatter plot from two numeric columns and save as PNG.
Optionally colors points by a third categorical column.
Args:
file_path: Path to the loaded CSV file.
x_column: Column for the x-axis (must be numeric).
y_column: Column for the y-axis (must be numeric).
color_column: Optional categorical column for color encoding.
title: Optional chart title.
sample_size: Maximum points to plot (prevents overcrowding).
Returns:
Dictionary with 'output_path', 'chart_type', and correlation info.
"""
if file_path not in _data_cache:
return {"error": f"File not loaded. Call load_csv('{file_path}') first."}
df = _data_cache[file_path]
# Validate columns
for col in [x_column, y_column]:
if col not in df.columns:
return {"error": f"Column '{col}' not found."}
if not pd.api.types.is_numeric_dtype(df[col]):
return {"error": f"Column '{col}' must be numeric for scatter plot."}
# Sample if needed
if len(df) > sample_size:
df_sample = df.sample(n=sample_size, random_state=42)
else:
df_sample = df
fig, ax = plt.subplots(figsize=(8, 8))
if color_column and color_column in df.columns:
# Use a categorical color palette
categories = df_sample[color_column].astype(str).unique()
palette = sns.color_palette("husl", len(categories))
color_map = {cat: palette[i] for i, cat in enumerate(categories)}
colors = df_sample[color_column].astype(str).map(color_map)
scatter = ax.scatter(
df_sample[x_column], df_sample[y_column],
c=colors.tolist(), alpha=0.6, s=30
)
# Add legend
handles = [plt.Line2D([0], [0], marker='o', color='w',
markerfacecolor=color_map[cat], markersize=8)
for cat in categories[:10]]
ax.legend(handles, list(categories[:10]), title=color_column,
loc='upper right', fontsize=8)
else:
ax.scatter(df_sample[x_column], df_sample[y_column], alpha=0.5, s=30)
chart_title = title or f"{y_column} vs {x_column}"
ax.set_title(chart_title)
ax.set_xlabel(x_column)
ax.set_ylabel(y_column)
plt.tight_layout()
filename = f"scatter_{x_column}_vs_{y_column}.png".replace("/", "_")
output_path = os.path.join(CHART_DIR, filename)
plt.savefig(output_path, dpi=150, bbox_inches='tight')
plt.close(fig)
# Compute correlation for reference
corr = df_sample[[x_column, y_column]].corr().iloc[0, 1]
return {
"output_path": output_path,
"chart_type": "scatter_plot",
"points_plotted": len(df_sample),
"correlation_coefficient": round(corr, 4),
}
Step 5: Assembling the Agent
With all tools defined, we now create the main agent. It receives comprehensive instructions that guide it through the analytical workflow. Create agent.py:
# agent.py
import asyncio
import os
from agents import Agent, Runner, trace
from tools import (
load_csv, get_data_summary,
compute_descriptive_stats, run_correlation,
create_bar_chart, create_scatter_plot,
)
# ─── Agent Definition ───────────────────────────────────────────
data_analyst = Agent(
name="DataAnalyst",
instructions="""You are an expert data analyst agent. Your job is to help users
explore, analyze, and visualize their datasets.
## Workflow
When a user asks a question about a dataset:
1. **Load the data first**: Always call `load_csv` before any other operation.
If the user hasn't specified a file path, ask them for it.
2. **Explore the structure**: After loading, use `get_data_summary` to understand
column types, distributions, and missing values. Summarize what you find.
3. **Answer the question**: Use the appropriate tool(s) to answer the user's query:
- For questions about specific columns: use `compute_descriptive_stats`
- For relationships between columns: use `run_correlation`
- For visualizations: use `create_bar_chart` or `create_scatter_plot`
4. **Interpret results**: Don't just dump raw JSON. Explain what the numbers mean
in plain language. Highlight interesting findings, trends, or anomalies.
5. **Create visualizations proactively**: Whenever a chart would help illustrate
your point, create one. Mention the output file path so the user can view it.
## Important Rules
- Always verify that `load_csv` succeeded before calling other tools.
- If a tool returns an error, explain it to the user and suggest a fix.
- When creating charts, explain what the chart shows and why it's relevant.
- For correlation results, interpret the strength and direction of relationships.
- Be concise but thorough. Prioritize actionable insights over raw data.
""",
tools=[
load_csv,
get_data_summary,
compute_descriptive_stats,
run_correlation,
create_bar_chart,
create_scatter_plot,
],
model="gpt-4o",
)
# ─── Interactive Runner ─────────────────────────────────────────
async def main():
"""Run the data analyst agent interactively from the terminal."""
print("=" * 60)
print(" Data Analysis Agent - Powered by OpenAI Agents SDK")
print("=" * 60)
print("\nType 'quit' to exit, or start by describing your dataset.\n")
conversation_id = None
while True:
user_input = input("\nYou: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("\nGoodbye!")
break
print("\nAnalyzing...", end="", flush=True)
try:
result = await Runner.run(
starting_agent=data_analyst,
input=user_input,
conversation_id=conversation_id,
)
conversation_id = result.conversation_id
print("\r" + " " * 20 + "\r", end="") # Clear "Analyzing..."
print(f"Agent: {result.final_output}")
except Exception as e:
print(f"\nError: {e}")
if __name__ == "__main__":
asyncio.run(main())
Step 6: Testing with Sample Data
Create a sample CSV file to test your agent. Save this as sales_data.csv in your project root:
date,product,category,units_sold,unit_price,region
2024-01-01,Widget A,Electronics,120,45.99,North
2024-01-02,Widget B,Electronics,85,62.50,South
2024-01-03,Gadget X,Home Goods,200,12.75,East
2024-01-04,Widget A,Electronics,95,45.99,West
2024-01-05,Gadget Y,Home Goods,150,18.90,North
2024-01-06,Widget B,Electronics,110,62.50,South
2024-01-07,Gadget X,Home Goods,175,12.75,East
2024-01-08,Widget A,Electronics,130,45.99,West
2024-01-09,Gadget Y,Home Goods,90,18.90,North
2024-01-10,Widget B,Electronics,140,62.50,South
2024-01-11,Gadget X,Home Goods,210,12.75,East
2024-01-12,Widget A,Electronics,105,45.99,North
2024-01-13,Gadget Y,Home Goods,160,18.90,South
2024-01-14,Widget B,Electronics,75,62.50,East
2024-01-15,Gadget X,Home Goods,195,12.75,West
2024-01-16,Widget A,Electronics,115,45.99,North
2024-01-17,Gadget Y,Home Goods,135,18.90,South
2024-01-18,Widget B,Electronics,125,62.50,East
2024-01-19,Gadget X,Home Goods,180,12.75,West
2024-01-20,Widget A,Electronics,100,45.99,North
2024-01-21,Gadget Y,Home Goods,145,18.90,South
2024-01-22,Widget B,Electronics,155,62.50,East
2024-01-23,Gadget X,Home Goods,190,12.75,West
2024-01-24,Widget A,Electronics,108,45.99,North
2024-01-25,Gadget Y,Home Goods,170,18.90,South
Run the agent and try queries like:
# Start the agent
python agent.py
# Example queries:
# "Load sales_data.csv and tell me what's in it."
# "What's the average units sold for each product?"
# "Show me the correlation between units_sold and unit_price."
# "Create a bar chart of total units sold by product."
# "Make a scatter plot of unit_price vs units_sold colored by region."
Advanced: Multi-Agent Architecture with Handoffs
For more complex analysis pipelines, you can split responsibilities across multiple specialized agents. Here's how to refactor the system to use handoffs for statistical modeling and visualization:
# advanced_agent.py
from agents import Agent, handoff, Runner
from tools import (
load_csv, get_data_summary,
compute_descriptive_stats, run_correlation,
create_bar_chart, create_scatter_plot,
)
# ─── Specialized Sub-Agents ─────────────────────────────────────
statistics_agent = Agent(
name="StatisticsExpert",
instructions="""You are a statistics expert. Perform deep statistical analysis.
Use compute_descriptive_stats for individual columns and run_correlation for
relationships. Interpret results with statistical significance context.
Always explain what the numbers mean in practical terms.""",
tools=[compute_descriptive_stats, run_correlation],
model="gpt-4o",
)
visualization_agent = Agent(
name="VisualizationExpert",
instructions="""You are a data visualization expert. Create clear, informative
charts that answer the user's questions. Use create_bar_chart for comparisons
and create_scatter_plot for relationships. After creating a chart, describe
what it reveals and provide the file path.""",
tools=[create_bar_chart, create_scatter_plot],
model="gpt-4o",
)
# ─── Orchestrator Agent ─────────────────────────────────────────
orchestrator = Agent(
name="DataAnalysisOrchestrator",
instructions="""You are the lead data analyst. Your role is to:
1. Load data with load_csv and explore it with get_data_summary.
2. Understand what the user wants.
3. For statistical questions: hand off to StatisticsExpert.
4. For visualization requests: hand off to VisualizationExpert.
5. For complex queries requiring both: coordinate by first getting statistics,
then requesting visualizations based on findings.
Always summarize findings from sub-agents for the user in clear language.""",
tools=[load_csv, get_data_summary],
handoffs=[
handoff(statistics_agent),
handoff(visualization_agent),
],
model="gpt-4o",
)
async def run_orchestrated():
result = await Runner.run(
starting_agent=orchestrator,
input="Load sales_data.csv, analyze the correlation between price and units sold, then visualize it.",
)
print(result.final_output)
# asyncio.run(run_orchestrated())
This multi-agent design offers several advantages:
- Separation of concerns — each agent focuses on its domain, leading to more reliable tool selection
- Better error recovery — if visualization fails, the statistics agent's work is preserved
- Parallel potential — in future SDK versions, independent handoffs could run concurrently
- Smaller prompts
— Ad —
Google AdSense will appear here after approval