Close Menu

    A Coding Guide to Build Intelligent Multi-Agent Systems with the PEER Pattern

    In this tutorial, we explore a powerful multi-agent system built around the PEER pattern: Plan, Execute, Express, and Review. We run the entire workflow in Google Colab/Notebook, integrating agents with specialized roles and leveraging Google’s Gemini 1.5 Flash model via a free API key. As we walk through the system, we observe how each agent collaborates to tackle complex tasks across different domains such as finance, technology, and creative strategy. This hands-on tutorial allows us to understand the architecture, workflow, and iterative refinement that underpin high-quality AI outputs.

    Copy CodeCopiedUse a different Browser
    !pip install agentUniverse google-generativeai python-dotenv pydantic


import os
import asyncio
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
import json
import time
import google.generativeai as genai


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

    We begin by installing the required libraries, including agentUniverse and google-generativeai, to set up our multi-agent system. After importing the necessary modules, we configure the Gemini API using our free API key to enable AI-powered content generation. Check out the Full Codes here.

    Copy CodeCopiedUse a different Browser
    class AgentRole(Enum):
   PLANNER = "planner"
   EXECUTOR = "executor"
   EXPRESSER = "expresser"
   REVIEWER = "reviewer"


@dataclass
class Task:
   id: str
   description: str
   context: Dict[str, Any]
   status: str = "pending"
   result: Optional[str] = None
   feedback: Optional[str] = None


