Patrones de comunicación entre agentes de IA con MQTT: pub/sub para sistemas multiagente

La comunicación entre agentes de IA sobre MQTT aporta a los sistemas multiagente las mismas propiedades que convirtieron a MQTT en el estándar de facto del IoT: emisores y receptores desacoplados, detección de presencia, mensajes pequeños y entrega fiable sobre redes imperfectas. Este artículo explica por qué la publicación/suscripción encaja bien con los agentes, qué convenciones de topics hacen funcionar el descubrimiento y la mensajería, y cómo las funciones de MQTT 5, como los response topics, los correlation data y la caducidad de mensajes, se corresponden con las interacciones entre agentes.

Por qué los agentes necesitan una capa de mensajería

En cuanto tienes más de un agente, por ejemplo un agente de monitorización que vigila la telemetría, un agente planificador que propone acciones y un agente de informes que redacta resúmenes, necesitan una forma de hablar entre ellos. El primer intento habitual son llamadas HTTP directas entre agentes. Funciona con dos agentes y se vuelve penoso con cinco:

• Cada agente debe conocer la dirección de todos los demás.
• Si un agente está ocupado o reiniciándose, las llamadas fallan y cada llamante tiene que implementar reintentos.
• Difundir un evento ("la línea 2 se ha parado") obliga a llamar uno por uno a cada agente interesado.
• No existe un único lugar desde el que observar qué se dicen los agentes entre sí.

Un broker resuelve todo esto situando topics entre los agentes. Los emisores publican en un topic, los receptores se suscriben y el broker MQTT se encarga del enrutamiento, la distribución y el almacenamiento para las sesiones persistentes. Los agentes pasan a ser servicios poco acoplados que pueden añadirse, reiniciarse o escalarse sin reconfigurar a sus pares.

¿Por qué MQTT y no otro protocolo?

Varios sistemas de mensajería podrían transportar el tráfico entre agentes. MQTT tiene ventajas concretas cuando los agentes conviven con dispositivos:

La última fila importa en despliegues edge: un agente en un gateway de fábrica solo necesita una conexión saliente por el puerto 8883, sin reglas de firewall entrantes.

Convenciones de topics para la comunicación entre agentes de IA

Un espacio de nombres coherente es lo que convierte un broker en una red de agentes. Una convención sencilla y práctica:

agents/<agent-id>/status        retained: online/offline (set as last will)
agents/<agent-id>/card          retained: capabilities, version, accepted message types
agents/<agent-id>/inbox         direct messages and requests to this agent
agents/<agent-id>/outbox/<n>    responses and results produced by this agent
agents/broadcast/<event>        events every agent may react to
tasks/<capability>              work queues, consumed by shared subscriptions

Descubrimiento con tarjetas de agente retenidas

Cada agente publica al arrancar una "tarjeta" retenida que lo describe:

{
  "id": "planner-1",
  "role": "maintenance planner",
  "version": "0.4.2",
  "accepts": ["plan.request"],
  "emits": ["plan.proposal"],
  "description": "Proposes maintenance actions from alarms and work order history."
}

Cualquier agente nuevo se suscribe a agents/+/card y recibe de inmediato todas las tarjetas, porque el broker entrega los mensajes retenidos en el momento de la suscripción. Es un registro de servicios completo sin infraestructura adicional. Para eliminar un agente del registro, publica un mensaje retenido vacío en el topic de su tarjeta.

Presencia con last will

Cada agente se conecta con un last will offline en su topic de estado y publica un online retenido una vez conectado. Si el agente se cae o pierde la red, el broker publica el will en su nombre. Los demás agentes suscritos a agents/+/status saben, dentro del periodo de keepalive, que un par ha desaparecido y pueden dejar de enviarle trabajo.

Mensajes directos y petición/respuesta

MQTT 5 añade dos propiedades que hacen limpio el patrón de petición/respuesta. Quien pide fija un response topic (adónde enviar la respuesta) y unos correlation data (un ID opaco para emparejar la respuesta con la pregunta). Añadir un intervalo de caducidad del mensaje garantiza que una petición que nadie atiende a tiempo se descarte en lugar de procesarse cuando ya no importa.

