A Coding Data To Assemble Intelligent Multi-Agent Applications With The PEER Pattern

!pip arrange agentUniverse google-generativeai python-dotenv pydantic


import os
import asyncio
from typing import Dict, Document, 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 Proper right here' 
genai.configure(api_key=GEMINI_API_KEY)

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


@dataclass
class Course of:
   id: str
   description: str
   context: Dict[str, Any]
   standing: str = "pending"
   final result: Non-obligatory[str] = None
   recommendations: Non-obligatory[str] = None


class BaseAgent:
   """Base agent class with core efficiency"""
   def __init__(self, title: str, operate: AgentRole, system_prompt: str):
       self.title = title
       self.operate = operate
       self.system_prompt = system_prompt
       self.memory: Document[Dict] = []
  
   async def course of(self, job: Course of) -> str:
       speedy = f"{self.system_prompt}nnTask: {job.description}nContext: {json.dumps(job.context)}"
      
       final result = await self._simulate_llm_call(speedy, job)
      
       self.memory.append({
           "task_id": job.id,
           "enter": job.description,
           "output": final result,
           "timestamp": time.time()
       })
      
       return final result
  
   async def _simulate_llm_call(self, speedy: str, job: Course of) -> str:
       """Title Google Gemini API for precise LLM processing"""
       attempt:
           model = genai.GenerativeModel('gemini-1.5-flash')
          
           enhanced_prompt = self._create_role_prompt(speedy, job)
          
           response = await asyncio.to_thread(
               lambda: model.generate_content(enhanced_prompt)
           )
          
           return response.textual content material.strip()
          
       apart from Exception as e:
           print(f"⚠️ Gemini API error for {self.operate.value}: {str(e)}")
           return self._get_fallback_response(job)
  
   def _create_role_prompt(self, base_prompt: str, job: Course of) -> str:
       """Create enhanced role-specific prompts for Gemini"""
       role_instructions = {
           AgentRole.PLANNER: "You are a strategic planning educated. Create detailed, actionable plans. Break down difficult duties into clear steps with priorities and dependencies.",
           AgentRole.EXECUTOR: "You are a consultant executor. Analyze the responsibility completely and provide detailed implementation insights. Cope with smart choices and potential challenges.",
           AgentRole.EXPRESSER: "You are a talented communicator. Present information clearly, professionally, and engagingly. Building your response with headers, bullet components, and clear conclusions.",
           AgentRole.REVIEWER: "You are a top quality assurance educated. Contemplate completeness, accuracy, and readability. Current specific, actionable enchancment suggestions."
       }
      
       context_info = f"Earlier context: {json.dumps(job.context, indent=2)}" if job.context else "No earlier context"
      
       return f"""
{role_instructions[self.role]}


{base_prompt}


{context_info}


Course of to course of: {job.description}


Current an entire, expert response relevant in your operate as {self.operate.value}.
"""
  
   def _get_fallback_response(self, job: Course of) -> str:
       """Fallback responses if Gemini API is unavailable"""
       fallbacks = {
           AgentRole.PLANNER: f"STRATEGIC PLAN for '{job.description}': 1) Requirement analysis 2) Helpful useful resource analysis 3) Implementation roadmap 4) Risk mitigation 5) Success metrics",
           AgentRole.EXECUTOR: f"EXECUTION ANALYSIS for '{job.description}': Full analysis achieved. Key findings acknowledged, smart choices developed, implementation considerations well-known.",
           AgentRole.EXPRESSER: f"PROFESSIONAL SUMMARY for '{job.description}': ## Analysis Completenn**Key Insights:** Detailed analysis performedn**Solutions:** Strategic actions identifiedn**Subsequent Steps:** Implementation ready",
           AgentRole.REVIEWER: f"QUALITY REVIEW for '{job.description}': **Analysis:** High quality output achieved. **Strengths:** Full analysis, clear development. **Suggestions:** Take note of additional quantitative metrics."
       }
       return fallbacks[self.role]

