En este tutorial, presentamos el Protocolo de la Red Gemini Agent, un marco poderoso y flexible diseñado para permitir una colaboración inteligente entre los agentes de IA especializados. Aprovechando los modelos Gemini de Google, el protocolo facilita la comunicación dinámica entre los agentes, cada uno equipado con roles distintos: analizador, investigador, sintetizador y validador. Los usuarios aprenderán a configurar y configurar una red de agentes asíncronos, permitiendo la distribución automatizada de tareas, la resolución de problemas de colaboración y la gestión de diálogo enriquecido. Ideal para escenarios como investigación en profundidad, análisis de datos complejos y validación de información, este marco faculta a los usuarios aprovechar de manera eficiente la inteligencia colectiva de IA.
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 genaiAprovechamos Asyncio para la ejecución concurrente, los datos de datos para la gestión de mensajes estructurados y la IA generativa de Google (Google.Generativeai) para facilitar las interacciones entre múltiples agentes impulsados por la IA. Incluye utilidades para el manejo dinámico de mensajes y roles de agente estructurado, mejorando la escalabilidad y la flexibilidad en las tareas de IA colaborativas.
API_KEY = None
try:
import google.colab
IN_COLAB = True
except ImportError:
IN_COLAB = FalseInicializamos el API_KEY y detectamos si el código se está ejecutando en un entorno de Colab. Si el módulo Google.Colab se importa correctamente, el indicador in_colab se establece en verdadero; De lo contrario, el valor predeterminado es falso, lo que permite que el script ajuste el comportamiento en consecuencia.
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 = NoneMira el Computadora portátil
Definimos las estructuras centrales para la interacción del agente. El AgentType Enum clasifica a los agentes en cuatro roles distintos, analizador, investigador, sintetizador y validador, cada uno con una función específica en la red colaborativa. El mensaje DataClass representa el formato para la comunicación entre agentes, encapsulando las ID de remitente y receptor, contenido de mensajes, tipo y metadatos opcionales.
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}")Mira el Computadora portátil
La clase geminiagente define el comportamiento y las capacidades de cada agente en la red. Tras la inicialización, asigna una identificación única, tipo de rol y una referencia a la red de agentes y carga el modelo Gemini 2.0 Flash. Utiliza indicaciones del sistema específicas de roles para generar respuestas inteligentes basadas en mensajes entrantes, que se procesan de forma asincrónica a través de una cola. Cada agente mantiene una memoria de contexto para retener interacciones recientes y puede responder directamente, enviar mensajes específicos o transmitir información a otros. El método Run () procesa continuamente los mensajes, promueve la colaboración al iniciar ocasionalmente las respuestas a otros agentes y administra el manejo de mensajes en un bucle sin bloqueo.
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 = FalseMira el Computadora portátil
La clase Agentnetwork gestiona la coordinación y la comunicación entre todos los agentes del sistema. Permite la adición dinámica de agentes con IDS únicos y roles especificados, mantiene un registro de todos los mensajes intercambiados y facilita el enrutamiento de mensajes al destinatario correcto. La red puede iniciar una tarea de colaboración enviando el mensaje inicial a un agente analizador, y ejecuta el bucle de evento asíncrono completo para una duración específica, lo que permite a los agentes operar de manera simultánea e interactiva dentro de un entorno compartido.
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: https://makersuite.google.com/app/apikey")
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 FalseMira el Computadora portátil
La función demo_agent_network () orquesta todo el flujo de trabajo del agente: inicializa una red de agente, agrega cuatro agentes específicos de roles, lanza una tarea de seguridad cibernética y ejecuta la red de forma asincrónica durante una duración fija mientras rastrea los intercambios de mensajes y la participación de los agentes. Mientras tanto, SETUP_API_KEY () proporciona un mecanismo interactivo para configurar de forma segura la clave de la API de Gemini, con lógica personalizada para entornos Colab y no COLAB, asegurando que los agentes de IA puedan comunicarse con el modelo de Gemini Backend antes de que comience la demostración.
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())
Finalmente, el código anterior sirve como punto de entrada para ejecutar el protocolo de red de Gemini Agent. Comienza incitando al usuario a configurar la clave de la API Gemini, saliendo si no se proporciona. Tras una configuración exitosa, se inicia la demostración. Si se ejecuta en Google Colab, aplica Nest_asyncio para manejar las restricciones de bucle de eventos de Colab; De lo contrario, utiliza el asyncio.run () nativo de Python para ejecutar la demostración asíncrona de la colaboración del agente.
En conclusión, al completar este tutorial, los usuarios obtienen un conocimiento práctico de implementar una red colaborativa con IA utilizando agentes de Gemini. La experiencia práctica proporcionada aquí demuestra cómo los agentes autónomos pueden romper efectivamente problemas complejos, generar ideas en colaboración y garantizar la precisión de la información a través de la validación.
Mira el Computadora portátil. Todo el crédito por esta investigación va a los investigadores de este proyecto. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro 99k+ ml de subreddit y suscribirse a Nuestro boletín.
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, MarktechPost, que se destaca por su cobertura profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.