A Deep Dive Into The OpenAI Agents SDK

from openai.agents import Agent

researcher = Agent(
    name="Customer Support Agent",
    model="gpt-4o",
    instructions="Help users without saying 'have you tried turning it off and on again?",
    tools=[web_search, document_retrieval]
)

from agents import Agent, ModelSettings

# Fast, cost-effective agent for simple tasks
quick_agent = Agent(
    name="Quick Responder",
    model="gpt-4o-mini",  # Fastest, most cost-effective
    instructions="Provide quick, helpful responses to simple questions.",
    model_settings=ModelSettings(
        temperature=0.3,  # Lower temperature for consistent responses
        max_tokens=150    # Limit response length for quick interactions
    )
)

# Powerful agent for complex reasoning
reasoning_agent = Agent(
    name="Deep Thinker",
    model="o1-preview",  # Most capable for complex reasoning
    instructions="Provide thorough analysis and step-by-step reasoning for complex problems.",
    model_settings=ModelSettings(
        temperature=0.7,  # Moderate temperature for balanced creativity
        # Note: o1 models don't support max_tokens in the same way
    )
)

# Creative agent for content generation
creative_agent = Agent(
    name="Creative Writer",
    model="gpt-4o",  # Good balance of capability and speed
    instructions="Write engaging, creative content with vivid descriptions.",
    model_settings=ModelSettings(
        temperature=0.9,  # Higher temperature for more creativity
        top_p=0.9,       # Use nucleus sampling for variety
        presence_penalty=0.1  # Encourage topic diversity
    )
)

# Define the structure you want the agent to return
# Pydantic models provide automatic validation and clear documentation
class WeatherReport(BaseModel):
    city: str
    temperature: int  # Temperature in Celsius
    conditions: str   # Weather conditions (sunny, cloudy, etc.)
    recommendation: str  # What the user should do based on the weather

# Create an agent that returns structured data instead of free-form text
weather_agent = Agent(
    name="Weather Reporter",
    instructions="""
    You are a weather reporting agent that provides structured weather information.
    
    When users ask about weather, use the get_weather tool and then format your response
    to include all required fields:
    - city: The city name (cleaned up and properly formatted)
    - temperature: The temperature as an integer
    - conditions: Brief weather description
    - recommendation: Practical advice based on the weather
    
    Be helpful and specific in your recommendations.
    """,
    output_type=WeatherReport,  # This ensures the response matches our structure
    tools=[get_weather]
)

from openai.agents import Runner<br><br>runner = Runner()
result = runner.run(researcher, "What are the latest developments in quantum computing?")
print(result.final_output)

from agents.tool import function_tool

@function_tool
def search_knowledge_base(query: str) -> str:
    # Your code to search a database
    return "Here's what I found about " + query

from agents import function_tool, Agent
import requests
from typing import Dict, Any

# Example 1: External API Integration
# This shows how to wrap external services as agent tools
@function_tool
def search_wikipedia(query: str) -> str:
    """Search Wikipedia for information about a topic.
    
    The agent will see this docstring and understand what this tool does.
    Clear descriptions help the agent choose the right tool for the task.
    
    Args:
        query: The search term to look up on Wikipedia.
        
    Returns:
        str: A summary of the Wikipedia article.
    """
    try:
        # Clean up the query for URL usage
        clean_query = query.replace(" ", "_")
        
        # Call Wikipedia's REST API - this is a real API call
        response = requests.get(
            f"https://en.wikipedia.org/api/rest_v1/page/summary/{clean_query}",
            headers={'User-Agent': 'OpenAI-Agents-Demo/1.0'}
        )
        
        if response.status_code == 200:
            data = response.json()
            # Format the response in a user-friendly way
            return f"**{data['title']}**: {data['extract']}"
        elif response.status_code == 404:
            return f"Sorry, I couldn't find a Wikipedia article for '{query}'. Try rephrasing your search."
        else:
            return f"Wikipedia search temporarily unavailable (status: {response.status_code})"
    
    except Exception as e:
        # Always handle errors gracefully in tools
        return f"Error searching Wikipedia: {str(e)}"

