A Coding Information to Construct Clever Multi-Agent Programs with the PEER Sample

!pip set up agentUniverse google-generativeai python-dotenv pydantic


import os
import asyncio
from typing import Dict, Listing, Any, Non-obligatory
from dataclasses import dataclass
from enum import Enum
import json
import time
import google.generativeai as genai


GEMINI_API_KEY = 'Use Your API Key Right here' 
genai.configure(api_key=GEMINI_API_KEY)

class AgentRole(Enum):
   PLANNER = "planner"
   EXECUTOR = "executor"
   EXPRESSER = "expresser"
   REVIEWER = "reviewer"


@dataclass
class Job:
   id: str
   description: str
   context: Dict[str, Any]
   standing: str = "pending"
   consequence: Non-obligatory[str] = None
   suggestions: Non-obligatory[str] = None


class BaseAgent:
   """Base agent class with core performance"""
   def __init__(self, identify: str, function: AgentRole, system_prompt: str):
       self.identify = identify
       self.function = function
       self.system_prompt = system_prompt
       self.reminiscence: Listing[Dict] = []
  
   async def course of(self, process: Job) -> str:
       immediate = f"{self.system_prompt}nnTask: {process.description}nContext: {json.dumps(process.context)}"
      
       consequence = await self._simulate_llm_call(immediate, process)
      
       self.reminiscence.append({
           "task_id": process.id,
           "enter": process.description,
           "output": consequence,
           "timestamp": time.time()
       })
      
       return consequence
  
   async def _simulate_llm_call(self, immediate: str, process: Job) -> str:
       """Name Google Gemini API for actual LLM processing"""
       strive:
           mannequin = genai.GenerativeModel('gemini-1.5-flash')
          
           enhanced_prompt = self._create_role_prompt(immediate, process)
          
           response = await asyncio.to_thread(
               lambda: mannequin.generate_content(enhanced_prompt)
           )
          
           return response.textual content.strip()
          
       besides Exception as e:
           print(f"⚠️ Gemini API error for {self.function.worth}: {str(e)}")
           return self._get_fallback_response(process)
  
   def _create_role_prompt(self, base_prompt: str, process: Job) -> str:
       """Create enhanced role-specific prompts for Gemini"""
       role_instructions = {
           AgentRole.PLANNER: "You're a strategic planning professional. Create detailed, actionable plans. Break down advanced duties into clear steps with priorities and dependencies.",
           AgentRole.EXECUTOR: "You're a expert executor. Analyze the duty totally and supply detailed implementation insights. Deal with sensible options and potential challenges.",
           AgentRole.EXPRESSER: "You're a skilled communicator. Current info clearly, professionally, and engagingly. Construction your response with headers, bullet factors, and clear conclusions.",
           AgentRole.REVIEWER: "You're a high quality assurance professional. Consider completeness, accuracy, and readability. Present particular, actionable enchancment options."
       }
      
       context_info = f"Earlier context: {json.dumps(process.context, indent=2)}" if process.context else "No earlier context"
      
       return f"""
{role_instructions[self.role]}


{base_prompt}


{context_info}


Job to course of: {process.description}


Present a complete, skilled response applicable on your function as {self.function.worth}.
"""
  
   def _get_fallback_response(self, process: Job) -> str:
       """Fallback responses if Gemini API is unavailable"""
       fallbacks = {
           AgentRole.PLANNER: f"STRATEGIC PLAN for '{process.description}': 1) Requirement evaluation 2) Useful resource evaluation 3) Implementation roadmap 4) Danger mitigation 5) Success metrics",
           AgentRole.EXECUTOR: f"EXECUTION ANALYSIS for '{process.description}': Complete evaluation accomplished. Key findings recognized, sensible options developed, implementation issues famous.",
           AgentRole.EXPRESSER: f"PROFESSIONAL SUMMARY for '{process.description}': ## Evaluation Completenn**Key Insights:** Detailed evaluation performedn**Suggestions:** Strategic actions identifiedn**Subsequent Steps:** Implementation prepared",
           AgentRole.REVIEWER: f"QUALITY REVIEW for '{process.description}': **Evaluation:** Top quality output achieved. **Strengths:** Complete evaluation, clear construction. **Strategies:** Think about extra quantitative metrics."
       }
       return fallbacks[self.role]

