AI Agents · Lektion

Kommunikationsprotokolle (Message-Busse)

Entkoppeln Sie Agents mit einem Message-Bus (Redis, NATS, Kafka), damit sie unabhängig skalieren und ausfallen können.

Kommunikationsprotokolle (Message-Busse) ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

Über In-Process-Konversationen hinaus

In kleinen Systemen rufen Agenten einander innerhalb desselben Prozesses auf. In größeren Systemen entkoppeln Sie sie über einen Nachrichtenbus – Redis, NATS oder Kafka.

Warum entkoppeln?

Agenten können unabhängig voneinander skaliert werden
Fehler bleiben auf den jeweiligen Agenten begrenzt
Asynchrone Workflows (langsame Agenten müssen nicht inline abgewartet werden)
Nachrichten können zur Fehlersuche erneut abgespielt werden
Sprachübergreifend: Ein Python-Agent und ein Go-Agent können beide abonnieren

Simple Bus: Redis Pub/Sub

import redis
r = redis.Redis()

# Publisher
r.publish('agent.research.task', json.dumps({'task_id': 'abc', 'query': '...'}))

# Subscriber
p = r.pubsub()
p.subscribe('agent.research.task')
for msg in p.listen():
    if msg['type'] == 'message':
        handle_task(json.loads(msg['data']))

Pub/Sub zum Broadcasten

Verwenden Sie dies, wenn viele Agenten von einem Ereignis erfahren möchten (z. B. „user-question-received“).

Warteschlangen zur Arbeitsverteilung

Verwenden Sie eine Warteschlange (Redis BLPOP, RabbitMQ), wenn genau ein Worker jede Aufgabe bearbeiten soll:

import json, queue

r = queue.Queue()

# Producer
r.put(json.dumps({'task_id': 'abc', 'kind': 'research'}))

def process(task):
    print('processing', task)

# Worker
while not r.empty():
    raw = r.get()
    task = json.loads(raw)
    process(task)

NATS für Geschwindigkeit

NATS ist ein leichter Nachrichtenbroker für Microservices. Latenz im Submillisekundenbereich und integrierte Request/Reply-Unterstützung:

import nats
nc = await nats.connect('nats://localhost:4222')

await nc.publish('agent.research', json.dumps(task).encode())

# Request/reply
response = await nc.request('agent.research', payload, timeout=10)

Kafka für dauerhafte Speicherung

Kafka bietet dauerhafte, erneut abspielbare Logs. Ideal für Auditing und Replay:

from confluent_kafka import Producer, Consumer

producer.produce('agent-events', key=task_id, value=json.dumps(task))
producer.flush()

Event-Schemas

Definieren Sie die Struktur jeder Nachricht mit Pydantic oder Protobuf. Ohne Schemas wird Ihr Bus zum Chaos:

class ResearchTask(BaseModel):
    task_id: str
    user_id: str
    query: str
    deadline: datetime

class ResearchResult(BaseModel):
    task_id: str
    findings: list[str]
    duration_ms: int

Korrelations-IDs

Jede zusammengehörige Nachricht enthält dieselbe correlation_id, damit Sie eine Aufgabe über mehrere Services hinweg nachverfolgen können:

record = {'correlation_id': 'abc', 'task_id': 'def', 'span': 'llm_call'}
print(record)

Idempotente Handler

Nachrichten können mehr als einmal zugestellt werden (At-Least-Once-Semantik). Machen Sie Handler idempotent – dieselbe task_id wird nur einmal verarbeitet.

Dead-Letter-Warteschlange

Wenn ein Handler wiederholt fehlschlägt, verschieben Sie die Nachricht zur manuellen Prüfung in eine DLQ, anstatt endlos weiterzuschleifen:

if attempts > MAX_RETRIES:
    r.rpush('queue:research:dlq', raw)
    log.error('Sent to DLQ', extra={'task_id': task_id})

Beobachtbarkeit

Verfolgen Sie jede Nachricht:

Trace-IDs werden über den Bus weitergegeben
Logs enthalten task_id und correlation_id
Metriken: Durchsatz, Latenz und Fehlerrate pro Topic

Wann sollten Sie umsteigen?

Beginnen Sie einfach – mit In-Process-Agenten. Wechseln Sie erst dann zu einem Bus, wenn Folgendes zutrifft:

Mindestens 5 Agenten laufen in Produktion
Agenten müssen unabhängig voneinander skaliert werden
Workflows erstrecken sich über mehrere Services

At-Least-Once?

Was erfordert „at-least-once delivery“ von Ihren Handlern?

Zusammenfassung

Entkoppeln Sie einfache Fälle mit Redis, nutzen Sie NATS für geringe Latenz und Kafka für dauerhafte Speicherung. Schemas, Korrelations-IDs, idempotente Handler und eine DLQ.

Häufig gestellte Fragen

Ist die Lektion „Kommunikationsprotokolle (Message-Busse)“ kostenlos?

Ja — der vollständige Text von „Kommunikationsprotokolle (Message-Busse)“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Kommunikationsprotokolle (Message-Busse)“?

Entkoppeln Sie Agents mit einem Message-Bus (Redis, NATS, Kafka), damit sie unabhängig skalieren und ausfallen können. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Kommunikationsprotokolle (Message-Busse)“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

Multi-Agent-Kommunikation (AutoGen)
Hierarchische Supervisors (Orchestrator + Workers)
Agent-Rollen und Spezialisierungen
Kommunikationsprotokolle (Message-Busse)

← Zurück zu AI Agents