we implement the Agent2Agent collaborative framework built atop Google’s Gemini models. The guide walks through the creation of specialized AI personas, ranging from data scientists and product strategists to risk analysts and creative innovators. It demonstrates how these agents can exchange structured messages to tackle complex, real-world challenges. By defining clear roles, personalities, and communication protocols, the tutorial highlights how to orchestrate multi-agent problem solving in three phases: individual analysis, cross-agent critique, and synthesis of solutions.
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)We import the core libraries for building your Agent2Agent system, handling JSON, timing, data structures, and regex utilities. Then, we set your Gemini API key and initialize the genai client for subsequent calls. This ensures that all subsequent requests to Google’s generative AI endpoints are authenticated.
class MessageType(Enum):
HANDSHAKE = "handshake"
TASK_PROPOSAL = "task_proposal"
ANALYSIS = "analysis"
CRITIQUE = "critique"
SYNTHESIS = "synthesis"
VOTE = "vote"
CONSENSUS = "consensus"This MessageType enum defines the stages of Agent2Agent communication, from initial handshakes and task proposals to analysis, critique, synthesis, voting, and final consensus. It allows you to tag and route messages according to their role in the collaborative workflow.
@dataclass
class A2AMessage:
sender_id: str
receiver_id: str
message_type: MessageType
payload: Dict[str, Any]
timestamp: float
priority: int = 1This A2AMessage dataclass encapsulates all the metadata needed for inter-agent communication, tracking who sent it, who should receive it, the message’s role in the protocol (message_type), its content (payload), when it was sent (timestamp), and its relative processing priority. It provides a structured, type-safe way to serialize and route messages between agents.
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}")
The GeminiAgent class wraps a Google Gemini model instance, encapsulating each agent’s identity, role, and personality to generate structured JSON responses. It provides helper methods to build system prompts, call the API with controlled temperature and token limits, and fall back to a default response format in case of parse or API errors. With analyze_task, critique_analysis, and synthesize_solutions, it streamlines each phase of the multi-agent workflow.
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
The Agent2AgentCollaborativeSystem class manages your fleet of GeminiAgent instances, providing methods to register new agents and orchestrate the four-phase collaboration workflow, individual analysis, cross-agent critique, solution synthesis, and consensus scoring. It handles logging and printing the intermediate results and returns each agent’s final proposed solutions for downstream use.
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 agentsThe create_specialized_gemini_agents function instantiates a balanced team of five Gemini agents, each with a unique role, personality, and temperature setting. The agents cover analytics, product strategy, system architecture, creative innovation, and risk management to ensure well-rounded collaborative problem solving.
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()Finally, the run_gemini_agent2agent_demo function ties everything together: it prints an overview header, ensures our Gemini API key is set, registers the five specialized agents, and then executes collaborative problem-solving sessions on each predefined challenge (with a brief pause between sessions).
In conclusion, by the end of this tutorial, we will have a fully functional Agent2Agent system capable of simulating high-level collaboration among diverse AI experts. The modular design allows for easy extension. New agent roles, message types, or decision criteria can be plugged in with minimal changes, making the framework adaptable to urban planning, product strategy, or risk management domains. Ultimately, this tutorial showcases the strength of Google’s Gemini models for individual generative tasks and illustrates how coordinated, structured AI-to-AI dialogue can yield robust, data-driven solutions to multifaceted problems.
Comments
Post a Comment