Step-by-step coding implementation of the proxy framework to solve collaboration and criticism-driven AI problems through consensus building
In this tutorial, we implemented the Agent2agent collaboration framework established on Google’s Gemini model. This guide introduces the creation of professional AI roles, from data scientists and product strategists to risk analysts and creative innovators. It shows how these agents exchange structured messages to address complex real-world challenges. Through well-defined roles, personality, and communication protocols, the tutorial highlights how to perform multi-agent problem solving in three stages: personal analysis, cross-subject criticism and a 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)
View the full notebook here
We import the core library for building your proxy system, handling JSON, timing, data structures, and regular utilities. We then set up your Gemini API key and initialize the Genai client for subsequent calls. This ensures that all subsequent requests from Google generates AI endpoints are authenticated.
class MessageType(Enum):
HANDSHAKE = "handshake"
TASK_PROPOSAL = "task_proposal"
ANALYSIS = "analysis"
CRITIQUE = "critique"
SYNTHESIS = "synthesis"
VOTE = "vote"
CONSENSUS = "consensus"View the full notebook here
The MessageType enumeration defines the stages of Agent 2agent communication, from initial handshake and task suggestions to analysis, criticism, synthesis, voting, and final consensus. It allows you to tag and route messages based on their role in the collaboration workflow.
@dataclass
class A2AMessage:
sender_id: str
receiver_id: str
message_type: MessageType
payload: Dict[str, Any]
timestamp: float
priority: int = 1View the full notebook here
This A2Amessage Dataclass encapsulates all metadata required for inter-agent communication, tracks who sends it, who should receive it, the role of the message in the protocol (Message_Type), its content (payload), when to send (TIMESTAMP) (TIMESTAMP), and its relative processing priority. It provides a structured, secure way to serialize and route messages between brokers.
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}")View the full notebook here
The Gemini class wraps Google Gemini model instances, encapsulating the identity, role, and personality of each agent to generate a structured JSON response. It provides helper methods for building system prompts to call the API with controlled temperature and token restrictions, and then return to the default response format in case of Parse or API errors. With Analysis_task, Critique_analysis, and Synthesize_solutions, it simplifies every stage of a 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_solutionsView the full notebook here
The Agent2AgentCollaboriveSystem class manages your Gemini toxin instances, provides a method to register a new agent, and coordinates collaborative workflows in four phases, personal analysis, cross-agent criticism, solution synthesis and consensus scores. It handles logging and printing intermediate results and returns the final suggested solution for downstream use by each agent.
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
View the full notebook here
The create_specialized_gemini_agents function instantiates a balanced team of five Gemini agents, each with unique roles, personality and temperature settings. These agents cover analytics, product strategies, system architecture, creative innovation and risk management to ensure comprehensive and collaborative solutions to solve problems.
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:
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 View the full notebook here
Finally, the run_gemini_agent2agent_demo function connects everything together: it prints an overview title, makes sure to set up our Gemini API key, register five dedicated agents, and then perform a session that collaborates on each scheduled challenge (a brief pause between sessions).
In short, by the end of this tutorial, we will have a fully-featured Agent 2 that can simulate advanced collaboration between different AI experts. Modular design can be easily expanded. New proxy roles, message types, or decision criteria can be inserted with minimal changes, thus adapting the framework to urban planning, product strategy, or risk management domains. Ultimately, the tutorial demonstrates the strength of Google’s Gemini model for a single generation task and illustrates how coordinated, structured AI-to-AI dialogue can generate powerful, data-driven solutions for multifaceted problems.
Check out the notebook here. All credits for this study are to the researchers on the project. Also, please stay tuned for us twitter And don’t forget to join us 95k+ ml reddit And subscribe Our newsletter.
Asif Razzaq is CEO of Marktechpost Media Inc. As a visionary entrepreneur and engineer, ASIF is committed to harnessing the potential of artificial intelligence to achieve social benefits. His recent effort is to launch Marktechpost, an artificial intelligence media platform that has an in-depth coverage of machine learning and deep learning news that can sound both technically, both through technical voices and be understood by a wide audience. The platform has over 2 million views per month, demonstrating its popularity among its audience.
