A Step-by-Step Coding Implementation of an Agent2Agent Framework for Collaborative and Critique-Driven AI Problem Solving with Consensus-Building

import google.generativeai as genai
import json
import time
from dataclasses import dataclass
from typing import Dict, List, Any
from enum import Enum
import random
import re


API_KEY = "Use Your Own API Key"  
genai.configure(api_key=API_KEY)

class MessageType(Enum):
    HANDSHAKE = "handshake"
    TASK_PROPOSAL = "task_proposal"
    ANALYSIS = "analysis"
    CRITIQUE = "critique"
    SYNTHESIS = "synthesis"
    VOTE = "vote"
    CONSENSUS = "consensus"

@dataclass
class A2AMessage:
    sender_id: str
    receiver_id: str
    message_type: MessageType
    payload: Dict[str, Any]
    timestamp: float
    priority: int = 1

class GeminiAgent:
    def __init__(self, agent_id: str, role: str, personality: str, temperature: float = 0.7):
        self.agent_id = agent_id
        self.role = role
        self.personality = personality
        self.temperature = temperature
        self.conversation_memory = []
        self.current_position = None
        self.confidence = 0.5
       
        self.model = genai.GenerativeModel('gemini-2.0-flash')
       
    def get_system_context(self, task_context: str = "") -> str:
        return f"""You are {self.agent_id}, an AI agent in a multi-agent collaborative system.


ROLE: {self.role}
PERSONALITY: {self.personality}


CONTEXT: {task_context}


You are participating in Agent2Agent protocol communication. Your responsibilities:
1. Analyze problems from your specialized perspective
2. Provide constructive feedback to other agents
3. Synthesize information from multiple sources
4. Make data-driven decisions
5. Collaborate effectively while maintaining your expertise


IMPORTANT: Always structure your response as JSON with these fields:
{{
    "agent_id": "{self.agent_id}",
    "main_response": "your primary response content",
    "confidence_level": 0.8,
    "key_insights": ["insight1", "insight2"],
    "questions_for_others": ["question1", "question2"],
    "next_action": "suggested next step"
}}


Stay true to your role and personality while being collaborative."""


    def generate_response(self, prompt: str, context: str = "") -> Dict[str, Any]:
        """Generate response using Gemini API"""
        try:
            full_prompt = f"{self.get_system_context(context)}nnPROMPT: {prompt}"
           
            response = self.model.generate_content(
                full_prompt,
                generation_config=genai.types.GenerationConfig(
                    temperature=self.temperature,
                    max_output_tokens=600,
                )
            )
           
            response_text = response.text
           
            json_match = re.search(r'{.*}', response_text, re.DOTALL)
            if json_match:
                try:
                    return json.loads(json_match.group())
                except json.JSONDecodeError:
                    pass
           
            return {
                "agent_id": self.agent_id,
                "main_response": response_text[:200] + "..." if len(response_text) > 200 else response_text,
                "confidence_level": random.uniform(0.6, 0.9),
                "key_insights": [f"Insight from {self.role}"],
                "questions_for_others": ["What do you think about this approach?"],
                "next_action": "Continue analysis"
            }
           
        except Exception as e:
            print(f"  Gemini API Error for {self.agent_id}: {e}")
            return {
                "agent_id": self.agent_id,
                "main_response": f"Error occurred in {self.agent_id}: {str(e)}",
                "confidence_level": 0.1,
                "key_insights": ["API error encountered"],
                "questions_for_others": [],
                "next_action": "Retry connection"
            }


    def analyze_task(self, task: str) -> Dict[str, Any]:
        prompt = f"Analyze this task from your {self.role} perspective: {task}"
        return self.generate_response(prompt, f"Task Analysis: {task}")
   
    def critique_analysis(self, other_analysis: Dict[str, Any], original_task: str) -> Dict[str, Any]:
        analysis_summary = other_analysis.get('main_response', 'No analysis provided')
        prompt = f"""
        ORIGINAL TASK: {original_task}
       
        ANOTHER AGENT'S ANALYSIS: {analysis_summary}
        THEIR CONFIDENCE: {other_analysis.get('confidence_level', 0.5)}
        THEIR INSIGHTS: {other_analysis.get('key_insights', [])}
       
        Provide constructive critique and alternative perspectives from your {self.role} expertise.
        """
        return self.generate_response(prompt, f"Critique Session: {original_task}")
   
    def synthesize_solutions(self, all_analyses: List[Dict[str, Any]], task: str) -> Dict[str, Any]:
        analyses_summary = "n".join([
            f"Agent {i+1}: {analysis.get('main_response', 'No response')[:100]}..."
            for i, analysis in enumerate(all_analyses)
        ])
       
        prompt = f"""
        TASK: {task}
       
        ALL AGENT ANALYSES:
        {analyses_summary}
       
        As the {self.role}, synthesize these perspectives into a comprehensive solution.
        Identify common themes, resolve conflicts, and propose the best path forward.
        """
        return self.generate_response(prompt, f"Synthesis Phase: {task}")