# Example 2: Complex Calculations with Structured Output
# This demonstrates how to return structured data from tools
@function_tool
def calculate_compound_interest(
    principal: float, 
    rate: float, 
    time: int, 
    compound_frequency: int = 12
) -> Dict[str, Any]:
    """Calculate compound interest with detailed breakdown.
    
    This tool shows how to handle multiple parameters and return structured data
    that the agent can interpret and present clearly to users.
    
    Args:
        principal: Initial amount invested (in dollars).
        rate: Annual interest rate as a percentage (e.g., 5.5 for 5.5%).
        time: Number of years to calculate.
        compound_frequency: How often interest compounds per year (default: 12 for monthly).
        
    Returns:
        dict: Calculation results including final amount and total interest.
    """
    # Input validation - important for tools that do calculations
    if principal <= 0:
        return {"error": "Principal amount must be positive"}
    if rate < 0:
        return {"error": "Interest rate cannot be negative"}
    if time <= 0:
        return {"error": "Time period must be positive"}
    if compound_frequency <= 0:
        return {"error": "Compound frequency must be positive"}
    
    # Convert percentage to decimal
    decimal_rate = rate / 100
    
    # Apply the compound interest formula: A = P(1 + r/n)^(nt)
    final_amount = principal * (1 + decimal_rate/compound_frequency) ** (compound_frequency * time)
    total_interest = final_amount - principal
    
    # Return structured data that the agent can easily work with
    return {
        "status": "success",
        "calculation": {
            "principal": principal,
            "final_amount": round(final_amount, 2),
            "total_interest": round(total_interest, 2),
            "rate_percent": rate,
            "years": time,
            "compound_frequency": compound_frequency
        },
        "summary": f"${principal:,.2f} grows to ${final_amount:,.2f} over {time} years at {rate}% annual interest"
    }

from agents import Agent, WebSearchTool, FileSearchTool, CodeInterpreterTool

# Create specialized agents that excel at specific tasks
# Each agent has focused expertise and tailored instructions

# Math specialist - optimized for numerical calculations and explanations
math_agent = Agent(
    name="Math Specialist",
    instructions="""
    You are an expert mathematician who excels at:
    - Solving complex mathematical problems step by step
    - Explaining mathematical concepts clearly
    - Performing accurate calculations
    - Showing your work and reasoning
    
    Always break down complex problems into understandable steps.
    When doing calculations, be precise and show intermediate results.
    """,
    tools=[calculate_compound_interest]  # From our earlier example
)

# Research specialist - optimized for information gathering
research_agent = Agent(
    name="Research Specialist", 
    instructions="""
    You are an expert researcher who excels at:
    - Finding authoritative and current information
    - Synthesizing information from multiple sources
    - Providing well-sourced and balanced perspectives
    - Distinguishing between facts and opinions
    
    Always cite your sources and explain the credibility of information.
    When possible, find multiple sources to verify important claims.
    """,
    tools=[search_wikipedia, WebSearchTool()]
)

# Create an orchestrator agent that coordinates the specialists
# This agent decides which specialist to use based on the user's needs
orchestrator = Agent(
    name="Task Orchestrator",
    instructions="""
    You are a smart coordinator that manages a team of specialist agents.
    Your job is to understand user requests and delegate to the right specialist.
    
    Use the math specialist for:
    - Numerical calculations and mathematical problems
    - Financial analysis and projections
    - Statistical analysis
    - Any task involving numbers or formulas
    
    Use the research specialist for:
    - Finding information about topics, people, or events
    - Gathering current news or developments
    - Researching background information
    - Fact-checking and verification
    
    For complex tasks, you can use both specialists:
    1. First gather information with the research specialist
    2. Then perform calculations with the math specialist
    3. Synthesize the results into a comprehensive answer
    
    Always explain which specialist you're consulting and why.
    """,
    tools=[
        # Convert agents into tools using the as_tool() method
        # This allows the orchestrator to call them like any other tool
        math_agent.as_tool(
            tool_name="get_math_help",
            tool_description="Get expert help with mathematical calculations, formulas, and numerical analysis"
        ),
        research_agent.as_tool(
            tool_name="get_research_help", 
            tool_description="Get expert help researching topics, finding information, and gathering current data"
        )
    ]
)

from dataclasses import dataclass
from agents import Agent, RunContextWrapper, function_tool
import time

# Define a context class that holds all the user session information
# This will be passed to every agent and tool in your system
@dataclass
class UserSession:
    user_id: str
    name: str
    preferences: dict
    conversation_history: list
    session_start_time: float

@function_tool
def get_user_preferences(ctx: RunContextWrapper[UserSession]) -> str:
    """Get the current user's preferences.
    
    The ctx parameter gives us access to the user session data.
    This allows tools to personalize their behavior based on user settings.
    """
    prefs = ctx.context.preferences
    if not prefs:
        return "No preferences set yet. You can update them anytime!"
    
    # Format preferences in a user-friendly way
    formatted_prefs = []
    for category, value in prefs.items():
        formatted_prefs.append(f"{category}: {value}")
    
    return f"Your current preferences: {', '.join(formatted_prefs)}"

