On this tutorial, we introduce the Gemini Agent Community Protocol, a robust 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 eventualities comparable to in-depth analysis, advanced information 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, Record, Non-obligatory, 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 consists of utilities for dynamic message dealing with and structured agent roles, enhancing scalability and suppleness in collaborative AI duties.
API_KEY = None
strive:
import google.colab
IN_COLAB = True
besides ImportError:
IN_COLAB = FalseWe initialize the API_KEY and detect whether or not the code is operating in a Colab atmosphere. If the google.colab module is efficiently imported, the IN_COLAB flag is about 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 buildings 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, sort, and non-obligatory metadata.
class GeminiAgent:
def __init__(self, agent_id: str, agent_type: AgentType, community: 'AgentNetwork'):
self.id = agent_id
self.sort = 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 information analyzer. Break down advanced issues into parts 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. Test accuracy and consistency of data 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.sort)}
Context from earlier interactions: {json.dumps(self.context_memory(-3:), indent=2)}
Message from {message.sender}: {message.content material}
Present a centered response (max 100 phrases) that provides worth to the community dialogue.
"""
strive:
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):
"""Important agent loop"""
whereas True:
strive:
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.sort.worth}): {response}")
if random.random() < 0.3:
other_agents = (support for support in self.community.brokers.keys() if support != self.id)
if other_agents:
goal = random.alternative(other_agents)
await self.send_message(goal, f"Constructing on that: {response(:50)}...")
besides asyncio.TimeoutError:
proceed
besides Exception as e:
print(f"❌ Error in {self.id}: {e}")Try the Pocket book
The GeminiAgent class defines the habits and capabilities of every agent within the community. Upon initialization, it assigns a singular ID, position sort, 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 current interactions and might both reply instantly, ship focused messages, or broadcast insights to others. The run() methodology repeatedly 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.operating = False
def add_agent(self, agent_type: AgentType, agent_id: Non-obligatory(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 = (support for support, agent in self.brokers.objects()
if agent.sort == 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, length: int = 30):
"""Run the agent community for specified length"""
self.operating = True
print(f"🌐 Beginning agent community for {length} seconds...")
agent_tasks = (agent.run() for agent in self.brokers.values())
strive:
await asyncio.wait_for(asyncio.collect(*agent_tasks), timeout=length)
besides asyncio.TimeoutError:
print("⏰ Community session accomplished")
lastly:
self.operating = 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 proper recipient. The community can provoke a collaborative job by sending the beginning message to an Analyzer agent, and runs the complete asynchronous occasion loop for a specified length, enabling brokers to function concurrently and interactively inside a shared atmosphere.
async def demo_agent_network():
"""Exhibit 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
strive:
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")
strive:
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 the complete agent workflow: it initializes an agent community, provides 4 role-specific brokers, launches a cybersecurity job, and runs the community asynchronously for a hard and fast length whereas monitoring message exchanges and agent participation. In the meantime, setup_api_key() offers 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 supplied. Upon profitable configuration, the demo is launched. If operating 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 information of implementing an AI-powered collaborative community utilizing Gemini brokers. The hands-on expertise supplied right here demonstrates how autonomous brokers can successfully break down advanced issues, collaboratively generate insights, and make sure the accuracy of data by way of validation.
Try the Pocket book. All credit score for this analysis goes to the researchers of this undertaking. Additionally, be happy to comply with us on Twitter and don’t neglect 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 recognition amongst audiences.
