How to use Gemini for research, analysis and verification tasks to build an asynchronous AI proxy network
In this tutorial, we introduce the Gemini Agent Network protocol, a powerful and flexible framework designed to collaborate intelligently among professional AI agents. The protocol utilizes Google’s Gemini model to facilitate dynamic communication between agents, each equipped with a different role: analyzer, researchers, synthesizers, and validators. Users will learn to set up and configure an asynchronous proxy network to enable automated task distribution, collaborative problem solving and rich dialogue management. The framework is ideal for scenarios such as in-depth research, complex data analysis, and information verification, effectively leveraging collective 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 utilize Asyncio concurrent execution, data elements for structured message management and Google’s Generative AI (Google.generativeai) to facilitate interaction between multiple AI-drion agents. It includes utilities for dynamic message processing and structured proxy roles, enhancing scalability and flexibility in collaborative AI tasks.
API_KEY = None
try:
import google.colab
IN_COLAB = True
except ImportError:
IN_COLAB = FalseWe initialize API_KEY and check if the code is running in the COLAB environment. If the Google.Colab module has been successfully imported, set the IN_Colab flag to true; otherwise, it defaults to false, allowing the script to adjust 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 = NoneView notebook
We define the core structure of proxy interactions. The AgentType enumeration divides the agent into four different roles, analyzers, researchers, synthesizers, and validators, each with specific features in a collaborative network. The message data class represents the format of inter-agent communication, encapsulating sender ID, message content, type, and 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() View notebook
The Gemini category defines the behavior and abilities of each agent in the network. After initialization, it assigns a unique ID, role type, and reference to the proxy network and loads the Gemini 2.0 Flash model. It uses role-specific system prompts to generate intelligent responses based on incoming messages that are processed out of sync through the queue. Each agent maintains context memory to retain the most recent interaction and can respond directly, sending target messages or broadcasting insights to others. The Run() method continuously processes messages, occasionally facilitating collaboration through responses to other agents, and manages message processing 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 = FalseView notebook
The AgentNetwork class manages coordination and communication between all agents in the system. It allows dynamic addition of proxy with unique ID and specified roles, maintains logs of all exchanged messages, and facilitates routing of messages to the correct recipients. The network can initiate a collaboration task by sending a startup message to the analyzer agent and running a full asynchronous event loop for a specified duration, allowing the agent 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:
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 FalseView notebook
The demo_agent_network() function coordinates the proxy workflow: it initializes the proxy network, adds four role-specific agents, initiates network security tasks, and runs the network for a fixed duration while tracking message exchange and proxy participation. Meanwhile, setup_api_key() provides an interactive mechanism that safely configures Gemini API keys and uses logic tailored for Colab and non-color environments to ensure that AI agents can communicate with Gemini Model Backend before the demonstration begins.
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 is the entry point for executing the Gemini proxy network protocol. It first prompts the user to set the GEMINI API key and exit if it is not provided. After successful configuration, the demo was started. If run in Google Colab, it applies NEST_ASYNCIO to handle COLAB’s event loop limit; otherwise, it uses Python’s native Asyncio.run() to perform an asynchronous demonstration of proxy collaboration.
In short, by completing this tutorial, users gain practical knowledge of implementing AI-driven collaborative networks using Gemini agents. The hands-on experience provided here shows how autonomous agents can effectively break down complex problems, collaborate to generate insights, and ensure the accuracy of information through verification.
View notebook. All credits for this study are to the researchers on the project. Also, please stay tuned for us twitter And don’t forget to join us 99K+ ml reddit And subscribe Our newsletter.
Asif Razzaq is CEO of Marktechpost Media Inc. As a visionary entrepreneur and engineer, ASIF is committed to harnessing the potential of artificial intelligence to achieve social benefits. His recent effort is to launch Marktechpost, an artificial intelligence media platform that has an in-depth coverage of machine learning and deep learning news that can sound both technically, both through technical voices and be understood by a wide audience. The platform has over 2 million views per month, demonstrating its popularity among its audience.