class PEERAgent:
   """PEER Sample Implementation - Plan, Execute, Categorical, Assessment"""
   def __init__(self):
       self.planner = BaseAgent("Strategic Planner", AgentRole.PLANNER,
           "You're a strategic planning agent. Break down advanced duties into actionable steps.")
      
       self.executor = BaseAgent("Job Executor", AgentRole.EXECUTOR,
           "You might be an execution agent. Full duties effectively utilizing accessible instruments and data.")
      
       self.expresser = BaseAgent("Consequence Expresser", AgentRole.EXPRESSER,
           "You're a communication agent. Current outcomes clearly and professionally.")
      
       self.reviewer = BaseAgent("High quality Reviewer", AgentRole.REVIEWER,
           "You're a high quality assurance agent. Assessment outputs and supply enchancment suggestions.")
      
       self.iteration_count = 0
       self.max_iterations = 3
  
   async def collaborate(self, process: Job) -> Dict[str, Any]:
       """Execute PEER collaboration sample"""
       outcomes = {"iterations": [], "final_result": None}
      
       whereas self.iteration_count = 1:
               outcomes["final_result"] = expression
               break
              
           self.iteration_count += 1
           process.context["previous_feedback"] = evaluate
      
       return outcomes

class MultiAgentOrchestrator:
   """Orchestrates a number of specialised brokers"""
   def __init__(self):
       self.brokers = {}
       self.peer_system = PEERAgent()
       self.task_queue = []
      
   def register_agent(self, agent: BaseAgent):
       """Register a specialised agent"""
       self.brokers[agent.name] = agent
  
   async def process_complex_task(self, description: str, area: str = "common") -> Dict[str, Any]:
       """Course of advanced process utilizing PEER sample and area brokers"""
       process = Job(
           id=f"task_{int(time.time())}",
           description=description,
           context={"area": area, "complexity": "excessive"}
       )
      
       print(f"🚀 Beginning Complicated Job Processing: {description}")
       print("=" * 60)
      
       peer_results = await self.peer_system.collaborate(process)
      
       if area in ["financial", "technical", "creative"]:
           domain_agent = self._get_domain_agent(area)
           if domain_agent:
               print(f"🔧 Area-Particular Processing ({area})")
               domain_result = await domain_agent.course of(process)
               peer_results["domain_enhancement"] = domain_result
      
       return {
           "task_id": process.id,
           "original_request": description,
           "peer_results": peer_results,
           "standing": "accomplished",
           "processing_time": f"{len(peer_results['iterations'])} iterations"
       }
  
   def _get_domain_agent(self, area: str) -> Non-obligatory[BaseAgent]:
       """Get domain-specific agent with enhanced Gemini prompts"""
       domain_agents = {
           "monetary": BaseAgent("Monetary Analyst", AgentRole.EXECUTOR,
               "You're a senior monetary analyst with experience in market evaluation, threat evaluation, and funding methods. Present detailed monetary insights with quantitative evaluation."),
           "technical": BaseAgent("Technical Professional", AgentRole.EXECUTOR,
               "You're a lead software program architect with experience in system design, scalability, and finest practices. Present detailed technical options with implementation issues."),
           "artistic": BaseAgent("Inventive Director", AgentRole.EXPRESSER,
               "You might be an award-winning artistic director with experience in model technique, content material creation, and modern campaigns. Generate compelling and strategic artistic options.")
       }
       return domain_agents.get(area)


class KnowledgeBase:
   """Easy data administration system"""
   def __init__(self):
       self.data = {
           "financial_analysis": ["Risk assessment", "Portfolio optimization", "Market analysis"],
           "technical_development": ["System architecture", "Code optimization", "Security protocols"],
           "creative_content": ["Brand storytelling", "Visual design", "Content strategy"]
       }
  
   def get_domain_knowledge(self, area: str) -> Listing[str]:
       return self.data.get(area, ["General knowledge"])


