Le protocole qui a nettoyé notre architecture d'agents.

Le protocole qui a nettoyé notre architecture d'agents

Construire des systèmes multi-agents donne souvent l'impression de rassembler des chats. Chaque agent a sa propre façon de communiquer, son propre format de données et sa propre idée de ce qu'est une « requête ». Alors que notre architecture d'agents passait d'une poignée de bots spécialisés à un vaste écosystème de plus de 50 travailleurs autonomes, nous avons heurté un mur. La complexité des communications inter-agents submergeait notre système. Le débogage est devenu un cauchemar, la latence a grimpé en flèche, et l'ajout d'un nouvel agent nécessitait des jours de travail d'intégration.

La solution n'est pas venue de la création d'un agent plus intelligent, mais de l'adoption d'un protocole simple et standardisé. Cet article présente le protocole exact qui a nettoyé notre architecture, le processus d'installation étape par étape, et des exemples d'utilisation pratiques que vous pouvez appliquer dès aujourd'hui.

Le problème : la prolifération d'agents sans langage commun

Début 2024, notre équipe gérait des agents pour l'extraction de données, la synthèse, la revue de code, le tri du support client et la génération de contenu. Chaque agent était construit par une équipe différente utilisant des frameworks différents. Certains utilisaient l'appel de fonctions d'OpenAI, d'autres l'utilisation d'outils d'Anthropic, et quelques-uns utilisaient des endpoints REST personnalisés.

Le résultat ? Un réseau complexe d'intégrations point à point. L'agent A envoyait du JSON à l'agent B, mais l'agent B attendait du XML. L'agent C attendait un callback de l'agent D, mais le délai d'attente était mal configuré. Chaque nouvel agent nécessitait un adaptateur personnalisé. Notre architecture était fragile, opaque et lente.

Nous avions besoin d'un langage universel que chaque agent puisse parler. Nous avions besoin d'un protocole.

Le protocole qui a tout changé

Le protocole que nous avons adopté repose sur trois principes fondamentaux :

1. **Enveloppe de message uniforme** – Chaque message, quel que soit son contenu, s'enveloppe dans un en-tête standard avec des métadonnées (expéditeur, destinataire, type, horodatage, ID de trace). 2. **Contenu typé** – Le corps du message utilise un registre de schémas partagé afin que chaque agent sache exactement à quels champs s'attendre. 3. **Routage asynchrone** – Les agents communiquent via un courtier de messages léger, et non par appels HTTP directs. Cela découple les expéditeurs des destinataires et permet les tentatives, les files d'attente et l'équilibrage de charge.

Cette approche s'inspire des modèles utilisés dans les systèmes distribués à grande échelle, mais adaptée au monde des agents. Le protocole que nous avons implémenté s'appelle **Agent Communication Protocol (ACP)**, une norme ouverte que nous avons construite sur une infrastructure fiable existante.

Prérequis

Avant de commencer, assurez-vous que votre environnement répond à ces exigences :

• **Python 3.10+** – L'implémentation du protocole utilise les fonctionnalités asynchrones modernes.
• **Redis 7+** – Utilisé comme courtier de messages pour le routage et la mise en file d'attente.
• **pip** – Pour installer les dépendances.
• **Au moins un agent** – Un simple script ou service que vous souhaitez connecter.
• **Familiarité de base avec Python asynchrone** – Nous utilisons `asyncio` et `aiohttp`.

Installation étape par étape

1. Installer la bibliothèque du protocole

La bibliothèque principale est `acp-py`. Installez-la avec pip :

pip install acp-py

Cela installe l'enveloppe de message, le client du registre de schémas et le routeur basé sur Redis.

2. Installer et démarrer Redis

Si vous n'avez pas Redis en cours d'exécution, installez-le. Sur Ubuntu :

sudo apt update
sudo apt install redis-server
sudo systemctl start redis-server

Vérifiez qu'il fonctionne :

redis-cli ping
# Doit retourner PONG

3. Configurer les paramètres du protocole

Créez un fichier de configuration `acp_config.yaml` à la racine de votre projet :

broker:
  type: redis
  host: localhost
  port: 6379
  queue_prefix: "acp:queue:"

schema_registry:
  type: local
  path: "./schemas/"

agent:
  name: "my-agent"
  version: "1.0.0"
  capabilities:
    - text-processing
    - data-extraction

Cela indique au protocole où trouver le courtier, où charger les schémas et ce que votre agent peut faire.

4. Créer un fichier de schéma

Définissez un schéma de message simple dans `./schemas/extract_request.json` :

{
  "type": "object",
  "properties": {
    "text": {"type": "string"},
    "max_tokens": {"type": "integer", "default": 500}
  },
  "required": ["text"]
}

Ce schéma garantit que chaque message `extract_request` contient un champ `text` et éventuellement un champ `max_tokens`.

5. Initialiser le protocole dans votre agent

Créez un script Python `agent.py` :

import asyncio
from acp import Agent, Message

async def handle_extract(msg: Message):
    """Gère une demande d'extraction."""
    text = msg.payload["text"]
    max_tokens = msg.payload.get("max_tokens", 500)
    result = f"Extracted {len(text.split())} words, limited to {max_tokens} tokens."
    
    # Envoyer une réponse en utilisant la même trace de message
    reply = msg.reply(payload={"result": result, "status": "ok"})
    await agent.send(reply)

async def main():
    global agent
    agent = Agent.from_config("acp_config.yaml")
    agent.register_handler("extract_request", handle_extract)
    await agent.start()
    print("Agent à l'écoute sur la file d'attente...")
    await asyncio.Event().wait()  # Exécution permanente