class PEERAgent:
   """PEER Pattern Implementation - Plan, Execute, Categorical, Overview"""
   def __init__(self):
       self.planner = BaseAgent("Strategic Planner", AgentRole.PLANNER,
           "You are a strategic planning agent. Break down difficult duties into actionable steps.")
      
       self.executor = BaseAgent("Course of Executor", AgentRole.EXECUTOR,
           "You are an execution agent. Full duties successfully using accessible devices and knowledge.")
      
       self.expresser = BaseAgent("Consequence Expresser", AgentRole.EXPRESSER,
           "You are a communication agent. Present outcomes clearly and professionally.")
      
       self.reviewer = BaseAgent("Prime quality Reviewer", AgentRole.REVIEWER,
           "You are a top quality assurance agent. Overview outputs and provide enchancment recommendations.")
      
       self.iteration_count = 0
       self.max_iterations = 3
  
   async def collaborate(self, job: Course of) -> Dict[str, Any]:
       """Execute PEER collaboration pattern"""
       outcomes = {"iterations": [], "final_result": None}
      
       whereas self.iteration_count = 1:
               outcomes["final_result"] = expression
               break
              
           self.iteration_count += 1
           job.context["previous_feedback"] = consider
      
       return outcomes

class MultiAgentOrchestrator:
   """Orchestrates various 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, space: str = "primary") -> Dict[str, Any]:
       """Course of difficult job using PEER pattern and space brokers"""
       job = Course of(
           id=f"task_{int(time.time())}",
           description=description,
           context={"space": space, "complexity": "extreme"}
       )
      
       print(f"🚀 Starting Sophisticated Course of Processing: {description}")
       print("=" * 60)
      
       peer_results = await self.peer_system.collaborate(job)
      
       if space in ["financial", "technical", "creative"]:
           domain_agent = self._get_domain_agent(space)
           if domain_agent:
               print(f"🔧 Space-Explicit Processing ({space})")
               domain_result = await domain_agent.course of(job)
               peer_results["domain_enhancement"] = domain_result
      
       return {
           "task_id": job.id,
           "original_request": description,
           "peer_results": peer_results,
           "standing": "achieved",
           "processing_time": f"{len(peer_results['iterations'])} iterations"
       }
  
   def _get_domain_agent(self, space: str) -> Non-obligatory[BaseAgent]:
       """Get domain-specific agent with enhanced Gemini prompts"""
       domain_agents = {
           "financial": BaseAgent("Financial Analyst", AgentRole.EXECUTOR,
               "You are a senior financial analyst with expertise in market analysis, hazard analysis, and funding strategies. Current detailed financial insights with quantitative analysis."),
           "technical": BaseAgent("Technical Expert", AgentRole.EXECUTOR,
               "You are a lead software program program architect with expertise in system design, scalability, and best practices. Current detailed technical choices with implementation considerations."),
           "creative": BaseAgent("Creative Director", AgentRole.EXPRESSER,
               "You are an award-winning creative director with expertise in mannequin method, content material materials creation, and revolutionary campaigns. Generate compelling and strategic creative choices.")
       }
       return domain_agents.get(space)


class KnowledgeBase:
   """Simple knowledge administration system"""
   def __init__(self):
       self.knowledge = {
           "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, space: str) -> Document[str]:
       return self.knowledge.get(space, ["General knowledge"])


async def run_advanced_demo():
    
   orchestrator = MultiAgentOrchestrator()
   knowledge_base = KnowledgeBase()
  
   print("n📊 DEMO 1: Financial Analysis with PEER Pattern")
   print("-" * 40)
  
   financial_task = "Analyze the potential affect of rising charges of curiosity on tech shares portfolio"
   result1 = await orchestrator.process_complex_task(financial_task, "financial")
  
   print(f"n✅ Course of Completed: {result1['processing_time']}")
   print(f"Final Consequence: {result1['peer_results']['final_result']}")
  
   print("n💻 DEMO 2: Technical Disadvantage Fixing")
   print("-" * 40)
  
   technical_task = "Design a scalable microservices construction for a heavy-traffic e-commerce platform"
   result2 = await orchestrator.process_complex_task(technical_task, "technical")
  
   print(f"n✅ Course of Completed: {result2['processing_time']}")
   print(f"Final Consequence: {result2['peer_results']['final_result']}")
  
   print("n🎨 DEMO 3: Creative Content material materials with Multi-Agent Collaboration")
   print("-" * 40)
  
   creative_task = "Create an entire mannequin method for a sustainable fashion startup"
   result3 = await orchestrator.process_complex_task(creative_task, "creative")
  
   print(f"n✅ Course of Completed: {result3['processing_time']}")
   print(f"Final Consequence: {result3['peer_results']['final_result']}")
  
   print("n🧠 AGENT MEMORY & LEARNING")
   print("-" * 40)
   print(f"Planner processed {len(orchestrator.peer_system.planner.memory)} duties")
   print(f"Executor processed {len(orchestrator.peer_system.executor.memory)} duties")
   print(f"Expresser processed {len(orchestrator.peer_system.expresser.memory)} duties")
   print(f"Reviewer processed {len(orchestrator.peer_system.reviewer.memory)} 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():
   """Make clear the PEER pattern intimately"""
   rationalization = """
   🔍 PEER Pattern Outlined:
  
   P - PLAN: Strategic decomposition of difficult duties
   E - EXECUTE: Systematic implementation using devices and knowledge 
   E - EXPRESS: Clear, structured communication of outcomes
   R - REVIEW: Prime quality assurance and iterative enchancment
  
   This pattern permits:
   ✅ Larger job decomposition
   ✅ Systematic execution
   ✅ Expert output formatting
   ✅ Regular top quality enchancment
   """
   print(rationalization)


def show_architecture():
   """Present the multi-agent construction"""
   construction = """
   🏗️ agentUniverse Construction:
  
   📋 Course of Enter
        ↓
   🎯 PEER System
   ├── Planner Agent
   ├── Executor Agent 
   ├── Expresser Agent
   └── Reviewer Agent
        ↓
   🔧 Space Specialists
   ├── Financial Analyst
   ├── Technical Expert
   └── Creative Director
        ↓
   📚 Information Base
        ↓
   📊 Outcomes & Analytics
   """
   print(construction)

if __name__ == "__main__":
   print("💡 Get your FREE API key at: https://makersuite.google.com/app/apikey")
   print("🔑 Make certain to interchange 'your-gemini-api-key-here' collectively together with your exact 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' collectively together with your key")
       print("   4. Re-run the tutorial")
   else:
       print("✅ API key configured! Starting tutorial...")
  
   explain_peer_pattern()
   show_architecture()
  
   print("n⏳ Working Superior Demo with Gemini AI (It will take a second)...")
  
   attempt:
       import nest_asyncio
       nest_asyncio.apply()
      
       demo_results = asyncio.run(run_advanced_demo())
      
       print("n🎉 TUTORIAL COMPLETED SUCCESSFULLY!")
       print("=" * 50)
       print(f"📈 Effectivity Summary:")
       print(f"   • Duties Processed: {demo_results['agent_stats']['total_tasks']}")
       print(f"   • Success Worth: {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 pattern permits systematic problem-solving")
       print("   • Multi-agent collaboration improves output top quality")
       print("   • Space expertise integration enhances specialization")
       print("   • Iterative refinement ensures high-quality outcomes")
       print("   • Gemini provides extremely efficient, free AI capabilities")
      
   apart from ImportError:
       print("📝 Bear in mind: Arrange nest_asyncio for full async assist in Colab")
       print("Run: !pip arrange nest_asyncio")
   apart from Exception as e:
       print(f"⚠️ Error working demo: {str(e)}")
       print("That is maybe as a consequence of API key configuration or group factors.")
  
   print("n🔗 Subsequent Steps:")
   print("   • Customise brokers in your specific space")
   print("   • Experiment with utterly completely different Gemini fashions (gemini-pro, gemini-1.5-flash)")
   print("   • Assemble production-ready multi-agent functions")