This tutorial introduces Gemini Agent Community Protocol, a robust and versatile framework designed to allow clever collaboration between specialised AI brokers. Utilizing Google’s Gemini mannequin, the protocol promotes dynamic communication between brokers. Every has a definite position as an analyzer, researcher, synthesizer, and validator. Customers will discover ways to arrange and configure an asynchronous agent community that permits automated activity distribution, collaborative drawback fixing, and enriched dialogue administration. Excellent for eventualities resembling in-depth analysis, advanced information evaluation, and knowledge verification, this framework permits customers to effectively make the most of collective AI intelligence.
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 genaiIt leverages Asyncio for concurrency, leveraging information courses for structured message administration, and Google’s Generated AI (Google.genericai) to advertise interplay between a number of AI-driven brokers. This consists of dynamic message processing and structured agent position utility, which will increase scalability and suppleness for collaborative AI duties.
API_KEY = None
strive:
import google.colab
IN_COLAB = True
besides ImportError:
IN_COLAB = FalseInitializes API_KEY to detect if the code is working in a colab surroundings. 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 its conduct 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 = NonePlease examine Notes
Defines the core construction of agent interactions. AgentType Enum classifies brokers into 4 completely different roles: analyzers, researchers, synthesizers, and validators, every with particular capabilities in a collaborative community. Message Dataclass represents the format of agent-to-agent communication, sender and recipient ID encapsulation, 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 information analyzer. Break down advanced issues into elements and determine key patterns.",
AgentType.RESEARCHER: "You're a researcher. Collect info and supply detailed context on matters.",
AgentType.SYNTHESIZER: "You're a synthesizer. Mix info from a number of sources into coherent insights.",
AgentType.VALIDATOR: "You're a validator. Verify 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.type]}
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 = "activity"):
"""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):
"""Foremost 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.kind.worth}): {response}")
if random.random() < 0.3:
other_agents = [aid for aid in self.network.agents.keys() if aid != self.id]
if other_agents:
goal = random.selection(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}")Please examine Notes
The Geminiagent class defines the conduct and capabilities of every agent in a community. At initialization, it assigns a novel ID, position kind, and references to the agent community and hundreds the Gemini 2.0 Flash mannequin. Use role-specific system prompts to generate clever responses based mostly on incoming messages. That is processed asynchronously via the queue. Every agent maintains contextual reminiscence to protect latest interactions, permitting them to reply immediately, ship goal messages, and broadcast insights to others. The run() technique promotes collaboration by constantly processing messages and infrequently beginning responses to different brokers, and manages message processing in non-blocking teams.
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: 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, activity: str):
"""Begin a collaborative activity"""
print(f"🚀 Beginning activity: {activity}")
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], activity, "activity")
await self.route_message(initial_message)
async def run_network(self, length: int = 30):
"""Run the agent community for specified length"""
self.working = True
print(f"🌐 Beginning agent community for {length} seconds...")
agent_tasks = [agent.run() for agent in self.agents.values()]
strive:
await asyncio.wait_for(asyncio.collect(*agent_tasks), timeout=length)
besides asyncio.TimeoutError:
print("⏰ Community session accomplished")
lastly:
self.working = FalsePlease examine Notes
The AgentNetwork class manages coordination and communication between all brokers within the system. It permits for dynamic addition of brokers with distinctive IDs and specified roles, maintains logs of all exchanged messages, and facilitates message routing to the proper recipients. The community can begin collaboration by sending a begin message to the analyzer agent, working a full asynchronous occasion loop for a specified time frame, permitting the agent to function concurrently interactively inside the shared surroundings.
async def demo_agent_network():
"""Reveal 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")
activity = "Analyze the potential impression of quantum computing on cybersecurity"
network_task = asyncio.create_task(community.run_network(20))
await asyncio.sleep(1)
await community.initiate_task(activity)
await network_task
print(f"n📊 Community accomplished with {len(community.message_log)} messages exchanged")
agent_participation = {assist: sum(1 for msg in community.message_log if msg['sender'] == assist)
for assist in community.brokers}
print("Agent participation:", agent_participation)
def setup_api_key():
"""Interactive API key setup"""
world 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 FalsePlease examine Notes
The demo_agent_network() operate coordinates the complete agent workflow. Initialize the agent community, add 4 role-specific brokers, launch cybersecurity duties, and run the community asynchronously over a set interval, monitoring message exchanges and agent participation. In the meantime, setup_api_key() supplies an interactive mechanism to securely configure Gemini API keys, utilizing logic tailor-made to each colab and non-colab environments, permitting AI brokers to speak 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 as an entry level for working the Gemini Agent Community Protocol. It begins with asking the consumer to arrange a Gemini API key and ends if it isn’t supplied. If the configuration is profitable, the demo will begin. If working on Google Colab, apply nest_asyncio to deal with Colab’s occasion loop limits. In any other case, run an asynchronous demo of agent collaboration utilizing Python’s native Asyncio.run().
In conclusion, by finishing this tutorial, customers will achieve sensible data about implementing AI-powered collaborative networks utilizing Gemini brokers. The sensible expertise supplied right here demonstrates how autonomous brokers can successfully break down advanced issues, collaboratively generate insights, and guarantee info accuracy via verification.
Please examine Notes. All credit for this examine can be directed to researchers on this venture. Additionally, please be happy to observe us Twitter And do not forget to affix us 99k+ ml subreddit And subscribe Our Newsletter.
Asif Razzaq is CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, ASIF is dedicated to leveraging the probabilities of synthetic intelligence for social advantages. His newest efforts are the launch of MarkTechPost, a synthetic intelligence media platform. That is distinguished by its detailed protection of machine studying and deep studying information, and is simple to know by a technically sound and extensive viewers. The platform has over 2 million views every month, indicating its reputation amongst viewers.
