In this tutorial, 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 genaiimport jsonimport timefrom dataclasses import dataclassfrom typing import Dict, List, Anyfrom enum import Enumimport randomimport reAPI_KEY = "Use Your Own API Key" genai.configure(api_key=API_KEY)Check out the full Notebook here
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"Check out the full Notebook here
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.
@dataclassclass A2AMessage: sender_id: str receiver_id: str message_type: MessageType payload: Dict[str, Any] timestamp: float priority: int = 1Check out the full Notebook here
This 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 perspective2. Provide constructive feedback to other agents3. Synthesize information from multiple sources4. Make data-driven decisions5. Collaborate effectively while maintaining your expertiseIMPORTANT: 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}")Check out the full Notebook here
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_solutionsCheck out the full Notebook here
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 agentsCheck out the full Notebook here
The 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()Check out the full Notebook here
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.
Check out the Notebook here. All credit for this research goes to the researchers of this project. Also, feel free to follow us on Twitter and don’t forget to join our 95k+ ML SubReddit and Subscribe to our Newsletter.
The post A Step-by-Step Coding Implementation of an Agent2Agent Framework for Collaborative and Critique-Driven AI Problem Solving with Consensus-Building appeared first on MarkTechPost.