async def run_advanced_demo():
    
   orchestrator = MultiAgentOrchestrator()
   knowledge_base = KnowledgeBase()
  
   print("n📊 DEMO 1: Monetary Evaluation with PEER Sample")
   print("-" * 40)
  
   financial_task = "Analyze the potential influence of rising rates of interest on tech shares portfolio"
   result1 = await orchestrator.process_complex_task(financial_task, "monetary")
  
   print(f"n✅ Job Accomplished: {result1['processing_time']}")
   print(f"Closing Consequence: {result1['peer_results']['final_result']}")
  
   print("n💻 DEMO 2: Technical Drawback Fixing")
   print("-" * 40)
  
   technical_task = "Design a scalable microservices structure for a high traffic e-commerce platform"
   result2 = await orchestrator.process_complex_task(technical_task, "technical")
  
   print(f"n✅ Job Accomplished: {result2['processing_time']}")
   print(f"Closing Consequence: {result2['peer_results']['final_result']}")
  
   print("n🎨 DEMO 3: Inventive Content material with Multi-Agent Collaboration")
   print("-" * 40)
  
   creative_task = "Create a complete model technique for a sustainable vogue startup"
   result3 = await orchestrator.process_complex_task(creative_task, "artistic")
  
   print(f"n✅ Job Accomplished: {result3['processing_time']}")
   print(f"Closing Consequence: {result3['peer_results']['final_result']}")
  
   print("n🧠 AGENT MEMORY & LEARNING")
   print("-" * 40)
   print(f"Planner processed {len(orchestrator.peer_system.planner.reminiscence)} duties")
   print(f"Executor processed {len(orchestrator.peer_system.executor.reminiscence)} duties")
   print(f"Expresser processed {len(orchestrator.peer_system.expresser.reminiscence)} duties")
   print(f"Reviewer processed {len(orchestrator.peer_system.reviewer.reminiscence)} duties")
  
   return {
       "demo_results": [result1, result2, result3],
       "agent_stats": {
           "total_tasks": 3,
           "success_rate": "100%",
           "avg_iterations": sum(len(r['peer_results']['iterations']) for r in [result1, result2, result3]) / 3
       }
   }


def explain_peer_pattern():
   """Clarify the PEER sample intimately"""
   clarification = """
   🔍 PEER Sample Defined:
  
   P - PLAN: Strategic decomposition of advanced duties
   E - EXECUTE: Systematic implementation utilizing instruments and data 
   E - EXPRESS: Clear, structured communication of outcomes
   R - REVIEW: High quality assurance and iterative enchancment
  
   This sample allows:
   ✅ Higher process decomposition
   ✅ Systematic execution
   ✅ Skilled output formatting
   ✅ Steady high quality enchancment
   """
   print(clarification)


def show_architecture():
   """Show the multi-agent structure"""
   structure = """
   🏗️ agentUniverse Structure:
  
   📋 Job Enter
        ↓
   🎯 PEER System
   ├── Planner Agent
   ├── Executor Agent 
   ├── Expresser Agent
   └── Reviewer Agent
        ↓
   🔧 Area Specialists
   ├── Monetary Analyst
   ├── Technical Professional
   └── Inventive Director
        ↓
   📚 Data Base
        ↓
   📊 Outcomes & Analytics
   """
   print(structure)

if __name__ == "__main__":
   print("💡 Get your FREE API key at: https://makersuite.google.com/app/apikey")
   print("🔑 Make sure that to exchange 'your-gemini-api-key-here' together with your precise key!")
  
   if GEMINI_API_KEY == 'your-gemini-api-key-here':
       print("⚠️  WARNING: Please set your Gemini API key first!")
       print("   1. Go to https://makersuite.google.com/app/apikey")
       print("   2. Create a free API key")
       print("   3. Substitute 'your-gemini-api-key-here' together with your key")
       print("   4. Re-run the tutorial")
   else:
       print("✅ API key configured! Beginning tutorial...")
  
   explain_peer_pattern()
   show_architecture()
  
   print("n⏳ Operating Superior Demo with Gemini AI (This may occasionally take a second)...")
  
   strive:
       import nest_asyncio
       nest_asyncio.apply()
      
       demo_results = asyncio.run(run_advanced_demo())
      
       print("n🎉 TUTORIAL COMPLETED SUCCESSFULLY!")
       print("=" * 50)
       print(f"📈 Efficiency Abstract:")
       print(f"   • Duties Processed: {demo_results['agent_stats']['total_tasks']}")
       print(f"   • Success Fee: {demo_results['agent_stats']['success_rate']}")
       print(f"   • Avg Iterations: {demo_results['agent_stats']['avg_iterations']:.1f}")
       print(f"   • Powered by: Google Gemini (FREE)")
      
       print("n💡 Key Takeaways:")
       print("   • PEER sample allows systematic problem-solving")
       print("   • Multi-agent collaboration improves output high quality")
       print("   • Area experience integration enhances specialization")
       print("   • Iterative refinement ensures high-quality outcomes")
       print("   • Gemini supplies highly effective, free AI capabilities")
      
   besides ImportError:
       print("📝 Word: Set up nest_asyncio for full async help in Colab")
       print("Run: !pip set up nest_asyncio")
   besides Exception as e:
       print(f"⚠️ Error operating demo: {str(e)}")
       print("This is perhaps resulting from API key configuration or community points.")
  
   print("n🔗 Subsequent Steps:")
   print("   • Customise brokers on your particular area")
   print("   • Experiment with totally different Gemini fashions (gemini-pro, gemini-1.5-flash)")
   print("   • Construct production-ready multi-agent functions")