!pip install agentUniverse google-generativeai python-dotenv pydanticimport osimport asynciofrom typing import Dict, List, Any, Optionalfrom dataclasses import dataclassfrom enum import Enumimport jsonimport timeimport google.generativeai as genaiGEMINI_API_KEY = 'Use Your API Key Here' genai.configure(api_key=GEMINI_API_KEY)

class AgentRole(Enum):   PLANNER = "planner"   EXECUTOR = "executor"   EXPRESSER = "expresser"   REVIEWER = "reviewer"@dataclassclass Task:   id: str   description: str   context: Dict[str, Any]   status: str = "pending"   result: Optional[str] = None   feedback: Optional[str] = Noneclass BaseAgent:   """Base agent class with core functionality"""   def __init__(self, name: str, role: AgentRole, system_prompt: str):       self.name = name       self.role = role       self.system_prompt = system_prompt       self.memory: List[Dict] = []     async def process(self, task: Task) -> str:       prompt = f"{self.system_prompt}\n\nTask: {task.description}\nContext: {json.dumps(task.context)}"             result = await self._simulate_llm_call(prompt, task)             self.memory.append({           "task_id": task.id,           "input": task.description,           "output": result,           "timestamp": time.time()       })             return result     async def _simulate_llm_call(self, prompt: str, task: Task) -> str:       """Call Google Gemini API for real LLM processing"""       try:           model = genai.GenerativeModel('gemini-1.5-flash')                     enhanced_prompt = self._create_role_prompt(prompt, task)                     response = await asyncio.to_thread(               lambda: model.generate_content(enhanced_prompt)           )                     return response.text.strip()                 except Exception as e:           print(f" Gemini API error for {self.role.value}: {str(e)}")           return self._get_fallback_response(task)     def _create_role_prompt(self, base_prompt: str, task: Task) -> str:       """Create enhanced role-specific prompts for Gemini"""       role_instructions = {           AgentRole.PLANNER: "You are a strategic planning expert. Create detailed, actionable plans. Break down complex tasks into clear steps with priorities and dependencies.",           AgentRole.EXECUTOR: "You are a skilled executor. Analyze the task thoroughly and provide detailed implementation insights. Focus on practical solutions and potential challenges.",           AgentRole.EXPRESSER: "You are a professional communicator. Present information clearly, professionally, and engagingly. Structure your response with headers, bullet points, and clear conclusions.",           AgentRole.REVIEWER: "You are a quality assurance expert. Evaluate completeness, accuracy, and clarity. Provide specific, actionable improvement suggestions."       }             context_info = f"Previous context: {json.dumps(task.context, indent=2)}" if task.context else "No previous context"             return f"""{role_instructions[self.role]}{base_prompt}{context_info}Task to process: {task.description}Provide a comprehensive, professional response appropriate for your role as {self.role.value}."""     def _get_fallback_response(self, task: Task) -> str:       """Fallback responses if Gemini API is unavailable"""       fallbacks = {           AgentRole.PLANNER: f"STRATEGIC PLAN for '{task.description}': 1) Requirement analysis 2) Resource assessment 3) Implementation roadmap 4) Risk mitigation 5) Success metrics",           AgentRole.EXECUTOR: f"EXECUTION ANALYSIS for '{task.description}': Comprehensive analysis completed. Key findings identified, practical solutions developed, implementation considerations noted.",           AgentRole.EXPRESSER: f"PROFESSIONAL SUMMARY for '{task.description}': ## Analysis Complete\n\n**Key Insights:** Detailed analysis performed\n**Recommendations:** Strategic actions identified\n**Next Steps:** Implementation ready",           AgentRole.REVIEWER: f"QUALITY REVIEW for '{task.description}': **Assessment:** High quality output achieved. **Strengths:** Comprehensive analysis, clear structure. **Suggestions:** Consider additional quantitative metrics."       }       return fallbacks[self.role]

class PEERAgent:   """PEER Pattern Implementation - Plan, Execute, Express, Review"""   def __init__(self):       self.planner = BaseAgent("Strategic Planner", AgentRole.PLANNER,           "You are a strategic planning agent. Break down complex tasks into actionable steps.")             self.executor = BaseAgent("Task Executor", AgentRole.EXECUTOR,           "You are an execution agent. Complete tasks efficiently using available tools and knowledge.")             self.expresser = BaseAgent("Result Expresser", AgentRole.EXPRESSER,           "You are a communication agent. Present results clearly and professionally.")             self.reviewer = BaseAgent("Quality Reviewer", AgentRole.REVIEWER,           "You are a quality assurance agent. Review outputs and provide improvement feedback.")             self.iteration_count = 0       self.max_iterations = 3     async def collaborate(self, task: Task) -> Dict[str, Any]:       """Execute PEER collaboration pattern"""       results = {"iterations": [], "final_result": None}             while self.iteration_count < self.max_iterations:           iteration_result = {}                     print(f" Planning Phase (Iteration {self.iteration_count + 1})")           plan = await self.planner.process(task)           iteration_result["plan"] = plan           task.context["current_plan"] = plan                     print(f" Execution Phase")           execution = await self.executor.process(task)           iteration_result["execution"] = execution           task.context["execution_result"] = execution                     print(f" Expression Phase")           expression = await self.expresser.process(task)           iteration_result["expression"] = expression           task.result = expression                     print(f" Review Phase")           review = await self.reviewer.process(task)           iteration_result["review"] = review           task.feedback = review                     results["iterations"].append(iteration_result)                     if "high" in review.lower() and self.iteration_count >= 1:               results["final_result"] = expression               break                         self.iteration_count += 1           task.context["previous_feedback"] = review             return results

