In this tutorial, we implement the collaborative agent2agent frame built on the Gemini models of Google. The guide runs through the creation of specialized AI characters, ranging from scientists to data and product strategists to risk creative analysts and innovators. It shows how these agents can exchange structured messages to meet complex and real challenges. By defining clear roles, personalities and communication protocols, the tutorial highlights how to orchestrate the resolution of multi-agent problems in three phases: individual analysis, criticism of crossed agent 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)
Discover the complete Notebook here
We import the basic libraries for the creation of your agent2agent system, JSON management, synchronization, data structures and regex utilities. Then we define your Gemini API key and initialize the Genai client for subsequent calls. This guarantees that all of Google's generational ends are authenticated.
class MessageType(Enum):
HANDSHAKE = "handshake"
TASK_PROPOSAL = "task_proposal"
ANALYSIS = "analysis"
CRITIQUE = "critique"
SYNTHESIS = "synthesis"
VOTE = "vote"
CONSENSUS = "consensus"Discover the complete Notebook here
This list of messageype defines the stages of the Communication Agent2agent, initial handles and proposals for tasks to analysis, criticism, synthesis, vote and final consensus. It allows you to label and buy 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 = 1Discover the complete Notebook here
This class of A2amessage data summarizes all the metadata necessary for inter-agent communication, following who sent it, which should receive it, the role of the message in the protocol (message_type), its content (payload), when sent (horodat) and its relative processing priority. It provides a structured and secure means of serializing and buying 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}")Discover the complete Notebook here
The geminiating class wores a Google Gemini model instance, encapsulating the identity, role and personality of each agent to generate structured JSON responses. It provides assistance methods to build system prompts, call the API with controlled temperature and token limits and fold in a default response format in the event of analysis or API errors. With analysis_task, critic_analysis and synthesize_Solutions, it rationalizes each phase of the multi-agent work flow.
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_solutionsDiscover the complete Notebook here
The CLASS Agent2AGENTCENTCRABRATIVESSEMSSEM manages your fleet of geminiating instances, providing methods to record new agents and orchestrate the collaborative work flow in four phases, individual analysis, transversal criticism, synthesis of solution and the rating of consensus. He manages journalization and the printing of intermediate results and returns the final solutions of each agent 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 agents
Discover the complete Notebook here
The Create_SPECIALIZED_GEMINI_AGENTS Installed function a balanced team of five Gemini agents, each with a unique role, personality and temperature adjustment. Agents cover the analysis, product strategy, system architecture, creative innovation and risk management to ensure a well -balanced collaborative collaborative resolution.
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 Discover the complete Notebook here
Finally, the RUN_GEMINI_AGENT2AGENT_DEMO function attaches everything together: it prints an overview header, guarantees that our API Gemini key is defined, records the five specialized agents, then performs collaborative problem solving sessions on each predefined challenge (with a brief break between sessions).
In conclusion, at the end of this tutorial, we will have an entirely functional agent2age system capable of simulating high -level collaboration between various IA experts. Modular design allows easy extension. The roles of new agents, the types of messages or the decision criteria can be connected with a minimum of modifications, which makes the framework adaptable to the areas of town planning, product strategy or risk management. In the end, this tutorial presents the strength of Google Gemini models for individual generative tasks and illustrates how the coordinated AI-Ai-Ai dialogue can produce robust solutions and data focused on multifaceted problems.
Discover the Notebook here. All the merit of this research goes to researchers in this project. Also, don't hesitate to follow us Twitter And don't forget to join our 95K + ML Subdreddit and subscribe to Our newsletter.
Asif Razzaq is the CEO of Marktechpost Media Inc .. as a visionary entrepreneur and engineer, AIF undertakes to exploit the potential of artificial intelligence for social good. His most recent company is the launch of an artificial intelligence media platform, Marktechpost, which stands out from its in-depth coverage of automatic learning and in-depth learning news which are both technically solid and easily understandable by a large audience. The platform has more than 2 million monthly views, illustrating its popularity with the public.
