Das Protokoll, das unsere Agentenarchitektur bereinigt hat

Das Protokoll, das unsere Agentenarchitektur bereinigt hat

Der Aufbau von Multi-Agenten-Systemen fühlt sich oft an wie das Hüten von Katzen. Jeder Agent hat seine eigene Art zu kommunizieren, sein eigenes Datenformat und seine eigene Vorstellung davon, wie eine "Anfrage" auszusehen hat. Als unsere Agentenarchitektur von einer Handvoll spezialisierter Bots zu einem ausufernden Ökosystem mit über 50 autonomen Workern heranwuchs, stießen wir an eine Wand. Die Komplexität der Kommunikation zwischen den Agenten überforderte unser System. Debugging wurde zum Albtraum, die Latenzzeiten schossen in die Höhe, und das Hinzufügen eines neuen Agenten erforderte Tage der Integrationsarbeit.

Die Lösung kam nicht durch einen intelligenteren Agenten, sondern durch die Einführung eines einfachen, standardisierten Protokolls. Dieser Artikel beschreibt genau das Protokoll, das unsere Architektur bereinigt hat, die schrittweise Installation und praktische Anwendungsbeispiele, die Sie noch heute umsetzen können.

Das Problem: Agenten-Wildwuchs ohne gemeinsame Sprache

Anfang 2024 betrieb unser Team Agenten für Datenextraktion, Zusammenfassung, Code-Review, Kundensupport-Triage und Inhaltserstellung. Jeder Agent wurde von einem anderen Team mit unterschiedlichen Frameworks entwickelt. Einige nutzten OpenAIs Function Calling, andere Anthropics Tool Use, und einige verwendeten benutzerdefinierte REST-Endpunkte.

Das Ergebnis? Ein undurchschaubares Netz aus Punkt-zu-Punkt-Integrationen. Agent A sendete JSON an Agent B, aber Agent B erwartete XML. Agent C wartete auf einen Callback von Agent D, aber das Timeout war falsch eingestellt. Jeder neue Agent erforderte einen benutzerdefinierten Adapter. Unsere Architektur war fragil, undurchsichtig und langsam.

Wir brauchten eine universelle Sprache, die jeder Agent sprechen konnte. Wir brauchten ein Protokoll.

Das Protokoll, das alles veränderte

Das von uns übernommene Protokoll basiert auf drei Kernprinzipien:

1. **Einheitlicher Nachrichtenumschlag** – Jede Nachricht, unabhängig vom Inhalt, wird in einen Standard-Header mit Metadaten (Absender, Empfänger, Typ, Zeitstempel, Trace-ID) verpackt. 2. **Typisierte Nutzdaten** – Der Inhalt verwendet ein gemeinsames Schema-Register, sodass jeder Agent genau weiß, welche Felder zu erwarten sind. 3. **Asynchrones Routing** – Agenten kommunizieren über einen leichtgewichtigen Message Broker, nicht über direkte HTTP-Aufrufe. Dies entkoppelt Sender und Empfänger und ermöglicht Wiederholungen, Warteschlangen und Lastverteilung.

Dieser Ansatz ist inspiriert von Mustern aus großen verteilten Systemen, aber für die Welt der Agenten angepasst. Das von uns implementierte Protokoll heißt **Agent Communication Protocol (ACP)**, ein offener Standard, den wir auf Basis bestehender zuverlässiger Infrastruktur aufgebaut haben.

Voraussetzungen

Stellen Sie vor dem Start sicher, dass Ihre Umgebung diese Anforderungen erfüllt:

• **Python 3.10+** – Die Protokollimplementierung nutzt moderne Async-Funktionen.
• **Redis 7+** – Wird als Message Broker für Routing und Warteschlangen verwendet.
• **pip** – Zur Installation von Abhängigkeiten.
• **Mindestens ein Agent** – Ein einfaches Skript oder ein Dienst, den Sie anbinden möchten.
• **Grundlegende Vertrautheit mit async Python** – Wir verwenden `asyncio` und `aiohttp`.

Schritt-für-Schritt-Installation

1. Installieren Sie die Protokollbibliothek

Die Kernbibliothek ist `acp-py`. Installieren Sie sie mit pip:

pip install acp-py

Dies installiert den Nachrichtenumschlag, den Schema-Registry-Client und den Redis-basierten Router.

2. Installieren und starten Sie Redis

Wenn Redis noch nicht läuft, installieren Sie es. Auf Ubuntu:

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

Überprüfen Sie, ob es läuft:

redis-cli ping
# Sollte PONG zurückgeben

3. Konfigurieren Sie die Protokolleinstellungen

Erstellen Sie eine Konfigurationsdatei `acp_config.yaml` im Stammverzeichnis Ihres Projekts:

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

Dies teilt dem Protokoll mit, wo der Broker zu finden ist, wo Schemata geladen werden und was Ihr Agent kann.

4. Erstellen Sie eine Schema-Datei

Definieren Sie ein einfaches Nachrichtenschema in `./schemas/extract_request.json`:

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

Dieses Schema stellt sicher, dass jede `extract_request`-Nachricht ein `text`-Feld und optional ein `max_tokens`-Feld enthält.

5. Initialisieren Sie das Protokoll in Ihrem Agenten

Erstellen Sie ein Python-Skript `agent.py`:

import asyncio
from acp import Agent, Message

async def handle_extract(msg: Message):
    """Bearbeitet eine Extraktionsanfrage."""
    text = msg.payload["text"]
    max_tokens = msg.payload.get("max_tokens", 500)
    result = f"Extracted {len(text.split())} words, limited to {max_tokens} tokens."
    
    # Sende eine Antwort mit derselben Nachrichten-Trace-ID
    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 hört auf der Warteschlange...")
    await asyncio.Event().wait()  # Läuft für immer

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