@function_tool
def update_preference(ctx: RunContextWrapper[UserSession], category: str, value: str) -> str:
    """Update a user preference.
    
    This tool modifies the context data, which persists throughout the session.
    Changes made here will be available to all subsequent tool calls and agent interactions.
    """
    # Update the preferences in the context
    ctx.context.preferences[category] = value
    
    # Add to conversation history for tracking
    ctx.context.conversation_history.append({
        "action": "preference_update",
        "category": category,
        "value": value,
        "timestamp": time.time()
    })
    
    return f"Updated your {category} preference to '{value}'. This will apply to all future interactions!"

@function_tool
def get_conversation_summary(ctx: RunContextWrapper[UserSession]) -> str:
    """Provide a summary of the user's session activity.
    
    This demonstrates how context can be used to track state across multiple interactions.
    """
    session = ctx.context
    session_duration = time.time() - session.session_start_time
    
    summary = f"""
    Session Summary for {session.name}:
    - Session duration: {session_duration/60:.1f} minutes
    - Conversation events: {len(session.conversation_history)}
    - Current preferences: {len(session.preferences)} set
    """
    
    # Show recent activity if available
    if session.conversation_history:
        recent_activity = session.conversation_history[-3:]  # Last 3 events
        summary += "\n\nRecent activity:"
        for event in recent_activity:
            if event.get("action") == "preference_update":
                summary += f"\n- Set {event['category']} to {event['value']}"
    
    return summary

personalized_agent = Agent[UserSession](
    name="Personal Assistant",
    instructions="""
    You are a personal assistant that provides personalized service based on user preferences.
    
    You can:
    - Check and update user preferences using the available tools
    - Provide personalized recommendations based on their settings
    - Track conversation history and provide session summaries
    
    Key behaviors:
    - Always use the user's name when you know it
    - Reference their preferences when giving advice
    - Offer to update preferences when you notice user needs
    - Be helpful and remember context from earlier in the conversation
    """,
    tools=[get_user_preferences, update_preference, get_conversation_summary]
)

from agents import Agent, Runner, trace

# Tracing is enabled by default, but you can customize it
async def traced_research_session():
    with trace(workflow_name="Research Session", group_id="session_123") as trace_context:
        # First research query
        result1 = await Runner.run(
            agent, 
            "Research the history of artificial intelligence",
            run_config=RunConfig(
                workflow_name="AI History Research",
                trace_metadata={"user_id": "user_456", "session_type": "research"}
            )
        )
        
        # Follow-up query in the same trace
        result2 = await Runner.run(
            agent,
            f"Based on this research: {result1.final_output}, what are the key milestones?"
        )
        
    # Traces are automatically sent to OpenAI dashboard
    print("Research complete. Check traces at https://platform.openai.com/traces")

# You can also disable tracing or customize processors
from agents import set_tracing_disabled, add_trace_processor

# Disable tracing entirely
set_tracing_disabled(True)

# Or add custom trace processors for external tools
# add_trace_processor(your_custom_processor)

support_agent = Agent(name="support", instructions="You handle customer questions.")
technical_agent = Agent(name="technical", instructions="You solve technical issues.")
billing_agent = Agent(name="billing", instructions="You handle billing inquiries.")

triage_agent = Agent(
    name="triage",
    instructions="Route customer inquiries to the appropriate specialized agent.",
    handoffs=[support_agent, technical_agent, billing_agent]
)

from agents.guardrails import CustomGuardrail

async def is_not_swearing(msgs, context) -> bool:
    content = " ".join(m["content"] for m in msgs if "content" in m)
    return "badword" not in content.lower()

my_guardrail = CustomGuardrail(
    guardrail_function=is_not_swearing,
    tripwire_config=lambda output: not output  # if 'False', raise error
)

agent = Agent(
    name="my_agent",
    input_guardrails=[my_guardrail]
)

class ResearchFinding(BaseModel):
    """A single research finding with source information."""
    statement: str
    source: str
    confidence: float  # 0.0 to 1.0


class VerifiedResearch(BaseModel):
    """Collection of verified research findings."""
    findings: List[ResearchFinding]
    verified: bool
    notes: Optional[str] = None


class FinalContent(BaseModel):
    """Final output content with structured sections."""
    title: str
    introduction: str
    key_points: List[str]
    body: str
    conclusion: str
    sources: List[str]

