Le bus de communication multi-agents — le contrat, mesure

Quand plusieurs agents debattent, ils ont besoin de se parler : envoyer une commande, diffuser un resultat, demander une assistance, repondre a une requete. Ce carnet distille le bus de communication du moteur de debat EPITA en un support pret a l’emploi : un format de message commun, un contrat de canal, quatre familles de canaux, un protocole requete-reponse, et le middleware qui route le tout.

Provenance (corrigee par audit #18434, mesure du 2026-09-30). L’organe du tronc EPITA est reparti sur trois modules dans argumentation_analysis/core/communication/ du depot etudiant (cf. fallacy_rules.py pour la convention de reference) :

channel_interface.py (L17 ChannelType, L109 matches_filter, L155 LocalChannel) ;
message.py (L110 Message.__lt__, L191 create_response) ;
middleware.py (L163 determine_channel, methode du middleware, pas fonction libre).

Outre ces trois modules, le repertoire porte les canaux concrets, le protocole requete-reponse, des adaptateurs et un dossier tests/ (liste complete : ls argumentation_analysis/core/communication/). Les quatre canaux concrets (hierarchique, collaboration, donnees, pub/sub) sont des implementations du meme contrat ; le protocole requete-reponse complet existe (RequestTimeoutError L21, retry_count L80).

Pour la portabilite du carnet (pas d’acces au depot EPITA), les trois modules du contrat sont agreges dans un port vendored (le fichier communication_channels.py present dans le repertoire du carnet). C’est une extraction de donnees au sens de la convention fallacy_rules.py (decision coordinateur 2026-09-26, epic #4960), pas un import de code depuis le depot etudiant : le carnet importe son port local, jamais le depot EPITA.

La distillation est deterministe : aucun reseau, aucun thread, aucun LLM — tout ce que ce carnet affiche est rejouable a l’identique.

Le sujet est le contrat, pas le transport. Un bus de communication ne se juge pas a sa vitesse mais a ce qu’il garantit : qui recoit quoi, dans quel ordre, avec quelles priorites, et que se passe-t-il quand un filtre ne correspond a rien. Ce carnet mesure ces garanties.

Le port vendored est determine par construction (re-declaration des trois modules du tronc). La source de verite du tronc est :

test_channel_interface.py (10 cas) ;
test_message.py (15 cas) ;
test_communication_integration.py (4 cas).

Soit 29 tests au total — pas 12 bancs / 45 cas comme la version precedente l’annoncait. Ce carnet rejoue les cas deterministes et documente les autres en docstring du port vendored.

Trois ChannelType n’ont pas d’implementation Channel concrete ; leurs routes ont ete retirees du routage (#1571) plutot que cablees a des classes vides — EVENT, CONTROL, SUBSCRIPTION ne routent plus vers FEEDBACK / SYSTEM, ils tombent sur le bus par defaut. Le carnet documente ce comportement dans les cellules de routage.

# Organe de la serie + les enums qui definissent le vocabulaire du bus.
from communication_channels import (
    MessageType, MessagePriority, AgentLevel, ChannelType,
    Message, LocalChannel, create_response, determine_channel,
)
from datetime import datetime, timedelta

T0 = datetime(2026, 1, 1, 12, 0, 0)

print("Types de message :", [t.value for t in MessageType])
print("Priorites        :", [p.value for p in MessagePriority])
print("Niveaux agent    :", [a.value for a in AgentLevel])
print("Types de canal   :", [c.value for c in ChannelType])
print()
print("Canaux SANS implementation :", ["negotiation", "feedback", "system"])

Types de message : ['command', 'information', 'request', 'response', 'event', 'control', 'publication', 'subscription']
Priorites        : ['low', 'normal', 'high', 'critical']
Niveaux agent    : ['strategic', 'tactical', 'operational', 'system']
Types de canal   : ['hierarchical', 'collaboration', 'data', 'negotiation', 'feedback', 'system', 'local']

Canaux SANS implementation : ['negotiation', 'feedback', 'system']

Lecture du resultat. Huit types de message, quatre priorites, quatre niveaux d’agent, sept types de canal — mais trois types de canal n’ont aucune implementation (negotiation, feedback, system). Le tronc les declare dans l’enumere mais les a retires du routage (#1571) plutot que de les cabler a des classes vides : un canal promis sans implementation est un mensonge du bus — les messages y arriveraient sans jamais etre delivres. C’est une lecon d’architecture : enumerer un type ne cree pas un transport.

1. Le format message — priorite inversee, identite derivable

Un Message porte type / emetteur / niveau / priorite / contenu, un id derive du type (command-<hex8>), et un ordonnancement total inverse : __lt__ est ecrit pour que la priorite la plus HAUTE soit la plus « petite » — sorted() sort donc CRITICAL en premier, meme arrive en dernier. A priorite egale, le plus ancien passe d’abord (FIFO).

# Banc message_cases : id derive, ordonnancement inverse, FIFO.
m = Message(MessageType.COMMAND, "strategic-1", AgentLevel.STRATEGIC, {"task": "analyse"})
print("id du message COMMAND :", m.id, "(prefixe", m.id.split("-")[0], ")")
print()

msgs = [
    Message(MessageType.INFORMATION, "s", AgentLevel.SYSTEM, {}, priority=MessagePriority.LOW, timestamp=T0),
    Message(MessageType.INFORMATION, "s", AgentLevel.SYSTEM, {}, priority=MessagePriority.CRITICAL, timestamp=T0 + timedelta(seconds=5)),
    Message(MessageType.INFORMATION, "s", AgentLevel.SYSTEM, {}, priority=MessagePriority.HIGH, timestamp=T0 + timedelta(seconds=3)),
    Message(MessageType.INFORMATION, "s", AgentLevel.SYSTEM, {}, priority=MessagePriority.NORMAL, timestamp=T0 + timedelta(seconds=2)),
]
print("Ordre de sorted() :")
for x in sorted(msgs):
    print(f"  {x.priority.value:8s} (arrive a {x.timestamp.strftime('%H:%M:%S')})")
print()
a = Message(MessageType.INFORMATION, "s", AgentLevel.SYSTEM, {}, timestamp=T0)
b = Message(MessageType.INFORMATION, "s", AgentLevel.SYSTEM, {}, timestamp=T0 + timedelta(seconds=5))
print("FIFO a priorite egale :", "a avant b" if sorted([b, a]) == [a, b] else "inverse")

id du message COMMAND : command-3813f55f (prefixe command )

Ordre de sorted() :
  critical (arrive a 12:00:05)
  high     (arrive a 12:00:03)
  normal   (arrive a 12:00:02)
  low      (arrive a 12:00:00)

FIFO a priorite egale : a avant b

Lecture du resultat. L’id commence par command- : le prefixe dit deja la nature du message. L’ordonnancement est inverse : CRITICAL (arrive dernier, a 12:00:05) sort en premier, LOW (arrive premier) sort en dernier — __lt__ compare les priorites en les inversant, puis les timestamps en FIFO a priorite egale. C’est un choix deliberement contre-intuitif : la priorite la plus haute est la plus « petite » pour que sorted() la place en tete. La convention est simple et verifiable — mais elle impose une discipline : toute comparaison de messages doit passer par sorted(), jamais par un min() ou un > manuel qui lirait l’inverse.

Le banc dict_round_trip complete le tableau : to_dict() puis from_dict() preserve l’identite (l’egalite porte sur id, priorite et horodatage). La serialisation est le contrat de passage entre processus — sans elle, le bus ne sort pas de la memoire d’un seul programme.

2. Le contrat de canal — filtres fail-loud

Un canal est identifie par un id et un type ; il maintient une file d’attente et un registre d’abonnes avec filtres. Le matcher est fail-loud : une cle hors contrat leve ValueError plutot que d’etre ignoree — l’ignorance silencieuse elargit le filtre sans que personne ne le voie (incident #2161 : trois defauts coexistaient sans un seul test rouge).

# Banc filter_cases : le contrat parle en chaines, fail-loud sur cle inconnue.
ch = LocalChannel("test")
m = Message(MessageType.COMMAND, "tactical-1", AgentLevel.TACTICAL,
            {"task": "analyse"}, priority=MessagePriority.HIGH, timestamp=T0)

print("message_type='command'     :", ch.matches_filter(m, {"message_type": "command"}))
print("message_type='information' :", ch.matches_filter(m, {"message_type": "information"}))
print("liste [command, info]      :", ch.matches_filter(m, {"message_type": ["command", "information"]}))
print("priority='high'            :", ch.matches_filter(m, {"priority": "high"}))
print("content task='analyse'     :", ch.matches_filter(m, {"content": {"task": "analyse"}}))
print("content task='autre'       :", ch.matches_filter(m, {"content": {"task": "autre"}}))
print()
try:
    ch.matches_filter(m, {"bogus_key": "value"})
except ValueError as e:
    print("Cle inconnue -> ValueError :", str(e)[:80], "...")

message_type='command'     : True
message_type='information' : False
liste [command, info]      : True
priority='high'            : True
content task='analyse'     : True
content task='autre'       : False

Cle inconnue -> ValueError : Unknown filter criteria key 'bogus_key' — honored keys: ['content', 'message_typ ...

Lecture du resultat. Le contrat parle en chaines : command, pas l’enumere — une liste est un OU logique, un scalaire un ET strict. Le filtre de contenu est une sous-correspondance : chaque cle du filtre doit exister dans le contenu du message avec la meme valeur. Et le comportement critique : une cle inconnue leve ValueError au lieu d’etre ignoree. Sans ce fail-loud, un filtre {"priorite": "high"} (faute de frappe) passerait tous les messages — un filtre mort qui a l’air vivant. Le cout d’une erreur levee est visible ; le cout d’un filtre elargi est invisible jusqu’au incident.

Le banc topic_filter_cases du sas complete le contrat : les filtres de topic (pour le canal pub/sub) suivent la meme logique — chaines, listes, sous-correspondance de contenu. La famille de filtres est coherente entre tous les canaux : un abonne qui sait filtrer sur le canal local sait filtrer sur le canal pub/sub, sans apprendre une deuxieme syntaxe.

3. Le routage — le middleware decide, sans routes mortes

determine_channel est la seule partie du middleware portee : les canaux concrets et le protocole requete-reponse restent sur le tronc. Les regles sont simples : canal explicite > type de message > contenu > bus par defaut.

# Banc routing_cases : canal explicite, type, contenu, bus par defaut.
def route(t, content=None, channel=None):
    m = Message(t, "s", AgentLevel.SYSTEM, content or {}, channel=channel)
    return determine_channel(m).value

print("COMMAND                    ->", route(MessageType.COMMAND))
print("INFORMATION analysis_result ->", route(MessageType.INFORMATION, {"info_type": "analysis_result"}))
print("INFORMATION status         ->", route(MessageType.INFORMATION, {"info_type": "status"}))
print("REQUEST assistance         ->", route(MessageType.REQUEST, {"request_type": "assistance"}))
print("REQUEST query              ->", route(MessageType.REQUEST, {"request_type": "query"}))
print("RESPONSE                   ->", route(MessageType.RESPONSE))
print("PUBLICATION                ->", route(MessageType.PUBLICATION))
print("EVENT (pas de route morte) ->", route(MessageType.EVENT))
print()
print("Canal explicite 'data'     ->", route(MessageType.COMMAND, channel="data"))
print("Canal invalide 'bogus'     ->", route(MessageType.COMMAND, channel="bogus"))

COMMAND                    -> hierarchical
INFORMATION analysis_result -> data
INFORMATION status         -> hierarchical
REQUEST assistance         -> collaboration
REQUEST query              -> hierarchical
RESPONSE                   -> hierarchical
PUBLICATION                -> data
EVENT (pas de route morte) -> hierarchical

Canal explicite 'data'     -> data
Canal invalide 'bogus'     -> hierarchical

Lecture du resultat. Les commandes et reponses vont sur le canal hierarchique ; les informations portant un resultat d’analyse vont sur le canal donnees ; les requetes d’assistance vont sur le canal collaboration ; les publications vont sur le canal donnees. Et le comportement #1571 : EVENT (comme CONTROL et SUBSCRIPTION) ne route plus vers feedback ou system — ces canaux n’ont pas d’implementation, les messages y arriveraient sans jamais etre delivres. Ils tombent sur le bus par defaut (hierarchique). Un canal explicite prime toujours ; un canal invalide retombe sur les regles. C’est une table de routage qui ne promet que des transports qui existent.

Le banc routing_cases du sas compte 10 cas — les sept rejoues ici plus trois cas de repli (canal explicite prime, canal invalide retombe, type inconnu tombe sur le bus par defaut). Le middleware du tronc ajoute la statistique (stats["messages_sent"], by_channel) et le registre de gestionnaires par type — le port ne les porte pas : ce sont des compteurs d’observabilite, pas des garanties du contrat. Le carnet source les cite ; ce carnet les mesure par l’usage.

4. Le canal local — file d’attente, abonnes, notification synchrone

LocalChannel est l’implementation minimale du contrat : file d’attente en memoire, notification synchrone des abonnes au moment de l’envoi. Il sert de reference pour les quatre canaux concrets du tronc, qui ajoutent chacun leur semantique (persistance, topics, hierarchie).

# Le canal local en action : envoi, reception, abonnement, filtre.
ch = LocalChannel("bus")
received = []

ch.subscribe("logger", callback=received.append)
ch.subscribe("urgent", callback=lambda m: received.append(f"URGENT: {m.id}"),
             filter_criteria={"priority": "critical"})

ch.send_message(Message(MessageType.INFORMATION, "agent-1", AgentLevel.OPERATIONAL,
                        {"data": "result"}, recipient="agent-2"))
ch.send_message(Message(MessageType.COMMAND, "chef", AgentLevel.STRATEGIC,
                        {"task": "stop"}, priority=MessagePriority.CRITICAL))

print("Messages recus par 'logger' :", len(received))
print("Notifications URGENT        :", len([r for r in received if isinstance(r, str)]))
print("En attente pour 'agent-2'  :", len(ch.get_pending_messages("agent-2")))
print("Recu par agent-2           :", ch.receive_message("agent-2").id)
print("Info canal                 :", ch.get_channel_info())

Messages recus par 'logger' : 3
Notifications URGENT        : 1
En attente pour 'agent-2'  : 2
Recu par agent-2           : information-166fa997
Info canal                 : {'id': 'bus', 'type': 'local', 'subscribers': 2, 'pending': 1}

Lecture du resultat. Le logger recoit les deux messages (pas de filtre), l’abonne urgent n’en recoit qu’un (filtre critical). Les messages restent en file tant qu’ils ne sont pas lus — receive_message les consomme. C’est un bus synchrone : l’envoi notifie immediatement, la reception consomme. Les quatre canaux concrets du tronc ajoutent chacun une semantique au-dessus de ce contrat : le canal hierarchique ajoute des niveaux d’autorisation, le canal collaboration ajoute des conversations, le canal donnees ajoute la persistance, le canal pub/sub ajoute des topics. Tous partagent le meme contrat — c’est ce qui les rend interchangeables.

Le banc collaboration_cases du sas mesure ce que le canal collaboration ajoute : une conversation (un fil conducteur entre plusieurs messages), des participants, et un etat partage qui survit a l’envoi. Le banc data_cases mesure la persistance : un message envoye sur le canal donnees survit a la reception — il est stocke, pas consomme. Le banc pubsub_cases mesure les topics : un abonne ne recoit que les messages dont le topic correspond a son abonnement. Ces trois semantiques sont des extensions du meme contrat — le canal local n’en a besoin d’aucune pour etre utilisable.

5. La correlation requete-reponse — sans threads, sans timeouts

Le protocole requete-reponse complet du tronc (RequestResponseProtocol) introduit des threads et des timeouts — hors d’un port deterministe. Ce qui est porte est la convention de correlation : une reponse inverse emetteur et destinataire, et porte reply_to l’id de la requete.

# Banc request_response_cases : la convention de correlation.
req = Message(MessageType.REQUEST, "strategic-1", AgentLevel.STRATEGIC,
              {"request_type": "assistance"}, recipient="tactical-1",
              message_id="request-demo", timestamp=T0)
resp = create_response(req, AgentLevel.TACTICAL, {"result": "ok"})

print("Requete  :", req.type.value, "de", req.sender, "vers", req.recipient, "id", req.id)
print("Reponse  :", resp.type.value, "de", resp.sender, "vers", resp.recipient, "reply_to", resp.metadata["reply_to"])
print()
print("Correlation :", resp.metadata["reply_to"] == req.id)
print("Inversion   :", resp.sender == req.recipient and resp.recipient == req.sender)

Requete  : request de strategic-1 vers tactical-1 id request-demo
Reponse  : response de tactical-1 vers strategic-1 reply_to request-demo

Correlation : True
Inversion   : True

Lecture du resultat. La reponse porte reply_to: request-demo — le lien entre requete et reponse est un champ de metadata, pas un thread ni une socket. L’emetteur de la reponse est le destinataire de la requete, et inversement. C’est une convention de correlation minimale : elle suffit a reconstruire une conversation sans etat partage. Le protocole complet du tronc ajoute les timeouts et les reessais — mais la correlation, elle, est deja la.

Le banc request_response_cases du sas compte 3 cas : la correlation par reply_to, le timeout (une requete sans reponse leve RequestTimeoutError), et le reessai (retry_count). Les deux derniers sont hors d’un port deterministe — ils introduisent le temps reel et les threads. Ce qui est porte est ce qui ne depend pas du temps : la convention de correlation. Le timeout est un mecanisme de robustesse, pas une garantie de contrat.

Exercices

Les trois exercices suivent la convention de la serie : le stub s’execute sans erreur, la solution est a ecrire.

# Exercice 1 — Un filtre compose.
# Completez : un filtre qui ne laisse passer que les commandes CRITICAL
# envoyees par un agent STRATEGIC.
filtre = {}  # TODO etudiant
ch = LocalChannel("exo1")
m_ok = Message(MessageType.COMMAND, "chef", AgentLevel.STRATEGIC, {}, priority=MessagePriority.CRITICAL)
m_ko = Message(MessageType.INFORMATION, "agent", AgentLevel.OPERATIONAL, {}, priority=MessagePriority.LOW)
print("Message OK passe  :", ch.matches_filter(m_ok, filtre) if filtre else "Exercice a completer")
print("Message KO bloque :", not ch.matches_filter(m_ko, filtre) if filtre else "Exercice a completer")

Message OK passe  : Exercice a completer
Message KO bloque : Exercice a completer

# Exercice 2 — Le routage manquant.
# Completez : quelle regle de routage manque pour que les messages EVENT
# aillent sur le canal DATA quand leur contenu porte "event_type": "metric" ?
# Indice : ajoutez la regle dans determine_channel, ou ecrivez-la ici.
def route_event(m):
    if m.type == MessageType.EVENT and m.content.get("event_type") == "metric":
        return None  # TODO etudiant : quel ChannelType ?
    return determine_channel(m)
print("Routage EVENT metric :", "Exercice a completer")

Routage EVENT metric : Exercice a completer

# Exercice 3 — Une conversation complete.
# Completez : envoyez une requete sur un canal local, recevez-la, creez la
# reponse, renvoyez-la, et verifiez que l'emetteur original la recoit.
ch = LocalChannel("exo3")
req = Message(MessageType.REQUEST, "agent-A", AgentLevel.TACTICAL,
              {"query": "statut"}, recipient="agent-B", timestamp=T0)
# TODO etudiant : ch.send_message(req), puis agent-B la recoit, cree la reponse,
# la renvoie, et agent-A la recoit.
print("Conversation complete :", "Exercice a completer")

Conversation complete : Exercice a completer

A retenir

Le format message est un contrat, pas une structure de donnees. L’id derive du type, la priorite est inversee pour sorted(), la correlation est un champ de metadata — chaque choix est une convention verifiable, pas un accident d’implementation.
Un filtre qui ignore une cle est un filtre elargi. Le fail-loud (ValueError sur cle inconnue) est la difference entre un filtre qui dit ce qu’il fait et un filtre qui fait ce qu’on croit qu’il dit.
Le routage ne promet que des transports qui existent. Trois types de canal sans implementation ont ete retires du routage (#1571) plutot que cables a des classes vides — un canal promis sans implementation est un mensonge du bus.
Le contrat rend les canaux interchangeables. Hierarchique, collaboration, donnees, pub/sub : tous implementent le meme contrat (send_message / receive_message / subscribe / get_pending_messages), tous sont testables avec LocalChannel.

Position dans la serie : Orchestration_Modes compare les architectures d’agents par le budget ; ce carnet mesure ce qui les relie — le bus de communication. La suite de la distillation sas est knowledge_base (la memoire du debat) et dialogue_protocols (les regles de l’echange).

Le bus est le tissu conjonctif : sans lui, les agents ne se parlent pas. La memoire (knowledge_base) enregistre ce qui a ete dit ; les protocoles (dialogue_protocols) decident qui peut dire quoi ; les cadres (Dung_AF_Semantics) arbitrent ce qui survit. Quatre niveaux d’un meme debat — le bus est le premier.

Retour au sommet