Führen Sie es aus:

python agent.py

Ihr Agent hört nun auf Nachrichten in der Redis-Warteschlange.

Anwendungsbeispiele

Beispiel 1: Senden einer Anfrage von einem anderen Agenten

Erstellen Sie einen zweiten Agenten `requester.py`, der eine Nachricht an den ersten Agenten sendet:

import asyncio
from acp import Agent, Message

async def main():
    requester = Agent.from_config("acp_config.yaml", name="requester-agent")
    await requester.start()
    
    # Erstelle eine gültige Nachricht mit dem Schema
    msg = Message(
        target="my-agent",
        type="extract_request",
        payload={"text": "Hallo Welt, dies ist eine Testextraktion.", "max_tokens": 100}
    )
    
    # Senden und auf Antwort warten
    reply = await requester.send_and_wait(msg, timeout=10.0)
    print(f"Antwort erhalten: {reply.payload}")

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

Führen Sie es aus:

python requester.py

Sie sehen die Antwort: `Antwort erhalten: {'result': 'Extracted 7 words, limited to 100 tokens.', 'status': 'ok'}`

Beispiel 2: Broadcast einer Nachricht an mehrere Agenten

Manchmal möchten Sie, dass alle Agenten mit einer bestimmten Fähigkeit eine Nachricht erhalten. Verwenden Sie die `broadcast`-Methode:

from acp import Agent, Message

async def broadcast_example():
    broadcaster = Agent.from_config("acp_config.yaml", name="broadcaster")
    await broadcaster.start()
    
    # Senden an alle Agenten mit der Fähigkeit "text-processing"
    msg = Message(
        target="*",  # Broadcast-Wildcard
        type="status_request",
        payload={"query": "health"}
    )
    # Broadcast gibt ein Dictionary agent_name -> reply zurück
    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}")

Beispiel 3: Hinzufügen eines Schema-Validierungsfehler-Handlers

Das Protokoll validiert Nutzdaten automatisch gegen Schemata. Bei fehlgeschlagener Validierung erhält der Sender eine Fehlermeldung. Sie können dies anpassen:

from acp import SchemaValidationError

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

agent.register_validation_error_handler(handle_validation_error)

Dies verwandelt Schemaverstöße in strukturierte Fehlerantworten anstelle von stillen Fehlern.

Wie dieses Protokoll unsere Architektur bereinigt hat

Vor dem Protokoll hatte unsere Architektur:

• 15 verschiedene Nachrichtenformate
• 30+ benutzerdefinierte Adapter
• 40% des Agentencodes für Parsing und Formatierung
• Durchschnittliche Latenz von 2,3 Sekunden pro agentenübergreifendem Aufruf
• Debugging erforderte die Verfolgung durch 6 verschiedene Logformate

Nach der Einführung von ACP:

• Alle Agenten sprechen ein Format
• Null Adapter zwischen Agenten
• Agentencode ist 60% kleiner (keine Parsing-Logik)
• Durchschnittliche Latenz auf 0,4 Sekunden gesunken (async Broker mit Warteschlangen)
• Jede Nachricht hat eine Trace-ID, was Debugging trivial macht

Das Protokoll ermöglichte auch Funktionen, die wir nicht geplant hatten:

• **Nachrichtenwiederholung** – Wir können fehlgeschlagene Nachrichten für Debugging oder Nachschulung wiederholen.
• **Schema-Versionierung** – Alte Agenten arbeiten weiterhin mit neuen Schemata durch abwärtskompatible Felder.
• **Beobachtbarkeit** – Jede Nachricht trägt Metadaten, die in unsere Überwachungs-Dashboards einfließen.

Gelernte Lektionen

1. **Beginnen Sie früh mit einem Schema-Register.** Selbst wenn Sie nur zwei Agenten haben, definieren Sie gemeinsame Schemata. Es verhindert Abweichungen. 2. **Verwenden Sie asynchrone Kommunikation.** Synchrone Aufrufe zwischen Agenten erzeugen enge Kopplung. Der Redis-Broker gab uns Resilienz. 3. **Überengineering Sie das Protokoll nicht.** Unsere erste Version versuchte, jeden Randfall abzudecken. Die vereinfachte Version (Umschlag + typisierte Nutzdaten + asynchrones Routing) deckte 95% der Bedürfnisse ab. 4. **Testen Sie mit aktivierter Schema-Validierung.** Sie fängt Fehler, bevor sie zu Laufzeitfehlern werden.

Fazit

Das Protokoll, das unsere Agentenarchitektur bereinigt hat, war kein revolutionärer KI-Durchbruch. Es war eine Rückkehr zu den Grundlagen der Softwareentwicklung: Kommunikation standardisieren, Komponenten entkoppeln und Daten an den Grenzen validieren. Durch die Einführung des Agent Communication Protocol mit Redis als Broker haben wir ein chaotisches Multi-Agenten-System in eine saubere, wartbare und schnelle Architektur verwandelt.

Wenn Ihr Agenten-Ökosystem schneller wächst, als Sie es verwalten können, sollten Sie die Einführung eines ähnlichen Protokolls in Betracht ziehen. Die Installation dauert 15 Minuten, aber der Gewinn an reduzierter Komplexität und erhöhter Geschwindigkeit ist enorm. Unsere Agenten verbringen jetzt ihre Zeit damit, das zu tun, was sie am besten können – Daten zu verarbeiten – anstatt sich mit den Nachrichtenformaten der anderen herumzuschlagen.

Das Protokoll hat nicht nur unsere Architektur bereinigt. Es hat unsere Agenten dazu gebracht, wie ein gut orchestriertes Team zusammenzuarbeiten. Und darum sollte es bei Multi-Agenten-Systemen letztendlich gehen.