A Code Implementation to Construct a Multi-Agent Analysis System with OpenAI Brokers, Operate Instruments, Handoffs, and Session Reminiscence

!pip set up 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 Personal API Key'

@function_tool
def web_search(question: str, max_results: int = 3) -> str:
   """Simulate internet 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 '{question}':n" + "n".be part of(outcomes[:max_results])


@function_tool
def analyze_data(information: str, analysis_type: str = "abstract") -> str:
   """Analyze offered information with completely different evaluation sorts"""
   analyses = {
       "abstract": f"Abstract: The info incorporates {len(information.cut up())} key factors with predominant themes round innovation and effectivity.",
       "detailed": f"Detailed Evaluation: Breaking down the {len(information)} characters of information reveals patterns in methodology and conclusions.",
       "developments": f"Development Evaluation: Present information suggests upward trajectory with 3 main inflection factors recognized."
   }
   return analyses.get(analysis_type, "Evaluation full: Normal analysis carried out.")


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

research_agent = Agent(
   title="Analysis Specialist",
   directions="""You're an knowledgeable researcher who:
   - Conducts thorough internet searches on any matter
   - Analyzes data critically and objectively
   - Identifies key insights and patterns
   - At all times makes use of instruments to assemble and analyze information earlier than responding""",
   instruments=[web_search, analyze_data]
)


analyst_agent = Agent(
   title="Information Analyst",
   directions="""You're a senior information analyst who:
   - Takes analysis findings and performs deep evaluation
   - Identifies developments, patterns, and actionable insights
   - Creates structured summaries and proposals
   - Makes use of evaluation instruments to boost understanding""",
   instruments=[analyze_data, save_research]
)


coordinator_agent = Agent(
   title="Analysis Coordinator",
   directions="""You're a analysis coordinator who:
   - Manages multi-step analysis initiatives
   - Delegates duties to applicable specialists
   - Synthesizes findings from a number of sources
   - Makes ultimate choices on analysis path
   - Handoff to research_agent for preliminary information gathering
   - Handoff to analyst_agent for detailed evaluation""",
   handoffs=[research_agent, analyst_agent],
   instruments=[save_research]
)

async def run_advanced_research_workflow():
   """Demonstrates a whole multi-agent analysis workflow"""
  
   session = SQLiteSession("research_session_001")
  
   print("🚀 Beginning Superior Multi-Agent Analysis System")
   print("=" * 60)
  
   research_topic = "synthetic intelligence in healthcare 2024"
  
   print(f"n📋 PHASE 1: Initiating analysis on '{research_topic}'")
   result1 = await Runner.run(
       coordinator_agent,
       f"I want complete analysis on '{research_topic}'. Please coordinate a full analysis workflow together with information gathering, evaluation, and ultimate report technology.",
       session=session
   )
   print(f"Coordinator Response: {result1.final_output}")
  
   print(f"n📊 PHASE 2: Requesting detailed development evaluation")
   result2 = await Runner.run(
       coordinator_agent,
       "Based mostly on the earlier analysis, I want an in depth development evaluation specializing in rising alternatives and potential challenges. Save the ultimate evaluation for future reference.",
       session=session
   )
   print(f"Evaluation Response: {result2.final_output}")
  
   print(f"n🔬 PHASE 3: Direct specialist evaluation")
   result3 = await Runner.run(
       analyst_agent,
       "Carry out an in depth evaluation of the healthcare AI market, specializing in regulatory challenges and market alternatives. Categorize this as 'market_analysis'.",
       session=session
   )
   print(f"Specialist Response: {result3.final_output}")
  
   print("n✅ Analysis workflow accomplished efficiently!")
   return result1, result2, result3


async def run_focused_analysis():
   """Exhibits targeted single-agent capabilities"""
  
   print("n🎯 FOCUSED ANALYSIS DEMO")
   print("-" * 40)
  
   end result = await Runner.run(
       research_agent,
       "Analysis in quantum computing and analyze the important thing breakthroughs from 2024.",
       max_turns=5
   )
  
   print(f"Centered Evaluation End result: {end result.final_output}")
   return end result


def quick_research_sync(matter: str):
   """Synchronous analysis for fast queries"""
  
   print(f"n⚡ QUICK SYNC RESEARCH: {matter}")
   print("-" * 40)
  
   end result = Runner.run_sync(
       research_agent,
       f"Shortly analysis {matter} and supply 3 key insights."
   )
  
   print(f"Fast End result: {end result.final_output}")
   return end result

async def predominant():
   """Essential operate demonstrating all capabilities"""
  
   print("🤖 OpenAI Brokers SDK - Superior Tutorial")
   print("Constructing a Multi-Agent Analysis System")
   print("=" * 60)
  
   attempt:
       await run_advanced_research_workflow()
      
       await run_focused_analysis()
      
       quick_research_sync("blockchain adoption in enterprise")
      
       print("n🎉 Tutorial accomplished efficiently!")
       print("nKey Options Demonstrated:")
       print("✅ Multi-agent coordination with handoffs")
       print("✅ Customized operate instruments")
       print("✅ Session reminiscence for dialog continuity")
       print("✅ Async and sync execution patterns")
       print("✅ Structured workflows with max_turns management")
       print("✅ Specialised agent roles and capabilities")
      
   besides Exception as e:
       print(f"❌ Error: {e}")
       print("nTroubleshooting ideas:")
       print("- Guarantee OPENAI_API_KEY is ready accurately")
       print("- Verify web connection")
       print("- Confirm openai-agents bundle is put in")


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


def create_custom_agent(title: str, position: str, tools_list: checklist = None):
   """Helper operate to create customized brokers shortly"""
   return Agent(
       title=title,
       directions=f"You're a {position} who supplies knowledgeable help.",
       instruments=tools_list or []
   )


custom_agent = create_custom_agent("Code Reviewer", "senior software program engineer", [analyze_data])
end result = Runner.run_sync(custom_agent, "Evaluate this Python code for finest practices")


print("n📚 Tutorial Notes:")
print("- Modify analysis subjects and agent directions to discover completely different use instances")
print("- Add your personal customized instruments utilizing the @function_tool decorator")
print("- Experiment with completely different agent handoff patterns")
print("- Use periods for multi-turn conversations")
print("- Excellent for Colab - simply add your OpenAI API key and run!")