On this tutorial, we introduce the Gemini Agent Community Protocol, a strong and versatile framework designed to allow clever collaboration amongst specialised AI brokers. Leveraging Google’s Gemini fashions, the protocol facilitates dynamic communication between brokers, every outfitted with distinct roles: Analyzer, Researcher, Synthesizer, and Validator. Customers will study to arrange and configure an asynchronous agent community, enabling automated job distribution, collaborative problem-solving, and enriched dialogue administration. Superb for situations similar to in-depth analysis, complicated knowledge evaluation, and data validation, this framework empowers customers to harness collective AI intelligence effectively.
import asyncio
import json
import random
from dataclasses import dataclass, asdict
from typing import Dict, Listing, Elective, Any
from enum import Enum
import google.generativeai as genaiWe leverage asyncio for concurrent execution, dataclasses for structured message administration, and Google’s Generative AI (google.generativeai) to facilitate interactions amongst a number of AI-driven brokers. It contains utilities for dynamic message dealing with and structured agent roles, enhancing scalability and adaptability in collaborative AI duties.
API_KEY = None
attempt:
import google.colab
IN_COLAB = True
besides ImportError:
IN_COLAB = FalseWe initialize the API_KEY and detect whether or not the code is working in a Colab surroundings. If the google.colab module is efficiently imported, the IN_COLAB flag is ready to True; in any other case, it defaults to False, permitting the script to regulate habits accordingly.
class AgentType(Enum):
ANALYZER = "analyzer"
RESEARCHER = "researcher"
SYNTHESIZER = "synthesizer"
VALIDATOR = "validator"
@dataclass
class Message:
sender: str
receiver: str
content material: str
msg_type: str
metadata: Dict = NoneTry the Pocket book
We outline the core constructions for agent interplay. The AgentType enum categorizes brokers into 4 distinct roles, Analyzer, Researcher, Synthesizer, and Validator, every with a selected perform within the collaborative community. The Message dataclass represents the format for inter-agent communication, encapsulating sender and receiver IDs, message content material, kind, and optionally available metadata.
class GeminiAgent:
def __init__(self, agent_id: str, agent_type: AgentType, community: 'AgentNetwork'):
self.id = agent_id
self.kind = agent_type
self.community = community
self.mannequin = genai.GenerativeModel('gemini-2.0-flash')
self.inbox = asyncio.Queue()
self.context_memory = []
self.system_prompts = {
AgentType.ANALYZER: "You're a knowledge analyzer. Break down complicated issues into elements and determine key patterns.",
AgentType.RESEARCHER: "You're a researcher. Collect data and supply detailed context on matters.",
AgentType.SYNTHESIZER: "You're a synthesizer. Mix data from a number of sources into coherent insights.",
AgentType.VALIDATOR: "You're a validator. Verify accuracy and consistency of knowledge and conclusions."
}
async def process_message(self, message: Message):
"""Course of incoming message and generate response"""
if not API_KEY:
return "❌ API key not configured. Please set API_KEY variable."
immediate = f"""
{self.system_prompts[self.type]}
Context from earlier interactions: {json.dumps(self.context_memory[-3:], indent=2)}
Message from {message.sender}: {message.content material}
Present a targeted response (max 100 phrases) that provides worth to the community dialogue.
"""
attempt:
response = await asyncio.to_thread(
self.mannequin.generate_content, immediate
)
return response.textual content.strip()
besides Exception as e:
return f"Error processing: {str(e)}"
async def send_message(self, receiver_id: str, content material: str, msg_type: str = "job"):
"""Ship message to a different agent"""
message = Message(self.id, receiver_id, content material, msg_type)
await self.community.route_message(message)
async def broadcast(self, content material: str, exclude_self: bool = True):
"""Broadcast message to all brokers in community"""
for agent_id in self.community.brokers:
if exclude_self and agent_id == self.id:
proceed
await self.send_message(agent_id, content material, "broadcast")
async def run(self):
"""Fundamental agent loop"""
whereas True:
attempt:
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 material": message.content material,
"my_response": response
})
if len(self.context_memory) > 10:
self.context_memory = self.context_memory[-10:]
print(f"🤖 {self.id} ({self.kind.worth}): {response}")
if random.random() Try the Pocket book
The GeminiAgent class defines the habits and capabilities of every agent within the community. Upon initialization, it assigns a novel ID, position kind, and a reference to the agent community and hundreds the Gemini 2.0 Flash mannequin. It makes use of role-specific system prompts to generate clever responses based mostly on incoming messages, that are processed asynchronously by way of a queue. Every agent maintains a context reminiscence to retain latest interactions and may both reply straight, ship focused messages, or broadcast insights to others. The run() methodology constantly processes messages, promotes collaboration by sometimes initiating responses to different brokers, and manages message dealing with in a non-blocking loop.
class AgentNetwork:
def __init__(self):
self.brokers: Dict[str, GeminiAgent] = {}
self.message_log = []
self.working = False
def add_agent(self, agent_type: AgentType, agent_id: Elective[str] = None):
"""Add new agent to community"""
if not agent_id:
agent_id = f"{agent_type.worth}_{len(self.brokers)+1}"
agent = GeminiAgent(agent_id, agent_type, self)
self.brokers[agent_id] = agent
print(f"✅ Added {agent_id} to community")
return agent_id
async def route_message(self, message: Message):
"""Route message to focus on agent"""
self.message_log.append(asdict(message))
if message.receiver in self.brokers:
await self.brokers[message.receiver].inbox.put(message)
else:
print(f"⚠️ Agent {message.receiver} not discovered")
async def initiate_task(self, job: str):
"""Begin a collaborative job"""
print(f"🚀 Beginning job: {job}")
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], job, "job")
await self.route_message(initial_message)
async def run_network(self, period: int = 30):
"""Run the agent community for specified period"""
self.working = True
print(f"🌐 Beginning agent community for {period} seconds...")
agent_tasks = [agent.run() for agent in self.agents.values()]
attempt:
await asyncio.wait_for(asyncio.collect(*agent_tasks), timeout=period)
besides asyncio.TimeoutError:
print("⏰ Community session accomplished")
lastly:
self.working = FalseTry the Pocket book
The AgentNetwork class manages the coordination and communication between all brokers within the system. It permits dynamic addition of brokers with distinctive IDs and specified roles, maintains a log of all exchanged messages, and facilitates message routing to the right recipient. The community can provoke a collaborative job by sending the beginning message to an Analyzer agent, and runs the total asynchronous occasion loop for a specified period, enabling brokers to function concurrently and interactively inside a shared surroundings.
async def demo_agent_network():
"""Display the Gemini Agent Community Protocol"""
community = AgentNetwork()
community.add_agent(AgentType.ANALYZER, "deep_analyzer")
community.add_agent(AgentType.RESEARCHER, "info_gatherer")
community.add_agent(AgentType.SYNTHESIZER, "insight_maker")
community.add_agent(AgentType.VALIDATOR, "fact_checker")
job = "Analyze the potential affect of quantum computing on cybersecurity"
network_task = asyncio.create_task(community.run_network(20))
await asyncio.sleep(1)
await community.initiate_task(job)
await network_task
print(f"n📊 Community accomplished with {len(community.message_log)} messages exchanged")
agent_participation = {support: sum(1 for msg in community.message_log if msg['sender'] == support)
for support in community.brokers}
print("Agent participation:", agent_participation)
def setup_api_key():
"""Interactive API key setup"""
international API_KEY
if IN_COLAB:
from google.colab import userdata
attempt:
API_KEY = userdata.get('GEMINI_API_KEY')
genai.configure(api_key=API_KEY)
print("✅ API key loaded from Colab secrets and techniques")
return True
besides:
print("💡 To make use of Colab secrets and techniques: Add 'GEMINI_API_KEY' within the secrets and techniques panel")
print("🔑 Please enter your Gemini API key:")
print(" Get it from: https://makersuite.google.com/app/apikey")
attempt:
if IN_COLAB:
from google.colab import userdata
API_KEY = enter("Paste your API key right here: ").strip()
else:
import getpass
API_KEY = getpass.getpass("Paste your API key right here: ").strip()
if API_KEY and len(API_KEY) > 10:
genai.configure(api_key=API_KEY)
print("✅ API key configured efficiently!")
return True
else:
print("❌ Invalid API key")
return False
besides KeyboardInterrupt:
print("n❌ Setup cancelled")
return FalseTry the Pocket book
The demo_agent_network() perform orchestrates all the agent workflow: it initializes an agent community, provides 4 role-specific brokers, launches a cybersecurity job, and runs the community asynchronously for a set period whereas monitoring message exchanges and agent participation. In the meantime, setup_api_key() gives an interactive mechanism to securely configure the Gemini API key, with tailor-made logic for each Colab and non-Colab environments, guaranteeing the AI brokers can talk with the Gemini mannequin backend earlier than the demo begins.
if __name__ == "__main__":
print("🧠 Gemini Agent Community Protocol")
print("=" * 40)
if not setup_api_key():
print("❌ Can't run with out legitimate API key")
exit()
print("n🚀 Beginning 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())
Lastly, the above code serves because the entry level for executing the Gemini Agent Community Protocol. It begins by prompting the consumer to arrange the Gemini API key, exiting if not offered. Upon profitable configuration, the demo is launched. If working in Google Colab, it applies nest_asyncio to deal with Colab’s occasion loop restrictions; in any other case, it makes use of Python’s native asyncio.run() to execute the asynchronous demo of agent collaboration.
In conclusion, by finishing this tutorial, customers achieve sensible data of implementing an AI-powered collaborative community utilizing Gemini brokers. The hands-on expertise offered right here demonstrates how autonomous brokers can successfully break down complicated issues, collaboratively generate insights, and make sure the accuracy of knowledge by way of validation.
Try the Pocket book. All credit score for this analysis goes to the researchers of this venture. Additionally, be happy to comply with us on Twitter and don’t overlook to hitch our 99k+ ML SubReddit and Subscribe to our Publication.
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.
