On this tutorial, we implement the Agent2Agent collaborative framework constructed atop Google’s Gemini fashions. The information walks by way of the creation of specialised AI personas, starting from knowledge scientists and product strategists to threat analysts and inventive innovators. It demonstrates how these brokers can alternate structured messages to deal with complicated, real-world challenges. By defining clear roles, personalities, and communication protocols, the tutorial highlights the way to orchestrate multi-agent drawback fixing in three phases: particular person evaluation, cross-agent critique, and synthesis of options.
import google.generativeai as genai
import json
import time
from dataclasses import dataclass
from typing import Dict, Listing, Any
from enum import Enum
import random
import re
API_KEY = "Use Your Personal API Key"
genai.configure(api_key=API_KEY)
Take a look at the complete Pocket book right here
We import the core libraries for constructing your Agent2Agent system, dealing with JSON, timing, knowledge constructions, and regex utilities. Then, we set your Gemini API key and initialize the genai consumer for subsequent calls. This ensures that each one subsequent requests to Google’s generative AI endpoints are authenticated.
class MessageType(Enum):
HANDSHAKE = "handshake"
TASK_PROPOSAL = "task_proposal"
ANALYSIS = "evaluation"
CRITIQUE = "critique"
SYNTHESIS = "synthesis"
VOTE = "vote"
CONSENSUS = "consensus"Take a look at the complete Pocket book right here
This MessageType enum defines the phases of Agent2Agent communication, from preliminary handshakes and activity proposals to evaluation, critique, synthesis, voting, and last consensus. It permits you to tag and route messages in accordance with their position within the collaborative workflow.
@dataclass
class A2AMessage:
sender_id: str
receiver_id: str
message_type: MessageType
payload: Dict[str, Any]
timestamp: float
precedence: int = 1Take a look at the complete Pocket book right here
This A2AMessage dataclass encapsulates all of the metadata wanted for inter-agent communication, monitoring who despatched it, who ought to obtain it, the message’s position within the protocol (message_type), its content material (payload), when it was despatched (timestamp), and its relative processing precedence. It gives a structured, type-safe solution to serialize and route messages between brokers.
class GeminiAgent:
def __init__(self, agent_id: str, position: str, character: str, temperature: float = 0.7):
self.agent_id = agent_id
self.position = position
self.character = character
self.temperature = temperature
self.conversation_memory = []
self.current_position = None
self.confidence = 0.5
self.mannequin = genai.GenerativeModel('gemini-2.0-flash')
def get_system_context(self, task_context: str = "") -> str:
return f"""You're {self.agent_id}, an AI agent in a multi-agent collaborative system.
ROLE: {self.position}
PERSONALITY: {self.character}
CONTEXT: {task_context}
You're taking part in Agent2Agent protocol communication. Your obligations:
1. Analyze issues out of your specialised perspective
2. Present constructive suggestions to different brokers
3. Synthesize info from a number of sources
4. Make data-driven selections
5. Collaborate successfully whereas sustaining your experience
IMPORTANT: All the time construction your response as JSON with these fields:
{{
"agent_id": "{self.agent_id}",
"main_response": "your major response content material",
"confidence_level": 0.8,
"key_insights": ["insight1", "insight2"],
"questions_for_others": ["question1", "question2"],
"next_action": "urged subsequent step"
}}
Keep true to your position and character whereas being collaborative."""
def generate_response(self, immediate: str, context: str = "") -> Dict[str, Any]:
"""Generate response utilizing Gemini API"""
strive:
full_prompt = f"{self.get_system_context(context)}nnPROMPT: {immediate}"
response = self.mannequin.generate_content(
full_prompt,
generation_config=genai.sorts.GenerationConfig(
temperature=self.temperature,
max_output_tokens=600,
)
)
response_text = response.textual content
json_match = re.search(r'{.*}', response_text, re.DOTALL)
if json_match:
strive:
return json.hundreds(json_match.group())
besides json.JSONDecodeError:
cross
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": "Proceed evaluation"
}
besides 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, activity: str) -> Dict[str, Any]:
immediate = f"Analyze this activity out of your {self.position} perspective: {activity}"
return self.generate_response(immediate, f"Activity Evaluation: {activity}")
def critique_analysis(self, other_analysis: Dict[str, Any], original_task: str) -> Dict[str, Any]:
analysis_summary = other_analysis.get('main_response', 'No evaluation supplied')
immediate = 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', [])}
Present constructive critique and different views out of your {self.position} experience.
"""
return self.generate_response(immediate, f"Critique Session: {original_task}")
def synthesize_solutions(self, all_analyses: Listing[Dict[str, Any]], activity: str) -> Dict[str, Any]:
analyses_summary = "n".be a part of([
f"Agent {i+1}: {analysis.get('main_response', 'No response')[:100]}..."
for i, evaluation in enumerate(all_analyses)
])
immediate = f"""
TASK: {activity}
ALL AGENT ANALYSES:
{analyses_summary}
Because the {self.position}, synthesize these views right into a complete answer.
Establish widespread themes, resolve conflicts, and suggest the perfect path ahead.
"""
return self.generate_response(immediate, f"Synthesis Section: {activity}")Take a look at the complete Pocket book right here
The GeminiAgent class wraps a Google Gemini mannequin occasion, encapsulating every agent’s id, position, and character to generate structured JSON responses. It gives helper strategies to construct system prompts, name the API with managed temperature and token limits, and fall again to a default response format in case of parse or API errors. With analyze_task, critique_analysis, and synthesize_solutions, it streamlines every part of the multi-agent workflow.
class Agent2AgentCollaborativeSystem:
def __init__(self):
self.brokers: Dict[str, GeminiAgent] = {}
self.collaboration_history: Listing[Dict[str, Any]] = []
def add_agent(self, agent: GeminiAgent):
self.brokers[agent.agent_id] = agent
print(f"🤖 Registered Gemini Agent: {agent.agent_id} ({agent.position})")
def run_collaborative_problem_solving(self, drawback: str):
print(f"n🎯 Multi-Gemini Collaborative Drawback Fixing")
print(f"🔍 Drawback: {drawback}")
print("=" * 80)
print("n📊 PHASE 1: Particular person Agent Evaluation")
initial_analyses = {}
for agent_id, agent in self.brokers.gadgets():
print(f"n🧠 {agent_id} analyzing...")
evaluation = agent.analyze_task(drawback)
initial_analyses[agent_id] = evaluation
print(f"✅ {agent_id} ({agent.position}):")
print(f" Response: {evaluation.get('main_response', 'No response')[:150]}...")
print(f" Confidence: {evaluation.get('confidence_level', 0.5):.2f}")
print(f" Key Insights: {evaluation.get('key_insights', [])}")
print(f"n🔄 PHASE 2: Cross-Agent Critique & Suggestions")
critiques = {}
agent_list = record(self.brokers.gadgets())
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 evaluation...")
critique = agent.critique_analysis(target_analysis, drawback)
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: Resolution Synthesis")
final_solutions = {}
all_analyses = record(initial_analyses.values())
for agent_id, agent in self.brokers.gadgets():
print(f"n🎯 {agent_id} synthesizing last answer...")
synthesis = agent.synthesize_solutions(all_analyses, drawback)
final_solutions[agent_id] = synthesis
print(f"🏆 {agent_id} Last Resolution:")
print(f" {synthesis.get('main_response', 'No synthesis')[:200]}...")
print(f" Confidence: {synthesis.get('confidence_level', 0.5):.2f}")
print(f" Subsequent Motion: {synthesis.get('next_action', 'No motion specified')}")
print(f"n🤝 PHASE 4: Consensus & Advice")
avg_confidence = sum(
sol.get('confidence_level', 0.5) for sol in final_solutions.values()
) / len(final_solutions)
print(f"📊 Common Resolution Confidence: {avg_confidence:.2f}")
most_confident_agent = max(
final_solutions.gadgets(),
key=lambda x: x[1].get('confidence_level', 0)
)
print(f"n🏅 Most Assured Resolution from: {most_confident_agent[0]}")
print(f"📝 Advisable Resolution: {most_confident_agent[1].get('main_response', 'No answer')}")
all_insights = []
for answer in final_solutions.values():
all_insights.lengthen(answer.get('key_insights', []))
print(f"n💡 Collective Intelligence Insights:")
for i, perception in enumerate(set(all_insights), 1):
print(f" {i}. {perception}")
return final_solutionsTake a look at the complete Pocket book right here
The Agent2AgentCollaborativeSystem class manages your fleet of GeminiAgent cases, offering strategies to register new brokers and orchestrate the four-phase collaboration workflow, particular person evaluation, cross-agent critique, answer synthesis, and consensus scoring. It handles logging and printing the intermediate outcomes and returns every agent’s last proposed options for downstream use.
def create_specialized_gemini_agents():
"""Create various Gemini brokers with totally different roles and personalities"""
brokers = [
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 brokers
Take a look at the complete Pocket book right here
The create_specialized_gemini_agents operate instantiates a balanced workforce of 5 Gemini brokers, every with a novel position, character, and temperature setting. The brokers cowl analytics, product technique, system structure, inventive innovation, and threat administration to make sure well-rounded collaborative drawback fixing.
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)
issues = [
"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, drawback in enumerate(issues, 1):
print(f"n{'🌟 COLLABORATION SESSION ' + str(i):=^80}")
collaborative_system.run_collaborative_problem_solving(drawback)
if i Take a look at the complete Pocket book right here
Lastly, the run_gemini_agent2agent_demo operate ties all the pieces collectively: it prints an outline header, ensures our Gemini API secret’s set, registers the 5 specialised brokers, after which executes collaborative problem-solving periods on every predefined problem (with a quick pause between periods).
In conclusion, by the top of this tutorial, we could have a totally useful Agent2Agent system able to simulating high-level collaboration amongst various AI consultants. The modular design permits for straightforward extension. New agent roles, message sorts, or resolution standards may be plugged in with minimal adjustments, making the framework adaptable to city planning, product technique, or threat administration domains. Finally, this tutorial showcases the energy of Google’s Gemini fashions for particular person generative duties and illustrates how coordinated, structured AI-to-AI dialogue can yield sturdy, data-driven options to multifaceted issues.
Take a look at the Pocket book right here. All credit score for this analysis goes to the researchers of this undertaking. Additionally, be happy to observe us on Twitter and don’t neglect to affix our 95k+ ML SubReddit and Subscribe to our E-newsletter.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its reputation amongst audiences.
