En este tutorial, implementamos el Protocolo de comunicación de agentes (ACP) a través de la construcción de un sistema de mensajería flexible y compatible con ACP en Python, aprovechando la API Gemini de Google para el procesamiento del lenguaje natural. Comenzando con la instalación y la configuración de la Biblioteca Google-Generativeai, el tutorial presenta abstracciones centrales, tipos de mensajes, performadores y la clase de datos ACPMessage, que estandariza la comunicación entre agentes. Al definir las clases de ACPAGENT y ACPMESSAGEBROKER, la guía demuestra cómo crear, enviar, enrutar y procesar mensajes estructurados entre múltiples agentes autónomos. A través de ejemplos de código claro, los usuarios aprenden a implementar consultas, solicitar acciones y transmisión de información, mientras mantienen hilos de conversación, reconocimientos y manejo de errores.
import google.generativeai as genai
import json
import time
import uuid
from enum import Enum
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
GEMINI_API_KEY = "Use Your Gemini API Key"
genai.configure(api_key=GEMINI_API_KEY)
Importamos módulos de pitón esenciales, que van desde el manejo y el tiempo de JSON hasta las anotaciones de generación y tipo de identificación únicos, para admitir una implementación de ACP estructurada. Luego recupera el marcador de posición Gemini API de Gemini del usuario y configura el cliente Google-Generativeai para las llamadas posteriores al modelo de idioma Gemini.
class ACPMessageType(Enum):
"""Standard ACP message types"""
REQUEST = "request"
RESPONSE = "response"
INFORM = "inform"
QUERY = "query"
SUBSCRIBE = "subscribe"
UNSUBSCRIBE = "unsubscribe"
ERROR = "error"
ACK = "acknowledge"La enumeración de ACPMessageType define las categorías de mensajes centrales utilizadas en el protocolo de comunicación del agente, incluidas las solicitudes, las respuestas, las transmisiones informativas, las consultas y las acciones de control como la gestión de la suscripción, la señalización de errores y los reconocimientos. Al centralizar estos tipos de mensajes, el protocolo garantiza un manejo y enrutamiento consistentes de las comunicaciones entre agentes en todo el sistema.
class ACPPerformative(Enum):
"""ACP speech acts (performatives)"""
TELL = "tell"
ASK = "ask"
REPLY = "reply"
REQUEST_ACTION = "request-action"
AGREE = "agree"
REFUSE = "refuse"
PROPOSE = "propose"
ACCEPT = "accept"
REJECT = "reject"La enumeración de AcperFormative captura la variedad de actos de habla que los agentes pueden usar al interactuar bajo el marco ACP, mapear intenciones de alto nivel, como hacer solicitudes, plantear preguntas, dar comandos o negociar acuerdos, en etiquetas estandarizadas. Esta taxonomía clara permite a los agentes interpretar y responder a los mensajes de maneras contextualmente apropiadas, asegurando una comunicación robusta y semánticamente rica.
@dataclass
class ACPMessage:
"""Agent Communication Protocol Message Structure"""
message_id: str
sender: str
receiver: str
performative: str
content: Dict[str, Any]
protocol: str = "ACP-1.0"
conversation_id: str = None
reply_to: str = None
language: str = "english"
encoding: str = "json"
timestamp: float = None
def __post_init__(self):
if self.timestamp is None:
self.timestamp = time.time()
if self.conversation_id is None:
self.conversation_id = str(uuid.uuid4())
def to_acp_format(self) -> str:
"""Convert to standard ACP message format"""
acp_msg = {
"message-id": self.message_id,
"sender": self.sender,
"receiver": self.receiver,
"performative": self.performative,
"content": self.content,
"protocol": self.protocol,
"conversation-id": self.conversation_id,
"reply-to": self.reply_to,
"language": self.language,
"encoding": self.encoding,
"timestamp": self.timestamp
}
return json.dumps(acp_msg, indent=2)
@classmethod
def from_acp_format(cls, acp_string: str) -> 'ACPMessage':
"""Parse ACP message from string format"""
data = json.loads(acp_string)
return cls(
message_id=data["message-id"],
sender=data["sender"],
receiver=data["receiver"],
performative=data["performative"],
content=data["content"],
protocol=data.get("protocol", "ACP-1.0"),
conversation_id=data.get("conversation-id"),
reply_to=data.get("reply-to"),
language=data.get("language", "english"),
encoding=data.get("encoding", "json"),
timestamp=data.get("timestamp", time.time())
)
La clase de datos de ACPMessage encapsula todos los campos necesarios para un intercambio de ACP estructurado, incluidos identificadores, participantes, performativos, carga útil y metadatos, como la versión de protocolo, el lenguaje y las marcas de tiempo. Su método __post_init__ se produce automáticamente los valores de timestamp y conversación_id, asegurando que cada mensaje se rastree de manera única. Métodos de utilidad TO_ACP_FORMAT y From_ACP_Format Handle Serialization hacia y desde la representación JSON estandarizada para transmisión y análisis sin problemas.
class ACPAgent:
"""Agent implementing Agent Communication Protocol"""
def __init__(self, agent_id: str, name: str, capabilities: List[str]):
self.agent_id = agent_id
self.name = name
self.capabilities = capabilities
self.model = genai.GenerativeModel("gemini-1.5-flash")
self.message_queue: List[ACPMessage] = []
self.subscriptions: Dict[str, List[str]] = {}
self.conversations: Dict[str, List[ACPMessage]] = {}
def create_message(self, receiver: str, performative: str,
content: Dict[str, Any], conversation_id: str = None,
reply_to: str = None) -> ACPMessage:
"""Create a new ACP-compliant message"""
return ACPMessage(
message_id=str(uuid.uuid4()),
sender=self.agent_id,
receiver=receiver,
performative=performative,
content=content,
conversation_id=conversation_id,
reply_to=reply_to
)
def send_inform(self, receiver: str, fact: str, data: Any = None) -> ACPMessage:
"""Send an INFORM message (telling someone a fact)"""
content = {"fact": fact, "data": data}
return self.create_message(receiver, ACPPerformative.TELL.value, content)
def send_query(self, receiver: str, question: str, query_type: str = "yes-no") -> ACPMessage:
"""Send a QUERY message (asking for information)"""
content = {"question": question, "query-type": query_type}
return self.create_message(receiver, ACPPerformative.ASK.value, content)
def send_request(self, receiver: str, action: str, parameters: Dict = None) -> ACPMessage:
"""Send a REQUEST message (asking someone to perform an action)"""
content = {"action": action, "parameters": parameters or {}}
return self.create_message(receiver, ACPPerformative.REQUEST_ACTION.value, content)
def send_reply(self, original_msg: ACPMessage, response_data: Any) -> ACPMessage:
"""Send a REPLY message in response to another message"""
content = {"response": response_data, "original-question": original_msg.content}
return self.create_message(
original_msg.sender,
ACPPerformative.REPLY.value,
content,
conversation_id=original_msg.conversation_id,
reply_to=original_msg.message_id
)
def process_message(self, message: ACPMessage) -> Optional[ACPMessage]:
"""Process incoming ACP message and generate appropriate response"""
self.message_queue.append(message)
conv_id = message.conversation_id
if conv_id not in self.conversations:
self.conversations[conv_id] = []
self.conversations[conv_id].append(message)
if message.performative == ACPPerformative.ASK.value:
return self._handle_query(message)
elif message.performative == ACPPerformative.REQUEST_ACTION.value:
return self._handle_request(message)
elif message.performative == ACPPerformative.TELL.value:
return self._handle_inform(message)
return None
def _handle_query(self, message: ACPMessage) -> ACPMessage:
"""Handle incoming query messages"""
question = message.content.get("question", "")
prompt = f"As agent {self.name} with capabilities {self.capabilities}, answer: {question}"
try:
response = self.model.generate_content(prompt)
answer = response.text.strip()
except:
answer = "Unable to process query at this time"
return self.send_reply(message, {"answer": answer, "confidence": 0.8})
def _handle_request(self, message: ACPMessage) -> ACPMessage:
"""Handle incoming action requests"""
action = message.content.get("action", "")
parameters = message.content.get("parameters", {})
if any(capability in action.lower() for capability in self.capabilities):
result = f"Executing {action} with parameters {parameters}"
status = "agreed"
else:
result = f"Cannot perform {action} - not in my capabilities"
status = "refused"
return self.send_reply(message, {"status": status, "result": result})
def _handle_inform(self, message: ACPMessage) -> Optional[ACPMessage]:
"""Handle incoming information messages"""
fact = message.content.get("fact", "")
print(f"[{self.name}] Received information: {fact}")
ack_content = {"status": "received", "fact": fact}
return self.create_message(message.sender, "acknowledge", ack_content,
conversation_id=message.conversation_id)
La clase ACPAGENT encapsula una entidad autónoma capaz de enviar, recibir y procesar mensajes compatibles con ACP utilizando el modelo de lenguaje de Gemini. Gestiona su propia cola de mensajes, historial de conversación y suscripciones, y proporciona métodos de ayuda (send_inform, send_query, send_request, send_reply) para construir instancias de acpMessage con formato correcto. Los mensajes entrantes se enrutan a través de Process_Message, que delega a manejadores especializados para consultas, solicitudes de acción y mensajes informativos.
class ACPMessageBroker:
"""Message broker implementing ACP routing and delivery"""
def __init__(self):
self.agents: Dict[str, ACPAgent] = {}
self.message_log: List[ACPMessage] = []
self.routing_table: Dict[str, str] = {}
def register_agent(self, agent: ACPAgent):
"""Register an agent with the message broker"""
self.agents[agent.agent_id] = agent
self.routing_table[agent.agent_id] = "local"
print(f"✓ Registered agent: {agent.name} ({agent.agent_id})")
def route_message(self, message: ACPMessage) -> bool:
"""Route ACP message to appropriate recipient"""
if message.receiver not in self.agents:
print(f"✗ Receiver {message.receiver} not found")
return False
print(f"\n📨 ACP MESSAGE ROUTING:")
print(f"From: {message.sender} → To: {message.receiver}")
print(f"Performative: {message.performative}")
print(f"Content: {json.dumps(message.content, indent=2)}")
receiver_agent = self.agents[message.receiver]
response = receiver_agent.process_message(message)
self.message_log.append(message)
if response:
print(f"\n📤 GENERATED RESPONSE:")
print(f"From: {response.sender} → To: {response.receiver}")
print(f"Content: {json.dumps(response.content, indent=2)}")
if response.receiver in self.agents:
self.agents[response.receiver].process_message(response)
self.message_log.append(response)
return True
def broadcast_message(self, message: ACPMessage, recipients: List[str]):
"""Broadcast message to multiple recipients"""
for recipient in recipients:
msg_copy = ACPMessage(
message_id=str(uuid.uuid4()),
sender=message.sender,
receiver=recipient,
performative=message.performative,
content=message.content.copy(),
conversation_id=message.conversation_id
)
self.route_message(msg_copy)El ACPMessageBroker sirve como el enrutador central para los mensajes ACP, manteniendo un registro de agentes y un registro de mensajes. Proporciona métodos para registrar agentes, entregar mensajes individuales a través de Route_Message, que maneja la búsqueda, registro y encadenamiento de respuesta, y para enviar el mismo mensaje a múltiples destinatarios con Broadcast_Message.
def demonstrate_acp():
"""Comprehensive demonstration of Agent Communication Protocol"""
print("🤖 AGENT COMMUNICATION PROTOCOL (ACP) DEMONSTRATION")
print("=" * 60)
broker = ACPMessageBroker()
researcher = ACPAgent("agent-001", "Dr. Research", ["analysis", "research", "data-processing"])
assistant = ACPAgent("agent-002", "AI Assistant", ["information", "scheduling", "communication"])
calculator = ACPAgent("agent-003", "MathBot", ["calculation", "mathematics", "computation"])
broker.register_agent(researcher)
broker.register_agent(assistant)
broker.register_agent(calculator)
print(f"\n📋 REGISTERED AGENTS:")
for agent_id, agent in broker.agents.items():
print(f" • {agent.name} ({agent_id}): {', '.join(agent.capabilities)}")
print(f"\n🔬 SCENARIO 1: Information Query (ASK performative)")
query_msg = assistant.send_query("agent-001", "What are the key factors in AI research?")
broker.route_message(query_msg)
print(f"\n🔢 SCENARIO 2: Action Request (REQUEST-ACTION performative)")
calc_request = researcher.send_request("agent-003", "calculate", {"expression": "sqrt(144) + 10"})
broker.route_message(calc_request)
print(f"\n📢 SCENARIO 3: Information Sharing (TELL performative)")
info_msg = researcher.send_inform("agent-002", "New research paper published on quantum computing")
broker.route_message(info_msg)
print(f"\n📊 PROTOCOL STATISTICS:")
print(f" • Total messages processed: {len(broker.message_log)}")
print(f" • Active conversations: {len(set(msg.conversation_id for msg in broker.message_log))}")
print(f" • Message types used: {len(set(msg.performative for msg in broker.message_log))}")
print(f"\n📋 SAMPLE ACP MESSAGE FORMAT:")
sample_msg = assistant.send_query("agent-001", "Sample question for format demonstration")
print(sample_msg.to_acp_format())La función demostración_acp orquesta un tutorial práctico de todo el marco ACP: inicializa un corredor y tres agentes distintos (investigador, asistente de IA y Mathbot), los registra e ilustra tres escenarios de interacción clave, consultando información, solicitando un cálculo y compartir una actualización. Después de enrutar cada mensaje y manejo de respuestas, imprime estadísticas resumidas en el flujo de mensajes. Muestra un mensaje ACP formateado, que proporciona a los usuarios un ejemplo claro y de extremo a extremo de cómo los agentes se comunican bajo el protocolo.
def setup_guide():
print("""
🚀 GOOGLE COLAB SETUP GUIDE:
1. Get Gemini API Key: https://makersuite.google.com/app/apikey
2. Replace: GEMINI_API_KEY = "YOUR_ACTUAL_API_KEY"
3. Run: demonstrate_acp()
🔧 ACP PROTOCOL FEATURES:
• Standardized message format with required fields
• Speech act performatives (TELL, ASK, REQUEST-ACTION, etc.)
• Conversation tracking and message threading
• Error handling and acknowledgments
• Message routing and delivery confirmation
📝 EXTEND THE PROTOCOL:
```python
# Create custom agent
my_agent = ACPAgent("my-001", "CustomBot", ["custom-capability"])
broker.register_agent(my_agent)
# Send custom message
msg = my_agent.send_query("agent-001", "Your question here")
broker.route_message(msg)
```
""")
if __name__ == "__main__":
setup_guide()
demonstrate_acp()
Finalmente, la función SETUP_GUIDE proporciona una referencia de arranque rápido para ejecutar la demostración de ACP en Google Colab, describiendo cómo obtener y configurar su clave API Gemini e invocar la rutina demostrate_ACP. También resume las características clave del protocolo, como formatos de mensajes estandarizados, performadores y enrutamiento de mensajes. Proporciona un fragmento de código conciso que ilustra cómo registrar agentes personalizados y enviar mensajes a medida.
En conclusión, este tutorial implementa sistemas de agentes múltiples basados en ACP capaces de investigaciones, computación y tareas de colaboración. Los escenarios de muestra proporcionados ilustran casos de uso comunes, consultas de información, solicitudes computacionales y intercambio de hechos, mientras que el corredor garantiza la entrega y registro de mensajes confiables. Se alienta a los lectores a extender el marco agregando nuevas capacidades de agente, integrando acciones específicas de dominio o incorporando mecanismos de suscripción y notificación más sofisticados.
Descargar el Cuaderno en Github. 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 95k+ 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.