@function_tool
async def search_web(context: AgentContextWrapper[ResearchContext], query: str) -> str:
    """
    Search the web for information about a topic using the Exa Search API.
    
    Args:
        query: The search query text
        
    Returns:
        Search results as formatted text with citations
    """
    topic = context.agent_context.topic
    # Combine the specific query with the general topic for better results
    full_query = f"{query} about {topic}"
    
    try:
        # Make a request to the Exa Search API
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.exa.ai/search",
                headers={
                    "Content-Type": "application/json",
                    "x-api-key": "YOUR_EXA_API_KEY"  # Replace with your actual API key
                },
                json={
                    "query": full_query,
                    "numResults": 5,
                    "useAutoprompt": True,
                    "type": "keyword"
                }
            ) as response:
                if response.status != 200:
                    error_text = await response.text()
                    return f"Error searching: {response.status} - {error_text}"
                
                search_results = await response.json()
    
        # Process the results
        formatted_results = f"Search results for '{query}' about {topic}:\n\n"
        
        if not search_results.get("results"):
            return f"No results found for '{query}' about {topic}."
        
        # Format each result with its title, content, and URL
        for i, result in enumerate(search_results.get("results", []), 1):
            title = result.get("title", "No title")
            url = result.get("url", "No URL")
            content = result.get("text", "").strip()
            
            # Limit content length for readability
            if len(content) > 500:
                content = content[:500] + "..."
            
            formatted_results += f"{i}. **{title}**\n"
            formatted_results += f"   {content}\n"
            formatted_results += f"   Source: {url}\n\n"
        
        # Add a summary if available
        if search_results.get("autopromptString"):
            formatted_results += f"Summary: {search_results.get('autopromptString')}\n\n"
            
        return formatted_results
    
    except Exception as e:
        # Provide a useful error message
        error_message = f"Error while searching for '{query}': {str(e)}"
        
        # Add fallback information if the search fails
        fallback_info = f"\n\nFallback information about {topic}:\n" + \
                        f"1. {topic} has been studied in recent publications.\n" + \
                        f"2. Current research suggests growing interest in {topic}.\n" + \
                        f"3. Common challenges in {topic} include implementation complexity and adoption barriers."
        
        return error_message + fallback_info

class ResearchContext:
    def __init__(self, topic: str):
        self.topic = topic
        self.findings = []
        self.verified_findings = []
        self.draft_content = ""
        self.history = []
        
    def add_finding(self, finding: ResearchFinding):
        self.findings.append(finding)
        self.history.append(f"Added finding: {finding.statement}")
        
    def add_verified_findings(self, verified: VerifiedResearch):
        self.verified_findings.extend(verified.findings)
        self.history.append(f"Added {len(verified.findings)} verified findings")
        
    def set_draft(self, draft: str):
        self.draft_content = draft
        self.history.append("Updated draft content")

async def is_fact_based(msgs, context) -> bool:
    """Check if messages appear to be fact-based and not opinion-heavy."""
    content = " ".join(m.get("content", "") for m in msgs if isinstance(m, dict))
    opinion_phrases = ["I believe", "I think", "in my opinion", "probably", "might be", "could be"]
    
    # Count opinion phrases
    opinion_count = sum(content.lower().count(phrase) for phrase in opinion_phrases)
    
    # Allow some opinion phrases, but not too many
    return opinion_count < 3


fact_based_guardrail = CustomGuardrail(
    guardrail_function=is_fact_based,
    tripwire_config=lambda output: not output,
    error_message="Output contains too many opinion statements rather than fact-based research."
)

fact_checker_agent = Agent(
    name="fact_checker_agent",
    model="gpt-4o",
    instructions="""You are a meticulous fact-checking agent. Your job is to:
    1. Review the research findings in the shared context
    2. Verify each statement using the verify_statement tool
    3. Consolidate verified findings using save_verified_research
    4. Be skeptical and thorough - only approve statements with sufficient evidence
    
    For each finding, check if the source is credible and if the statement contains verifiable
    facts rather than opinions or generalizations.
    """,
    context_type=ResearchContext,
    tools=[verify_statement, save_verified_research],
    output_type=str,
    output_guardrails=[fact_based_guardrail],
    description="Verifies research findings for accuracy and proper sourcing"
)

triage_agent = Agent(
    name="triage_agent",
    model="gpt-3.5-turbo",
    instructions="""You are a research coordinator who manages the research process.
    For any research query:
    1. First, hand off to the researcher_agent to gather information
    2. Then, hand off to the fact_checker_agent to verify the findings
    3. Finally, hand off to the writer_agent to create the final content
    
    Monitor the process and ensure each specialized agent completes their task.
    """,
    context_type=ResearchContext,
    handoffs=[
        handoff(researcher_agent),
        handoff(fact_checker_agent),
        handoff(writer_agent)
    ],
    output_type=FinalContent,
    description="Coordinates the research process across specialized agents"
)

async def run_research_system(topic: str) -> FinalContent:
    """Run the multi-agent research system on a given topic."""
    # Create the shared context
    context = ResearchContext(topic=topic)
    
    # Configure the run with tracing enabled
    config = AgentRunConfig(
        run_name=f"research_{topic.replace(' ', '_')}",
        tracing_disabled=False
    )
    
    # Run the triage agent with the initial query
    result = await AgentRunner.run(
        triage_agent,
        [f"Research the following topic thoroughly: {topic}"],
        context=context,
        run_config=config
    )
    
    return result.agent_output