Una implementación de codificación paso a paso de un marco Agent2Agent para la resolución de problemas de IA colaborativos y de crítica con construcción de consenso

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)}\n\nPROMPT: {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()