In this tutorial, we present the Gemini Agent Network protocol, a powerful and flexible framework designed to allow intelligent collaboration between specialized AI agents. Taking advantage of Google Gemini models, the protocol facilitates dynamic communication between agents, each equipped with distinct roles: analyzer, researcher, synthesizer and validator. Users will learn to configure and configure an asynchronous agent network, allowing automated task distribution, collaborative problem solving and enriched dialogs management. Ideal for scenarios such as in -depth research, complex data analysis and information validation, this framework allows users to effectively transport the collective intelligence of AI.
import asyncio
import json
import random
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional, Any
from enum import Enum
import google.generativeai as genaiWe take advantage of Asyncio for simultaneous execution, data classes for the management of structured messages and a generative AI of Google (Google.generativateai) to facilitate interactions between several agents led by AI. It includes public services for the management of dynamic messages and structured agent roles, improving scalability and flexibility in collaborative AI tasks.
API_KEY = None
try:
import google.colab
IN_COLAB = True
except ImportError:
IN_COLAB = FalseWe initialize the API_Key and detect if the code runs in a colaab environment. If the Google.colab module is successfully imported, the in_colab indicator is defined on True; Otherwise, it is by default false, allowing the script to adjust the behavior accordingly.
class AgentType(Enum):
ANALYZER = "analyzer"
RESEARCHER = "researcher"
SYNTHESIZER = "synthesizer"
VALIDATOR = "validator"
@dataclass
class Message:
sender: str
receiver: str
content: str
msg_type: str
metadata: Dict = NoneDiscover the Notebook
We define the central structures for the interaction of the agent. The Enum agentype classifies the agents in four distinct roles, analyzer, researcher, synthesizer and validator, each with a specific function in the collaborative network. The message database represents the inter-agent communication format, encapsulating the IDs of the sender and the receiver, the content of the message, the type and the optional metadata.
class GeminiAgent:
def __init__(self, agent_id: str, agent_type: AgentType, network: 'AgentNetwork'):
self.id = agent_id
self.type = agent_type
self.network = network
self.model = genai.GenerativeModel('gemini-2.0-flash')
self.inbox = asyncio.Queue()
self.context_memory = ()
self.system_prompts = {
AgentType.ANALYZER: "You are a data analyzer. Break down complex problems into components and identify key patterns.",
AgentType.RESEARCHER: "You are a researcher. Gather information and provide detailed context on topics.",
AgentType.SYNTHESIZER: "You are a synthesizer. Combine information from multiple sources into coherent insights.",
AgentType.VALIDATOR: "You are a validator. Check accuracy and consistency of information and conclusions."
}
async def process_message(self, message: Message):
"""Process incoming message and generate response"""
if not API_KEY:
return "❌ API key not configured. Please set API_KEY variable."
prompt = f"""
{self.system_prompts(self.type)}
Context from previous interactions: {json.dumps(self.context_memory(-3:), indent=2)}
Message from {message.sender}: {message.content}
Provide a focused response (max 100 words) that adds value to the network discussion.
"""
try:
response = await asyncio.to_thread(
self.model.generate_content, prompt
)
return response.text.strip()
except Exception as e:
return f"Error processing: {str(e)}"
async def send_message(self, receiver_id: str, content: str, msg_type: str = "task"):
"""Send message to another agent"""
message = Message(self.id, receiver_id, content, msg_type)
await self.network.route_message(message)
async def broadcast(self, content: str, exclude_self: bool = True):
"""Broadcast message to all agents in network"""
for agent_id in self.network.agents:
if exclude_self and agent_id == self.id:
continue
await self.send_message(agent_id, content, "broadcast")
async def run(self):
"""Main agent loop"""
while True:
try:
message = await asyncio.wait_for(self.inbox.get(), timeout=1.0)
response = await self.process_message(message)
self.context_memory.append({
"from": message.sender,
"content": message.content,
"my_response": response
})
if len(self.context_memory) > 10:
self.context_memory = self.context_memory(-10:)
print(f"🤖 {self.id} ({self.type.value}): {response}")
if random.random() Discover the Notebook
The geminiating class defines the behavior and capacities of each network agent. During initialization, it attributes a unique ID, a type of role and a reference to the agent network and loads the Flash Gemini 2.0 model. It uses specific system prompts to generate intelligent responses based on incoming messages, which are treated asynchronously via a queue. Each agent maintains a context memory to maintain recent interactions and can either respond directly, send targeted messages or disseminate information to others. The RUN () method constantly treats messages, promotes collaboration by occasionally initiating responses to other agents and manages the management of messages in a non -blocking loop.
class AgentNetwork:
def __init__(self):
self.agents: Dict(str, GeminiAgent) = {}
self.message_log = ()
self.running = False
def add_agent(self, agent_type: AgentType, agent_id: Optional(str) = None):
"""Add new agent to network"""
if not agent_id:
agent_id = f"{agent_type.value}_{len(self.agents)+1}"
agent = GeminiAgent(agent_id, agent_type, self)
self.agents(agent_id) = agent
print(f"✅ Added {agent_id} to network")
return agent_id
async def route_message(self, message: Message):
"""Route message to target agent"""
self.message_log.append(asdict(message))
if message.receiver in self.agents:
await self.agents(message.receiver).inbox.put(message)
else:
print(f"⚠️ Agent {message.receiver} not found")
async def initiate_task(self, task: str):
"""Start a collaborative task"""
print(f"🚀 Starting task: {task}")
analyzer_agents = (aid for aid, agent in self.agents.items()
if agent.type == AgentType.ANALYZER)
if analyzer_agents:
initial_message = Message("system", analyzer_agents(0), task, "task")
await self.route_message(initial_message)
async def run_network(self, duration: int = 30):
"""Run the agent network for specified duration"""
self.running = True
print(f"🌐 Starting agent network for {duration} seconds...")
agent_tasks = (agent.run() for agent in self.agents.values())
try:
await asyncio.wait_for(asyncio.gather(*agent_tasks), timeout=duration)
except asyncio.TimeoutError:
print("⏰ Network session completed")
finally:
self.running = FalseDiscover the Notebook
The Agentnetwork class manages coordination and communication between all system agents. It allows the dynamic addition of agents with unique IDs and specified roles, retains a journal of all messages exchanged and facilitates the routing of messages to the correct recipient. The network can launch a collaborative task by sending the starting message to an analyzer agent and performs the full asynchronous event loop for a specified period, allowing agents to operate simultaneously and interactively in a shared environment.
async def demo_agent_network():
"""Demonstrate the Gemini Agent Network Protocol"""
network = AgentNetwork()
network.add_agent(AgentType.ANALYZER, "deep_analyzer")
network.add_agent(AgentType.RESEARCHER, "info_gatherer")
network.add_agent(AgentType.SYNTHESIZER, "insight_maker")
network.add_agent(AgentType.VALIDATOR, "fact_checker")
task = "Analyze the potential impact of quantum computing on cybersecurity"
network_task = asyncio.create_task(network.run_network(20))
await asyncio.sleep(1)
await network.initiate_task(task)
await network_task
print(f"\n📊 Network completed with {len(network.message_log)} messages exchanged")
agent_participation = {aid: sum(1 for msg in network.message_log if msg('sender') == aid)
for aid in network.agents}
print("Agent participation:", agent_participation)
def setup_api_key():
"""Interactive API key setup"""
global API_KEY
if IN_COLAB:
from google.colab import userdata
try:
API_KEY = userdata.get('GEMINI_API_KEY')
genai.configure(api_key=API_KEY)
print("✅ API key loaded from Colab secrets")
return True
except:
print("💡 To use Colab secrets: Add 'GEMINI_API_KEY' in the secrets panel")
print("🔑 Please enter your Gemini API key:")
print(" Get it from: https://makersuite.google.com/app/apikey")
try:
if IN_COLAB:
from google.colab import userdata
API_KEY = input("Paste your API key here: ").strip()
else:
import getpass
API_KEY = getpass.getpass("Paste your API key here: ").strip()
if API_KEY and len(API_KEY) > 10:
genai.configure(api_key=API_KEY)
print("✅ API key configured successfully!")
return True
else:
print("❌ Invalid API key")
return False
except KeyboardInterrupt:
print("\n❌ Setup cancelled")
return FalseDiscover the Notebook
The Demo_AGENT_NETWORK () function orchestrates the entire agent's work flow: it initializes an agent network, adds four agents specific to the role, launches a cybersecurity task and performs the network asynchronous for a fixed period while following the exchange of messages and the participation of agents. Meanwhile, Setup_Api_Key () provides an interactive mechanism to safely configure the API Gemini key, with a tailor-made logic for colab and non-colab environments, ensuring that AI agents can communicate with the gemini model backend before the start of the demo.
if __name__ == "__main__":
print("🧠 Gemini Agent Network Protocol")
print("=" * 40)
if not setup_api_key():
print("❌ Cannot run without valid API key")
exit()
print("\n🚀 Starting demo...")
if IN_COLAB:
import nest_asyncio
nest_asyncio.apply()
loop = asyncio.get_event_loop()
loop.run_until_complete(demo_agent_network())
else:
asyncio.run(demo_agent_network())
Finally, the above code serves as an entry point for the execution of the Gemini agent network protocol. He starts by inviting the user to configure the API Gemini key, leaving if it is not provided. During the successful configuration, the demo is launched. If you run in Google Colarb, it applies Nest_asyncio to manage the restrictions on buckle of colaab events; Otherwise, he uses Asyncio.run () native of Python to execute the asynchronous demo of the collaboration of agents.
In conclusion, by finishing this tutorial, users acquire a practical knowledge of the implementation of a collaborative network powered by AI using Gemini agents. The practical experience provided here shows how autonomous agents can effectively decompose complex problems, generate information in collaboration and ensure the accuracy of information by validation.
Discover the Notebook. 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 99K + ML Subreddit 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.