El ejemplo siguiente muestra un agente que responde y otro que pide, usando paho-mqtt 2.x con MQTT 5 contra el servidor MQTT público gratuito en broker.connectmq.com, pensado para pruebas y no para datos de producción. Sustituye 9e1f3b por tu propia cadena aleatoria.

# responder.py - un agente "summarizer" que responde a las peticiones de su bandeja de entrada
import json
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BASE = "connectmq/demo/9e1f3b/agents"
ME = "summarizer-1"


def on_connect(client, userdata, flags, reason_code, properties):
    client.publish(f"{BASE}/{ME}/status", "online", qos=1, retain=True)
    client.publish(f"{BASE}/{ME}/card", json.dumps({"id": ME, "accepts": ["summarize.request"]}), qos=1, retain=True)
    client.subscribe(f"{BASE}/{ME}/inbox", qos=1)


def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    if not response_topic:
        return
    request = json.loads(msg.payload)
    text = request.get("text", "")
    # Un agente real llamaría aquí a un LLM
    result = {"summary": text[:80] + ("..." if len(text) > 80 else ""), "agent": ME}
    reply_props = Properties(PacketTypes.PUBLISH)
    if hasattr(props, "CorrelationData"):
        reply_props.CorrelationData = props.CorrelationData
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{ME}-9e1f3b", protocol=mqtt.MQTTv5)
client.will_set(f"{BASE}/{ME}/status", "offline", qos=1, retain=True)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

# requester.py - otro agente que pide ayuda al summarizer
import json
import threading
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BASE = "connectmq/demo/9e1f3b/agents"
ME = "monitor-1"
REPLY_TOPIC = f"{BASE}/{ME}/outbox/replies"
pending = {}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REPLY_TOPIC, qos=1)
    userdata.set()


def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"").decode()
    if corr in pending:
        print("reply for", pending.pop(corr), "->", msg.payload.decode())


ready = threading.Event()
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{ME}-9e1f3b",
                     protocol=mqtt.MQTTv5, userdata=ready)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
ready.wait(10)

corr_id = str(uuid.uuid4())
pending[corr_id] = "alarm digest"
props = Properties(PacketTypes.PUBLISH)
props.ResponseTopic = REPLY_TOPIC
props.CorrelationData = corr_id.encode()
props.MessageExpiryInterval = 30   # segundos; descarta la petición si no se atiende a tiempo
request = {"type": "summarize.request", "text": "Line 2 conveyor stopped three times this morning due to jam sensor trips at station 4."}
client.publish(f"{BASE}/summarizer-1/inbox", json.dumps(request), qos=1, properties=props)

input("waiting for replies, press Enter to quit\n")

Ejecuta el agente que responde, después el que pide, y observa toda la conversación con:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i spy-9e1f3b -t "connectmq/demo/9e1f3b/#" -v

Grupos de trabajadores con suscripciones compartidas

Cuando un tipo de tarea tiene mucha carga, ejecuta varias réplicas del mismo agente y haz que compartan una suscripción como $share/planners/tasks/plan. El broker entrega cada mensaje a un solo miembro del grupo, lo que te da balanceo de carga sin un sistema de colas aparte. En ConnectMQ, las suscripciones compartidas están disponibles en el plan Serverless (Paid) y superiores.

Diseño de mensajes para agentes

Los agentes intercambian lenguaje natural, pero el sobre debe estar estructurado. Un buen mensaje incluye:

1. Tipo (plan.request, plan.proposal) para que los receptores puedan enrutar sin analizar prosa.
2. ID y correlación para enlazar respuestas y seguimientos.
3. Emisor y marca de tiempo para la auditoría y el orden.
4. Payload con el lenguaje natural en un campo claramente identificado, separado de los datos estructurados.
5. Caducidad o plazo para que no se actúe sobre instrucciones obsoletas.

Mantén los mensajes pequeños. Pasa referencias, como un nombre de topic o un ID de documento, en lugar de incrustar documentos grandes. El servidor público limita los mensajes a 64 KB y, en cualquier caso, los mensajes más pequeños son más rápidos para cada suscriptor.

