How to build asynchronous Agent network using Gemini to research, analyzing, and verification activities
In this study, we present the Gemini Agent Network Protocol, a powerful and convertible framework designed to enable understanding of a discreetly between special agents in AI. Ginini's Gemini models, Protocol helps strong links between agents, each is included in different fields: Analyzer, researcher, synthesizer, and validator. Users will learn to set and prepare for asynchronous agent network, which allows the default distribution of the operating work, partnerships with discussion management problems. It is good for such conditions as serious research, complex analysis of data, and the verification of information, this framework gives us the power of the ability to meet the Ai Intelligence.
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 find the asyncio's asyncio of the same execution, the DACKassS Management of organized messages, and the productive AI of Google (Google.Generatious) to facilitate interactions between many AI driven agents conducted by AI. Includes powerful management services and organized agents, developing fluctuations and fluctuations in AI cooperation.
API_KEY = None
try:
import google.colab
IN_COLAB = True
except ImportError:
IN_COLAB = FalseWe start API_KEY and we find that the code works in the colob area. If the Google Module.Colab is successfully imported, the in_Colab flag is scheduled for the truth; Besides, it drops lies, allowing the script to change moral decisions properly.
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 = NoneSee a letter of writing
We describe basic agencies for agent interactions. Agenttype ENUM separates the agents of four different roles, researcher, researcher, synthesizer, and the vitator, each has a specific job on a cooperative network. Message Dataclass represents the medium of the Agent communication format, an approved sender and receiver ID, message content, the preferred 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() < 0.3:
other_agents = [aid for aid in self.network.agents.keys() if aid != self.id]
if other_agents:
target = random.choice(other_agents)
await self.send_message(target, f"Building on that: {response[:50]}...")
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"❌ Error in {self.id}: {e}")See a letter of writing
The Geminagent section describes the behavior and skills of each agent on the network. When starting, it gives a unique ID, a different kind, and reference is a reference to an agent network and loaded the Gemini 2.0 Flash model. It uses the passage to the role moves producing wise answers based on income, audio-processing in asynchronong. Each agent stores the maintenance memory memory for the latest partnerships and can respond directly, send referred messages, or to spread understanding of others. The running method () continues to process the messages, improves interventions from time to time starting answers to other agents, and regulates the message management to the weekly 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 = FalseSee a letter of writing
The agentnetwork section treats coordinates and communication between all agents in the system. Allows the installation of dynamic agents with unique IDs and the specified fields, keeping the log of all modified messages, and helps a message route to the relevant recipient. The network can start a collective team by sending the first message to analalyzer's agent, and conducts asynchronous the loop event during the specified period of time, enabling agents at the center of the shared area.
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:
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 FalseSee a letter of writing
The demo_Nigent_Network () The operating work of the Agent's Function: Starting AGENT Network, adding a specific agent's work, and conducts a specific term of the appointment during the transfer of the messages and messaging. At that time, sets_api_key () Provide effective equipment for safety for the safety of the Gemino API system, with its circumstances in both colab and non-Colab, to look after the Gemini Model Backend before the commutation 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 a Gemini Agent Agent Agent Network. It starts by pulling the user to set the Gemini API key, which comes out when given. When successfully established, shortening. If applicable to Google Colab, the nest running a nest_asyndio to manage ColoB event limits; Besides, it uses Python's Nave Asyncio.run () to make asynchronous demo worker.
In conclusion, by completing this lesson, users receive practical information to use a powerful network that uses Gemini. The manual experience provided here shows that independent agents may be successfully separate from complex problems, produce discretion, and ensure the accuracy of information on verification.
See a letter of writing. All credit for this study goes to research for this project. Also, feel free to follow it Sane and don't forget to join ours 99k + ml subreddit Then sign up for Our newspaper.
Asphazzaq is a Markteach Media Inc. According to a View Business and Developer, Asifi is committed to integrating a good social intelligence. His latest attempt is launched by the launch of the chemistrylife plan for an intelligence, MarktechPost, a devastating intimate practice of a machine learning and deep learning issues that are clearly and easily understood. The platform is adhering to more than two million moon visits, indicating its popularity between the audience.
