La communication d'agents IA via MQTT apporte aux systèmes multi-agents les propriétés qui ont fait de MQTT la référence de l'IoT : émetteurs et récepteurs découplés, détection de présence, messages légers et livraison fiable sur des réseaux imparfaits. Cet article explique pourquoi le publish/subscribe convient bien aux agents, quelles conventions de topics rendent possibles la découverte et la messagerie, et comment les fonctionnalités de MQTT 5 telles que les response topics, les correlation data et l'expiration des messages s'appliquent aux interactions entre agents.
Pourquoi les agents ont besoin d'une couche de messagerie
Dès que vous avez plus d'un agent, par exemple un agent de surveillance qui observe la télémétrie, un agent de planification qui propose des actions et un agent de reporting qui rédige des synthèses, ils ont besoin d'un moyen de communiquer. La première tentative consiste souvent en des appels HTTP directs entre agents. Cela fonctionne avec deux agents et devient pénible à cinq :
- Chaque agent doit connaître l'adresse de tous les autres.
- Si un agent est occupé ou redémarre, les appels échouent et chaque appelant doit gérer ses propres nouvelles tentatives.
- Diffuser un événement (« la ligne 2 est arrêtée ») implique d'appeler individuellement chaque agent intéressé.
- Personne ne dispose d'un point unique pour observer ce que les agents se disent.
Un broker règle ces problèmes en plaçant des topics entre les agents. Les émetteurs publient sur un topic, les récepteurs s'y abonnent, et le broker MQTT se charge du routage, de la diffusion et de la mise en mémoire tampon pour les sessions persistantes. Les agents deviennent des services faiblement couplés que l'on peut ajouter, redémarrer ou dupliquer sans reconfigurer leurs pairs.
Pourquoi MQTT plutôt qu'un autre protocole ?
Plusieurs systèmes de messagerie pourraient transporter le trafic des agents. MQTT présente des avantages précis lorsque les agents cohabitent avec des appareils :
| Besoin | Fonctionnalité MQTT |
|---|---|
| Agents et appareils sur la même infrastructure | Les agents lisent la télémétrie et envoient des commandes via le broker que les appareils utilisent déjà |
| Savoir quels agents sont actifs | Last will and testament et messages de statut en retained message |
| Fournir le contexte actuel aux nouveaux venus | Retained messages sur les topics de capacités et d'état |
| Requête/réponse entre agents | Response topic et correlation data de MQTT 5 |
| Écarter les requêtes périmées | Intervalle d'expiration des messages de MQTT 5 |
| Répartir le travail entre réplicas d'un agent | Abonnements partagés de MQTT 5 |
| Fonctionner sur des hôtes contraints ou distants | Faible surcoût protocolaire, une seule connexion sortante, fonctionne via TLS et WebSockets |
La dernière ligne compte pour les déploiements en périphérie : un agent sur une passerelle d'usine n'a besoin que d'une connexion sortante sur le port 8883, sans aucune règle de pare-feu entrante.
Conventions de topics pour la communication d'agents IA
Un espace de noms cohérent est ce qui transforme un broker en réseau d'agents. Voici une convention simple et efficace :
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
Découverte grâce aux cartes d'agent en retained message
Chaque agent publie au démarrage une « carte » en retained message qui le décrit :
{
"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."
}
Tout nouvel agent s'abonne à agents/+/card et reçoit immédiatement toutes les cartes, car le broker livre les retained messages au moment de l'abonnement. Vous obtenez ainsi un registre de services complet sans infrastructure supplémentaire. Pour retirer un agent du registre, publiez un retained message vide sur le topic de sa carte.
Présence grâce au last will
Chaque agent se connecte avec un last will offline sur son topic de statut et publie un online en retained message une fois connecté. Si l'agent plante ou perd le réseau, le broker publie le will à sa place. Les autres agents abonnés à agents/+/status savent, dans le délai du keepalive, qu'un pair a disparu et peuvent cesser de lui confier du travail.
Messages directs et requête/réponse
MQTT 5 ajoute deux propriétés qui rendent la requête/réponse propre. Le demandeur définit un response topic (où envoyer la réponse) et des correlation data (un identifiant opaque pour associer la réponse à la question). Ajouter un intervalle d'expiration du message garantit qu'une requête que personne ne traite à temps est écartée au lieu d'être traitée bien après qu'elle a perdu son intérêt.
L'exemple ci-dessous présente un agent qui répond et un agent demandeur utilisant paho-mqtt 2.x avec MQTT 5, connectés au serveur MQTT public gratuit broker.connectmq.com, réservé aux tests et aux prototypes. Remplacez 9e1f3b par votre propre chaîne aléatoire.
# responder.py - a "summarizer" agent that answers requests in its inbox
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", "")
# A real agent would call an LLM here
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 - another agent asking the summarizer for help
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 # seconds; drop the request if not handled in time
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")
Lancez l'agent qui répond, puis le demandeur, et observez toute la conversation avec :
mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i spy-9e1f3b -t "connectmq/demo/9e1f3b/#" -v
Pools de workers avec abonnements partagés
Lorsqu'un type de tâche devient chargé, exécutez plusieurs réplicas du même agent et faites-leur partager un abonnement comme $share/planners/tasks/plan. Le broker livre chaque message à un seul membre du groupe, ce qui vous offre de la répartition de charge sans système de files d'attente séparé. Sur ConnectMQ, les abonnements partagés sont disponibles à partir de l'offre Serverless (Paid).
Conception des messages entre agents
Les agents échangent du langage naturel, mais l'enveloppe doit être structurée. Un bon message contient :
- Un type (
plan.request,plan.proposal) pour que les récepteurs puissent router sans analyser de prose. - Un identifiant et une corrélation pour relier réponses et suites.
- L'émetteur et un horodatage pour l'audit et l'ordonnancement.
- Un payload où le langage naturel figure dans un champ clairement identifié, séparé des données structurées.
- Une expiration ou une échéance pour qu'aucune instruction périmée ne soit exécutée.
Gardez des messages légers. Transmettez des références, comme un nom de topic ou un identifiant de document, au lieu d'intégrer de gros documents. Le serveur MQTT public gratuit limite les messages à 64 Ko, et des messages plus petits sont de toute façon plus rapides pour chaque abonné.
Sécurité de la communication entre agents
Les systèmes multi-agents créent un nouveau risque : la sortie d'un agent devient l'entrée d'un autre. Si un agent résume une note de maintenance contenant des instructions injectées, un agent en aval peut les traiter comme des ordres. Défenses pratiques :
- Traitez chaque message entrant comme une donnée, jamais comme une instruction qui remplacerait le prompt système de l'agent.
- Attribuez à chaque agent ses propres identifiants de broker et son propre client ID, et réservez les identifiants autorisés à publier des commandes d'appareils à un seul dispatcher chargé de la validation. Notre guide sur les agents LLM pour appareils IoT détaille cette passerelle de règles.
- Journalisez tout le trafic des agents en abonnant un client d'audit à leur espace de noms.
- Limitez le débit des agents pour qu'une boucle entre deux d'entre eux ne puisse pas inonder le broker. Les brokers dotés d'une protection contre le flood et de quotas par offre, comme un broker MQTT managé, fournissent un filet de sécurité, mais la correction doit se faire dans les agents.
Anti-patterns courants à éviter
Certaines conceptions semblent pratiques au départ et posent problème ensuite :
- Un topic géant pour tout. Publier tout le trafic des agents sur un seul topic oblige chaque agent à analyser chaque message. Utilisez plutôt des boîtes de réception par agent et des topics d'événements typés.
- Des boucles d'agents trop bavardes. Deux agents qui répondent aux diffusions l'un de l'autre peuvent créer une conversation sans fin. Donnez un identifiant à chaque message, ignorez vos propres messages et limitez la profondeur des réponses.
- Des payloads volumineux. Intégrer des documents entiers ou des images dans les messages ralentit chaque abonné. Publiez des références et laissez les agents récupérer le contenu au besoin.
- Retenir ce qu'il ne faut pas. Conservez en retained message les statuts et les cartes de capacités, pas les requêtes ni les résultats, sinon les nouveaux abonnés reçoivent d'anciennes tâches.
Lien avec les protocoles d'agents
Les protocoles spécifiques aux agents définissent comment un assistant appelle des outils ou comment les agents se décrivent, tandis que MQTT est un transport qui achemine des messages entre de nombreux participants. Ils sont complémentaires : un serveur MCP pour MQTT permet à un assistant IA d'atteindre, via des outils, les appareils et agents présents sur le broker, et les conventions de topics ci-dessus permettent aux agents autonomes de se coordonner entre eux. Pour une vue d'ensemble de l'IA dans les systèmes connectés, consultez notre guide IA et IoT, et pour en savoir plus sur les fonctionnalités du protocole utilisées ici, lisez notre article sur les fonctionnalités de MQTT 5.
Démarrer avec ConnectMQ
Offrez à vos agents une infrastructure commune avec ConnectMQ, un broker MQTT cloud qui prend en charge les response topics de MQTT 5, l'expiration des messages, les retained messages et l'expiration de session. Créez un broker Serverless (Free), sans carte bancaire, et connectez vos premiers agents en quelques minutes.
Questions fréquentes
MQTT est-il assez rapide pour des agents IA ?
Oui. La livraison des messages par un broker est bien plus rapide que l'inférence d'un LLM : le transport est donc rarement le goulot d'étranglement. L'essentiel du travail de conception porte sur la structure des topics et le format des messages.
Comment les agents se trouvent-ils via MQTT ?
Chaque agent publie une carte de capacités en retained message sur un modèle de topic connu. Les nouveaux agents s'abonnent avec un wildcard et reçoivent immédiatement toutes les cartes.
Peut-on faire de la requête/réponse avec MQTT ?
Oui. Les response topics et les correlation data de MQTT 5 offrent un pattern de requête/réponse standard, et l'expiration des messages écarte les requêtes non traitées à temps.
Que se passe-t-il si un agent plante ?
Le broker publie le last will de l'agent, généralement un statut hors ligne, pour que les autres agents cessent de lui confier du travail jusqu'à sa reconnexion.