class Agent2AgentCollaborativeSystem:
    def __init__(self):
        self.agents: Dict[str, GeminiAgent] = {}
        self.collaboration_history: List[Dict[str, Any]] = []
       
    def add_agent(self, agent: GeminiAgent):
        self.agents[agent.agent_id] = agent
        print(f" Registered Gemini Agent: {agent.agent_id} ({agent.role})")
   
    def run_collaborative_problem_solving(self, problem: str):
        print(f"n Multi-Gemini Collaborative Problem Solving")
        print(f" Problem: {problem}")
        print("=" * 80)
       
        print("n PHASE 1: Individual Agent Analysis")
        initial_analyses = {}
       
        for agent_id, agent in self.agents.items():
            print(f"n {agent_id} analyzing...")
            analysis = agent.analyze_task(problem)
            initial_analyses[agent_id] = analysis
           
            print(f" {agent_id} ({agent.role}):")
            print(f"   Response: {analysis.get('main_response', 'No response')[:150]}...")
            print(f"   Confidence: {analysis.get('confidence_level', 0.5):.2f}")
            print(f"   Key Insights: {analysis.get('key_insights', [])}")
       
        print(f"n PHASE 2: Cross-Agent Critique & Feedback")
        critiques = {}
       
        agent_list = list(self.agents.items())
        for i, (agent_id, agent) in enumerate(agent_list):
            target_agent_id = agent_list[(i + 1) % len(agent_list)][0]
            target_analysis = initial_analyses[target_agent_id]
           
            print(f"n {agent_id} critiquing {target_agent_id}'s analysis...")
            critique = agent.critique_analysis(target_analysis, problem)
            critiques[f"{agent_id}_critiques_{target_agent_id}"] = critique
           
            print(f" {agent_id} → {target_agent_id}:")
            print(f"   Critique: {critique.get('main_response', 'No critique')[:120]}...")
            print(f"   Questions: {critique.get('questions_for_others', [])}")
       
        print(f"n PHASE 3: Solution Synthesis")
        final_solutions = {}
        all_analyses = list(initial_analyses.values())
       
        for agent_id, agent in self.agents.items():
            print(f"n {agent_id} synthesizing final solution...")
            synthesis = agent.synthesize_solutions(all_analyses, problem)
            final_solutions[agent_id] = synthesis
           
            print(f" {agent_id} Final Solution:")
            print(f"   {synthesis.get('main_response', 'No synthesis')[:200]}...")
            print(f"   Confidence: {synthesis.get('confidence_level', 0.5):.2f}")
            print(f"   Next Action: {synthesis.get('next_action', 'No action specified')}")
       
        print(f"n PHASE 4: Consensus & Recommendation")
       
        avg_confidence = sum(
            sol.get('confidence_level', 0.5) for sol in final_solutions.values()
        ) / len(final_solutions)
       
        print(f" Average Solution Confidence: {avg_confidence:.2f}")
       
        most_confident_agent = max(
            final_solutions.items(),
            key=lambda x: x[1].get('confidence_level', 0)
        )
       
        print(f"n Most Confident Solution from: {most_confident_agent[0]}")
        print(f" Recommended Solution: {most_confident_agent[1].get('main_response', 'No solution')}")
       
        all_insights = []
        for solution in final_solutions.values():
            all_insights.extend(solution.get('key_insights', []))
       
        print(f"n Collective Intelligence Insights:")
        for i, insight in enumerate(set(all_insights), 1):
            print(f"   {i}. {insight}")
       
        return final_solutions

def create_specialized_gemini_agents():
    """Create diverse Gemini agents with different roles and personalities"""
    agents = [
        GeminiAgent(
            "DataScientist_Alpha",
            "Data Scientist & Analytics Specialist",
            "Methodical, evidence-based, loves patterns and statistical insights",
            temperature=0.3
        ),
        GeminiAgent(
            "ProductManager_Beta",
            "Product Strategy & User Experience Expert",
            "User-focused, strategic thinker, balances business needs with user value",
            temperature=0.5
        ),
        GeminiAgent(
            "TechArchitect_Gamma",
            "Technical Architecture & Engineering Lead",
            "System-oriented, focuses on scalability, performance, and technical feasibility",
            temperature=0.4
        ),
        GeminiAgent(
            "CreativeInnovator_Delta",
            "Innovation & Creative Problem Solving Specialist",
            "Bold, unconventional, pushes boundaries and suggests breakthrough approaches",
            temperature=0.8
        ),
        GeminiAgent(
            "RiskAnalyst_Epsilon",
            "Risk Management & Compliance Expert",
            "Cautious, thorough, identifies potential issues and mitigation strategies",
            temperature=0.2
        )
    ]
    return agents

def run_gemini_agent2agent_demo():
    print(" Agent2Agent Protocol: Multi-Gemini Collaborative Intelligence")
    print("=" * 80)
   
    if API_KEY == "your-gemini-api-key-here":
        print("  Please set your Gemini API key!")
        print(" Get your free API key from: https://makersuite.google.com/app/apikey")
        return
   
    collaborative_system = Agent2AgentCollaborativeSystem()
   
    for agent in create_specialized_gemini_agents():
        collaborative_system.add_agent(agent)
   
    problems = [
        "Design a sustainable urban transportation system for a city of 2 million people that reduces carbon emissions by 50% while maintaining economic viability.",
        "Create a strategy for a tech startup to compete against established players in the AI-powered healthcare diagnostics market."
    ]
   
    for i, problem in enumerate(problems, 1):
        print(f"n{' COLLABORATION SESSION ' + str(i):=^80}")
        collaborative_system.run_collaborative_problem_solving(problem)
       
        if i < len(problems):
            print(f"n{'  BREAK BETWEEN SESSIONS':=^80}")
            time.sleep(3)
   
    print(f"n Multi-Gemini Agent2Agent Collaboration Complete!")
    print(" This demonstrates true AI-to-AI collaboration using Google's Gemini models!")
    print(" Each agent brought unique expertise to solve complex problems collectively!")


if __name__ == "__main__":
    run_gemini_agent2agent_demo()