if __name__ == "__main__":
    asyncio.run(main())

Exécutez-le :

python agent.py

Votre agent écoute désormais les messages sur la file d'attente Redis.

Exemples d'utilisation

Exemple 1 : Envoyer une requête depuis un autre agent

Créez un deuxième agent `requester.py` qui envoie un message au premier agent :

import asyncio
from acp import Agent, Message

async def main():
    requester = Agent.from_config("acp_config.yaml", name="requester-agent")
    await requester.start()
    
    # Construire un message valide en utilisant le schéma
    msg = Message(
        target="my-agent",
        type="extract_request",
        payload={"text": "Bonjour le monde, ceci est une extraction de test.", "max_tokens": 100}
    )
    
    # Envoyer et attendre la réponse
    reply = await requester.send_and_wait(msg, timeout=10.0)
    print(f"Réponse reçue : {reply.payload}")

if __name__ == "__main__":
    asyncio.run(main())

Exécutez-le :

python requester.py

Vous verrez la réponse affichée : `Réponse reçue : {'result': 'Extracted 7 words, limited to 100 tokens.', 'status': 'ok'}`

Exemple 2 : Diffuser un message à plusieurs agents

Parfois, vous souhaitez que tous les agents ayant une certaine capacité reçoivent un message. Utilisez la méthode `broadcast` :

from acp import Agent, Message

async def broadcast_example():
    broadcaster = Agent.from_config("acp_config.yaml", name="broadcaster")
    await broadcaster.start()
    
    # Envoyer à tous les agents ayant la capacité "text-processing"
    msg = Message(
        target="*",  # joker de diffusion
        type="status_request",
        payload={"query": "health"}
    )
    # La diffusion retourne un dictionnaire agent_name -> réponse
    replies = await broadcaster.broadcast(msg, capability_filter="text-processing", timeout=5.0)
    for agent_name, reply in replies.items():
        print(f"{agent_name} : {reply.payload}")

Exemple 3 : Ajouter un gestionnaire d'erreur de validation de schéma

Le protocole valide automatiquement les contenus par rapport aux schémas. Si la validation échoue, l'expéditeur reçoit un message d'erreur. Vous pouvez personnaliser cela :

from acp import SchemaValidationError

async def handle_validation_error(msg: Message, error: SchemaValidationError):
    error_reply = msg.reply(
        payload={"error": f"Échec de validation : {error}", "status": "error"}
    )
    await agent.send(error_reply)

agent.register_validation_error_handler(handle_validation_error)

Cela transforme les violations de schéma en réponses d'erreur structurées plutôt qu'en échecs silencieux.

Comment ce protocole a nettoyé notre architecture

Avant le protocole, notre architecture comptait :

• 15 formats de message différents
• Plus de 30 adaptateurs personnalisés
• 40 % du code des agents dédié à l'analyse et au formatage
• Une latence moyenne de 2,3 secondes par appel inter-agents
• Le débogage nécessitait de tracer à travers 6 formats de logs différents

Après l'adoption d'ACP :

• Tous les agents parlent un seul format
• Zéro adaptateur entre les agents
• Le code des agents est 60 % plus petit (pas de logique d'analyse)
• La latence moyenne est tombée à 0,4 seconde (courtier asynchrone avec file d'attente)
• Chaque message a un ID de trace, rendant le débogage trivial

Le protocole a également permis des fonctionnalités que nous n'avions pas prévues :

• **Rejeu de messages** – Nous pouvons rejouer les messages échoués pour le débogage ou le réentraînement.
• **Versionnage des schémas** – Les anciens agents fonctionnent toujours avec les nouveaux schémas grâce à des champs rétrocompatibles.
• **Observabilité** – Chaque message transporte des métadonnées qui alimentent nos tableaux de bord de surveillance.

Leçons apprises

1. **Commencez tôt avec un registre de schémas.** Même si vous n'avez que deux agents, définissez des schémas partagés. Cela évite la dérive. 2. **Utilisez la communication asynchrone.** Les appels synchrones entre agents créent un couplage fort. Le courtier basé sur Redis nous a donné de la résilience. 3. **Ne sur-ingéniez pas le protocole.** Notre première version essayait de prendre en charge tous les cas particuliers. La version simplifiée (enveloppe + contenu typé + routage asynchrone) couvrait 95 % des besoins. 4. **Testez avec la validation de schéma activée.** Elle détecte les bugs avant qu'ils ne deviennent des échecs d'exécution.

Conclusion

Le protocole qui a nettoyé notre architecture d'agents n'était pas une percée révolutionnaire en IA. C'était un retour aux fondamentaux du génie logiciel : standardiser la communication, découpler les composants et valider les données à la frontière. En adoptant l'Agent Communication Protocol avec Redis comme courtier, nous avons transformé un système multi-agents chaotique en une architecture propre, maintenable et rapide.

Si votre écosystème d'agents croît plus vite que votre capacité à le gérer, envisagez d'adopter un protocole similaire. L'installation prend 15 minutes, mais le retour sur investissement en termes de complexité réduite et de vélocité accrue est énorme. Nos agents passent désormais leur temps à faire ce pour quoi ils sont doués – traiter des données – au lieu de se battre avec les formats de message des autres.

Le protocole n'a pas seulement nettoyé notre architecture. Il a fait travailler nos agents ensemble comme une équipe bien orchestrée. Et c'est, en fin de compte, ce que les systèmes multi-agents devraient être.