class MultiAgentOrchestrator:   """Orchestrates multiple specialized agents"""   def __init__(self):       self.agents = {}       self.peer_system = PEERAgent()       self.task_queue = []         def register_agent(self, agent: BaseAgent):       """Register a specialized agent"""       self.agents[agent.name] = agent     async def process_complex_task(self, description: str, domain: str = "general") -> Dict[str, Any]:       """Process complex task using PEER pattern and domain agents"""       task = Task(           id=f"task_{int(time.time())}",           description=description,           context={"domain": domain, "complexity": "high"}       )             print(f" Starting Complex Task Processing: {description}")       print("=" * 60)             peer_results = await self.peer_system.collaborate(task)             if domain in ["financial", "technical", "creative"]:           domain_agent = self._get_domain_agent(domain)           if domain_agent:               print(f" Domain-Specific Processing ({domain})")               domain_result = await domain_agent.process(task)               peer_results["domain_enhancement"] = domain_result             return {           "task_id": task.id,           "original_request": description,           "peer_results": peer_results,           "status": "completed",           "processing_time": f"{len(peer_results['iterations'])} iterations"       }     def _get_domain_agent(self, domain: str) -> Optional[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, risk assessment, and investment strategies. Provide detailed financial insights with quantitative analysis."),           "technical": BaseAgent("Technical Expert", AgentRole.EXECUTOR,               "You are a lead software architect with expertise in system design, scalability, and best practices. Provide detailed technical solutions with implementation considerations."),           "creative": BaseAgent("Creative Director", AgentRole.EXPRESSER,               "You are an award-winning creative director with expertise in brand strategy, content creation, and innovative campaigns. Generate compelling and strategic creative solutions.")       }       return domain_agents.get(domain)class KnowledgeBase:   """Simple knowledge management 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, domain: str) -> List[str]:       return self.knowledge.get(domain, ["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 impact of rising interest rates on tech stocks portfolio"   result1 = await orchestrator.process_complex_task(financial_task, "financial")     print(f"\n Task Completed: {result1['processing_time']}")   print(f"Final Result: {result1['peer_results']['final_result']}")     print("\n DEMO 2: Technical Problem Solving")   print("-" * 40)     technical_task = "Design a scalable microservices architecture for a high-traffic e-commerce platform"   result2 = await orchestrator.process_complex_task(technical_task, "technical")     print(f"\n Task Completed: {result2['processing_time']}")   print(f"Final Result: {result2['peer_results']['final_result']}")     print("\n DEMO 3: Creative Content with Multi-Agent Collaboration")   print("-" * 40)     creative_task = "Create a comprehensive brand strategy for a sustainable fashion startup"   result3 = await orchestrator.process_complex_task(creative_task, "creative")     print(f"\n Task Completed: {result3['processing_time']}")   print(f"Final Result: {result3['peer_results']['final_result']}")     print("\n AGENT MEMORY & LEARNING")   print("-" * 40)   print(f"Planner processed {len(orchestrator.peer_system.planner.memory)} tasks")   print(f"Executor processed {len(orchestrator.peer_system.executor.memory)} tasks")   print(f"Expresser processed {len(orchestrator.peer_system.expresser.memory)} tasks")   print(f"Reviewer processed {len(orchestrator.peer_system.reviewer.memory)} tasks")     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():   """Explain the PEER pattern in detail"""   explanation = """    PEER Pattern Explained:     P - PLAN: Strategic decomposition of complex tasks   E - EXECUTE: Systematic implementation using tools and knowledge    E - EXPRESS: Clear, structured communication of results   R - REVIEW: Quality assurance and iterative improvement     This pattern enables:    Better task decomposition    Systematic execution    Professional output formatting    Continuous quality improvement   """   print(explanation)def show_architecture():   """Display the multi-agent architecture"""   architecture = """    agentUniverse Architecture:      Task Input        ↓    PEER System   ├── Planner Agent   ├── Executor Agent    ├── Expresser Agent   └── Reviewer Agent        ↓    Domain Specialists   ├── Financial Analyst   ├── Technical Expert   └── Creative Director        ↓    Knowledge Base        ↓    Results & Analytics   """   print(architecture)

if __name__ == "__main__":   print(" Get your FREE API key at: https://makersuite.google.com/app/apikey")   print(" Make sure to replace 'your-gemini-api-key-here' with your actual 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. Replace 'your-gemini-api-key-here' with your key")       print("   4. Re-run the tutorial")   else:       print(" API key configured! Starting tutorial...")     explain_peer_pattern()   show_architecture()     print("\n Running Advanced Demo with Gemini AI (This may take a moment)...")     try:       import nest_asyncio       nest_asyncio.apply()             demo_results = asyncio.run(run_advanced_demo())             print("\n TUTORIAL COMPLETED SUCCESSFULLY!")       print("=" * 50)       print(f" Performance Summary:")       print(f"   • Tasks Processed: {demo_results['agent_stats']['total_tasks']}")       print(f"   • Success Rate: {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 enables systematic problem-solving")       print("   • Multi-agent collaboration improves output quality")       print("   • Domain expertise integration enhances specialization")       print("   • Iterative refinement ensures high-quality results")       print("   • Gemini provides powerful, free AI capabilities")         except ImportError:       print(" Note: Install nest_asyncio for full async support in Colab")       print("Run: !pip install nest_asyncio")   except Exception as e:       print(f" Error running demo: {str(e)}")       print("This might be due to API key configuration or network issues.")     print("\n Next Steps:")   print("   • Customize agents for your specific domain")   print("   • Experiment with different Gemini models (gemini-pro, gemini-1.5-flash)")   print("   • Build production-ready multi-agent applications")