A Code Implementation To Assemble A Multi-Agent Evaluation System With OpenAI Brokers, Function Devices, Handoffs, And Session Memory

!pip arrange openai-agents python-dotenv


import asyncio
import json
from datetime import datetime
from brokers import Agent, Runner, function_tool, SQLiteSession
import os


os.environ['OPENAI_API_KEY'] = 'Use Your Private API Key'

@function_tool
def web_search(query: str, max_results: int = 3) -> str:
   """Simulate web search outcomes for demonstration"""
   outcomes = [
       f"Result 1 for '{query}': Latest findings show significant developments...",
       f"Result 2 for '{query}': Research indicates new approaches in this field...",
       f"Result 3 for '{query}': Expert analysis suggests important implications..."
   ]
   return f"Search outcomes for '{query}':n" + "n".be part of(outcomes[:max_results])


@function_tool
def analyze_data(info: str, analysis_type: str = "summary") -> str:
   """Analyze supplied info with fully completely different analysis varieties"""
   analyses = {
       "summary": f"Summary: The data incorporates {len(info.reduce up())} key elements with major themes spherical innovation and effectivity.",
       "detailed": f"Detailed Analysis: Breaking down the {len(info)} characters of information reveals patterns in methodology and conclusions.",
       "traits": f"Sample Analysis: Current info suggests upward trajectory with 3 primary inflection elements acknowledged."
   }
   return analyses.get(analysis_type, "Analysis full: Customary evaluation carried out.")


@function_tool
def save_research(title: str, content material materials: str, class: str = "primary") -> str:
   """Save evaluation findings to a structured format"""
   timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
   research_entry = {
       "title": title,
       "content material materials": content material materials,
       "class": class,
       "timestamp": timestamp,
       "id": f"research_{len(content material materials) % 1000}"
   }
   return f"✅ Evaluation saved: '{title}' at school '{class}' at {timestamp}"

research_agent = Agent(
   determine="Evaluation Specialist",
   instructions="""You are an educated researcher who:
   - Conducts thorough web searches on any topic
   - Analyzes information critically and objectively
   - Identifies key insights and patterns
   - On a regular basis makes use of devices to gather and analyze info sooner than responding""",
   devices=[web_search, analyze_data]
)


analyst_agent = Agent(
   determine="Data Analyst",
   instructions="""You are a senior info analyst who:
   - Takes evaluation findings and performs deep analysis
   - Identifies traits, patterns, and actionable insights
   - Creates structured summaries and ideas
   - Makes use of analysis devices to spice up understanding""",
   devices=[analyze_data, save_research]
)


coordinator_agent = Agent(
   determine="Evaluation Coordinator",
   instructions="""You are a evaluation coordinator who:
   - Manages multi-step evaluation initiatives
   - Delegates duties to relevant specialists
   - Synthesizes findings from a variety of sources
   - Makes final decisions on evaluation path
   - Handoff to research_agent for preliminary info gathering
   - Handoff to analyst_agent for detailed analysis""",
   handoffs=[research_agent, analyst_agent],
   devices=[save_research]
)

async def run_advanced_research_workflow():
   """Demonstrates a complete multi-agent evaluation workflow"""
  
   session = SQLiteSession("research_session_001")
  
   print("🚀 Starting Superior Multi-Agent Evaluation System")
   print("=" * 60)
  
   research_topic = "artificial intelligence in healthcare 2024"
  
   print(f"n📋 PHASE 1: Initiating evaluation on '{research_topic}'")
   result1 = await Runner.run(
       coordinator_agent,
       f"I need full evaluation on '{research_topic}'. Please coordinate a full evaluation workflow along with info gathering, analysis, and final report period.",
       session=session
   )
   print(f"Coordinator Response: {result1.final_output}")
  
   print(f"n📊 PHASE 2: Requesting detailed growth analysis")
   result2 = await Runner.run(
       coordinator_agent,
       "Primarily based totally on the sooner evaluation, I need an in depth growth analysis specializing in rising alternate options and potential challenges. Save the final word analysis for future reference.",
       session=session
   )
   print(f"Analysis Response: {result2.final_output}")
  
   print(f"n🔬 PHASE 3: Direct specialist analysis")
   result3 = await Runner.run(
       analyst_agent,
       "Perform an in depth analysis of the healthcare AI market, specializing in regulatory challenges and market alternate options. Categorize this as 'market_analysis'.",
       session=session
   )
   print(f"Specialist Response: {result3.final_output}")
  
   print("n✅ Evaluation workflow achieved effectively!")
   return result1, result2, result3


async def run_focused_analysis():
   """Displays focused single-agent capabilities"""
  
   print("n🎯 FOCUSED ANALYSIS DEMO")
   print("-" * 40)
  
   consequence = await Runner.run(
       research_agent,
       "Evaluation in quantum computing and analyze the vital factor breakthroughs from 2024.",
       max_turns=5
   )
  
   print(f"Focused Analysis Finish outcome: {consequence.final_output}")
   return consequence


def quick_research_sync(topic: str):
   """Synchronous evaluation for quick queries"""
  
   print(f"n⚡ QUICK SYNC RESEARCH: {topic}")
   print("-" * 40)
  
   consequence = Runner.run_sync(
       research_agent,
       f"Shortly evaluation {topic} and provide 3 key insights."
   )
  
   print(f"Quick Finish outcome: {consequence.final_output}")
   return consequence

async def major():
   """Predominant carry out demonstrating all capabilities"""
  
   print("🤖 OpenAI Brokers SDK - Superior Tutorial")
   print("Establishing a Multi-Agent Evaluation System")
   print("=" * 60)
  
   attempt:
       await run_advanced_research_workflow()
      
       await run_focused_analysis()
      
       quick_research_sync("blockchain adoption in enterprise")
      
       print("n🎉 Tutorial achieved effectively!")
       print("nKey Choices Demonstrated:")
       print("✅ Multi-agent coordination with handoffs")
       print("✅ Personalized carry out devices")
       print("✅ Session memory for dialog continuity")
       print("✅ Async and sync execution patterns")
       print("✅ Structured workflows with max_turns administration")
       print("✅ Specialised agent roles and capabilities")
      
   moreover Exception as e:
       print(f"❌ Error: {e}")
       print("nTroubleshooting concepts:")
       print("- Assure OPENAI_API_KEY is about appropriately")
       print("- Study net connection")
       print("- Verify openai-agents package deal deal is put in")


if __name__ == "__main__":
   import nest_asyncio
   nest_asyncio.apply()
  
   asyncio.run(major())


def create_custom_agent(determine: str, place: str, tools_list: document = None):
   """Helper carry out to create custom-made brokers shortly"""
   return Agent(
       determine=determine,
       instructions=f"You are a {place} who provides educated assist.",
       devices=tools_list or []
   )


custom_agent = create_custom_agent("Code Reviewer", "senior software program program engineer", [analyze_data])
consequence = Runner.run_sync(custom_agent, "Overview this Python code for best practices")


print("n📚 Tutorial Notes:")
print("- Modify evaluation issues and agent instructions to find fully completely different use cases")
print("- Add your private custom-made devices using the @function_tool decorator")
print("- Experiment with fully completely different agent handoff patterns")
print("- Use courses for multi-turn conversations")
print("- Glorious for Colab - merely add your OpenAI API key and run!")