Seguridad en la comunicación entre agentes

Los sistemas multiagente crean un riesgo nuevo: la salida de un agente se convierte en la entrada de otro. Si un agente resume una nota de mantenimiento que contiene instrucciones inyectadas, un agente posterior puede tomarlas como órdenes. Defensas prácticas:

• Trata cada mensaje entrante como datos, nunca como instrucciones que anulen el prompt de sistema del propio agente.
• Da a cada agente sus propias credenciales de broker y su propio client ID, y deja las credenciales que pueden publicar comandos a dispositivos en manos de un único despachador que valide. Nuestra guía sobre agentes LLM para dispositivos IoT muestra en detalle esta puerta de políticas.
• Registra todo el tráfico entre agentes suscribiendo un cliente de auditoría al espacio de nombres de los agentes.
• Limita la frecuencia de los agentes para que un bucle entre dos de ellos no inunde el broker. Los brokers con protección contra inundaciones y cuotas por plan, como un broker MQTT gestionado, ofrecen una red de seguridad, pero la solución corresponde a los agentes.

Antipatrones comunes que conviene evitar

Algunos diseños parecen cómodos al principio y causan problemas después:

• Un único topic gigante para todo. Publicar todo el tráfico de agentes en un solo topic obliga a cada agente a analizar cada mensaje. Usa bandejas de entrada por agente y topics de eventos tipados.
• Bucles de agentes parlanchines. Dos agentes que responden a las difusiones del otro pueden crear una conversación interminable. Da un ID a cada mensaje, ignora tus propios mensajes y limita la profundidad de las respuestas.
• Payloads grandes. Incrustar documentos o imágenes completos en los mensajes de los agentes ralentiza a cada suscriptor. Publica referencias y deja que los agentes obtengan el contenido cuando lo necesiten.
• Retener lo que no toca. Retén el estado y las tarjetas de capacidades, no las peticiones ni los resultados; de lo contrario, los nuevos suscriptores recibirán trabajos antiguos.

Relación con los protocolos de agentes

Los protocolos específicos para agentes definen cómo un asistente llama a herramientas o cómo se describen los agentes, mientras que MQTT es un transporte que mueve mensajes entre muchas partes. Se complementan: un servidor MCP para MQTT permite a un asistente de IA llegar mediante herramientas a los dispositivos y agentes del broker, y las convenciones de topics anteriores permiten que los agentes autónomos se coordinen entre sí. Para una visión más amplia de la IA en sistemas conectados, consulta nuestra guía de IA e IoT, y para saber más sobre las funciones del protocolo usadas aquí, lee sobre las características de MQTT 5.

Empieza con ConnectMQ

Dale a tus agentes una columna vertebral común con ConnectMQ, un broker MQTT en la nube compatible con response topics de MQTT 5, caducidad de mensajes, retained messages y caducidad de sesión. Crea un broker Serverless gratuito, sin tarjeta de crédito, y conecta tus primeros agentes en minutos.

Preguntas frecuentes

¿Es MQTT lo bastante rápido para agentes de IA?

Sí. La entrega de mensajes a través de un broker es mucho más rápida que la inferencia de un LLM, así que el transporte rara vez es el cuello de botella. El trabajo de diseño principal está en la estructura de topics y en los formatos de mensaje.

¿Cómo se encuentran los agentes entre sí en MQTT?

Cada agente publica una tarjeta de capacidades retenida en un patrón de topic conocido. Los agentes nuevos se suscriben con un comodín y reciben de inmediato todas las tarjetas.

¿Puedo hacer petición/respuesta con MQTT?

Sí. Los response topics y los correlation data de MQTT 5 proporcionan un patrón estándar de petición/respuesta, y la caducidad de mensajes descarta las peticiones que no se atienden a tiempo.

¿Qué ocurre si un agente se cae?

El broker publica el last will del agente, normalmente un estado offline, para que los demás agentes dejen de enviarle trabajo hasta que vuelva a conectarse.