class 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}nnTask: {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"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/26a0.png" alt="⚠" class="wp-smiley" /> 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 Completenn**Key Insights:** Detailed analysis performedn**Recommendations:** Strategic actions identifiedn**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]

    We define four distinct agent roles, Planner, Executor, Expresser, and Reviewer, using an Enum to represent their specialized functions. Then, we create a Task dataclass to manage task metadata, including status, result, and feedback. The BaseAgent class serves as the core blueprint for all agents, enabling them to process tasks, call the Gemini API with role-specific prompts, store results in memory, and gracefully fall back to predefined responses if the API fails. Check out the Full Codes here.

    Copy CodeCopiedUse a different Browser
    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"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f3af.png" alt="🎯" class="wp-smiley" /> Planning Phase (Iteration {self.iteration_count + 1})")
           plan = await self.planner.process(task)
           iteration_result["plan"] = plan
           task.context["current_plan"] = plan
          
           print(f"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/26a1.png" alt="⚡" class="wp-smiley" /> Execution Phase")
           execution = await self.executor.process(task)
           iteration_result["execution"] = execution
           task.context["execution_result"] = execution
          
           print(f"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4dd.png" alt="📝" class="wp-smiley" /> Expression Phase")
           expression = await self.expresser.process(task)
           iteration_result["expression"] = expression
           task.result = expression
          
           print(f"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f50d.png" alt="🔍" class="wp-smiley" /> 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

    We implement the PEER pattern, Plan, Execute, Express, Review, through the PEERAgent class, which coordinates four specialized agents for collaborative task-solving. Each iteration runs through all four phases, refining the task output based on structured planning, execution, professional expression, and quality review. We allow up to three iterations, concluding early if the review indicates high-quality completion, making the workflow both adaptive and efficient. Check out the Full Codes here.

    Copy CodeCopiedUse a different Browser
    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"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f680.png" alt="🚀" class="wp-smiley" /> 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"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f527.png" alt="🔧" class="wp-smiley" /> 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<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4ca.png" alt="📊" class="wp-smiley" /> 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<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Task Completed: {result1['processing_time']}")
   print(f"Final Result: {result1['peer_results']['final_result']}")
  
   print("n<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4bb.png" alt="💻" class="wp-smiley" /> 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<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Task Completed: {result2['processing_time']}")
   print(f"Final Result: {result2['peer_results']['final_result']}")
  
   print("n<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f3a8.png" alt="🎨" class="wp-smiley" /> 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<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Task Completed: {result3['processing_time']}")
   print(f"Final Result: {result3['peer_results']['final_result']}")
  
   print("n<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f9e0.png" alt="🧠" class="wp-smiley" /> 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 = """
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f50d.png" alt="🔍" class="wp-smiley" /> 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:
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Better task decomposition
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Systematic execution
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Professional output formatting
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> Continuous quality improvement
   """
   print(explanation)


def show_architecture():
   """Display the multi-agent architecture"""
   architecture = """
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f3d7.png" alt="🏗" class="wp-smiley" /> agentUniverse Architecture:
  
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4cb.png" alt="📋" class="wp-smiley" /> Task Input
        ↓
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f3af.png" alt="🎯" class="wp-smiley" /> PEER System
   ├── Planner Agent
   ├── Executor Agent 
   ├── Expresser Agent
   └── Reviewer Agent
        ↓
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f527.png" alt="🔧" class="wp-smiley" /> Domain Specialists
   ├── Financial Analyst
   ├── Technical Expert
   └── Creative Director
        ↓
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4da.png" alt="📚" class="wp-smiley" /> Knowledge Base
        ↓
   <img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4ca.png" alt="📊" class="wp-smiley" /> Results & Analytics
   """
   print(architecture)

    We bring everything together through the MultiAgentOrchestrator, which coordinates the PEER system and, when needed, invokes domain-specific agents like the Financial Analyst or Technical Expert. This orchestrator handles each complex task by first leveraging the PEER pattern and then enhancing results with specialized knowledge. We also define a simple KnowledgeBase to support domain-aware reasoning. In the run_advanced_demo() function, we test the full pipeline with three tasks, financial, technical, and creative, while capturing agent performance and iteration metrics to showcase the power and versatility of our multi-agent setup. Check out the Full Codes here.

    Copy CodeCopiedUse a different Browser
    if __name__ == "__main__":
   print("<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4a1.png" alt="💡" class="wp-smiley" /> Get your FREE API key at: https://makersuite.google.com/app/apikey")
   print("<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f511.png" alt="🔑" class="wp-smiley" /> Make sure to replace 'your-gemini-api-key-here' with your actual key!")
  
   if GEMINI_API_KEY == 'your-gemini-api-key-here':
       print("<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/26a0.png" alt="⚠" class="wp-smiley" />  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("<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/2705.png" alt="✅" class="wp-smiley" /> API key configured! Starting tutorial...")
  
   explain_peer_pattern()
   show_architecture()
  
   print("n<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/23f3.png" alt="⏳" class="wp-smiley" /> 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<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f389.png" alt="🎉" class="wp-smiley" /> TUTORIAL COMPLETED SUCCESSFULLY!")
       print("=" * 50)
       print(f"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4c8.png" alt="📈" class="wp-smiley" /> 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<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4a1.png" alt="💡" class="wp-smiley" /> 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("<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f4dd.png" alt="📝" class="wp-smiley" /> Note: Install nest_asyncio for full async support in Colab")
       print("Run: !pip install nest_asyncio")
   except Exception as e:
       print(f"<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/26a0.png" alt="⚠" class="wp-smiley" /> Error running demo: {str(e)}")
       print("This might be due to API key configuration or network issues.")
  
   print("n<img src="https://s.w.org/images/core/emoji/16.0.1/72x72/1f517.png" alt="🔗" class="wp-smiley" /> 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")

    We conclude the tutorial by initializing the system, verifying the Gemini API key, and executing the full PEER-based multi-agent workflow. We explain the architecture and pattern before running the demo, and upon successful completion, we display a performance summary and key takeaways.

    In conclusion, we successfully demonstrate how a multi-agent system can systematically solve complex problems with the help of domain-specific reasoning, structured communication, and iterative quality checks. We gain insights into the collaborative power of the PEER framework and witness how Gemini enhances each agent’s output. Through this experience, we realize the potential of modular AI systems in creating scalable, reliable, and intelligent applications ready for real-world deployment.

    Check out the Full Codes here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter.

    The post A Coding Guide to Build Intelligent Multi-Agent Systems with the PEER Pattern appeared first on MarkTechPost.

    Source: Read More 

    Related Posts

    Leave A Reply