Kafka

Einrichtung

1. 1

   Kafka-Cluster vorbereiten

   Stelle sicher, dass der Cluster läuft und das Quell-Topic existiert.

2. 2

   Verbindung erstellen

   Öffne den Agenten, klicke im Verbindungsflussdiagramm auf Add inbound connector, dann auf Create New Connector und wähle Apache Kafka.

3. 3

   Konfigurieren und erstellen

   Wähle den Mode Inbound (Consumer), gib Bootstrap Servers und Topic sowie optional Group ID, Auto Offset Reset und Sicherheitseinstellungen (SASL/SSL) ein. Klicke auf Create. Die Verbindung wird pausiert angelegt; klicke in der Verbindungsansicht auf Resume, damit connic Nachrichten aus dem Topic konsumiert.

So funktioniert Inbound

Inbound-Kafka-Verbindungen arbeiten als Consumer, die ein Topic abonnieren. Trifft eine Nachricht ein, wird sie geparst und an alle verknüpften Agenten weitergeleitet. Das eignet sich für ereignisgesteuerte Verarbeitung, Echtzeit-Pipelines, die Auswertung laufend eintreffender Daten und die Kommunikation zwischen Microservices.

{
  "order_id": "12345",
  "customer": "john@example.com",
  "items": ["widget-a", "widget-b"],
  "_kafka": {
    "topic": "orders",
    "partition": 0,
    "offset": 1542,
    "timestamp": 1705312800000,
    "key": "order-12345"
  }
}

Die _kafka-Metadaten enthalten topic, partition, offset, timestamp und key.

Nachrichten mit JSON-Objekten werden wie oben gezeigt mit ihren Feldern auf oberster Ebene weitergeleitet. Alle anderen Werte werden unter einem message-Schlüssel verpackt: Nicht-JSON-Werte kommen als {"message": "<raw text>", "_kafka": ...} an, und Nachrichten mit null-Wert (Compaction Tombstones) starten weiterhin Runs mit message: null. Verwende _kafka.key, um die gelöschte Entität zu identifizieren:

{
  "message": null,
  "_kafka": {
    "topic": "orders",
    "partition": 0,
    "offset": 1543,
    "timestamp": 1705312800000,
    "key": "order-12345"
  }
}

JSON-Objekte können ein files-Array mit Dateinamen, MIME-Typen und Base64-kodierten Inhalten enthalten. connic nimmt bis zu 10 vollständige Dateien an; es gelten 10 MiB pro Datei und insgesamt 10 MiB. Ungültige optionale Dateien werden einzeln ausgelassen. Sobald eine Datei die Anzahl- oder Größenbegrenzung überschreitet, bleiben die angenommenen Dateien erhalten; diese Datei und alle folgenden werden ausgelassen. file_issues enthält die Auslassungen mit stabilen Codes, auch wenn Dateien das Budget der Input-Prüfung überschreiten. Der Run kann selbst ohne verbleibende optionale Dateien fortfahren. Unter Linked Agents lässt sich Files required aktivieren, wenn die Aufgabe bei jeder Auslassung oder bei fehlenden Dateien stoppen muss.

End-to-End-Beispiel

Erstelle eine Inbound-Verbindung für das Quell-Topic, verknüpfe sie mit einem Agenten und verknüpfe außerdem eine Outbound-Verbindung, die die Agentenausgabe in einem Ziel-Topic veröffentlicht.

version: "1.0"

name: order-processor
type: llm
model: connic/gpt-5.6-luna
description: "Validate orders and compute routing"
system_prompt: |
  You receive an order event in JSON (from Kafka).
  1) Call orders.validate_order
  2) Call orders.score_risk
  3) Return JSON with order_id, status, risk_score, route

tools:
  - orders.validate_order
  - orders.score_risk
output_schema: order-result.json

from typing import Dict, Any

def validate_order(order_id: str, items: list[str]) -> Dict[str, Any]:
    """Basic order validation."""
    if not order_id or not items:
        return {"ok": False, "reason": "missing_fields"}
    return {"ok": True}

async def score_risk(customer: str, total: float) -> Dict[str, Any]:
    """Return a simple risk score and routing hint."""
    score = 0.02 if total < 100 else 0.12
    route = "standard" if score < 0.1 else "manual_review"
    return {"risk_score": score, "route": route}

Die JSON-Ausgabe des Agenten erscheint im Feld output des Outbound-Run-Wrappers:

{
  "order_id": "12345",
  "status": "approved",
  "risk_score": 0.02,
  "route": "standard"
}

Jede Verbindung verwendet eine Consumer Group, um verarbeitete Nachrichten zu verfolgen. Unterschiedliche Group IDs = dieselben Nachrichten für alle. Gleiche Group ID = Lastverteilung.

Eine Verbindung liest ihre Partitionen Nachricht für Nachricht. Um ein Topic mit hohem Volumen parallel zu konsumieren, erstelle mehrere Inbound-Verbindungen mit derselben Group ID und verknüpfe sie mit demselben Agenten. Kafka weist jede Partition genau einer davon zu.

connic committet den Offset einer Nachricht, sobald für jeden verknüpften Agenten ein Run in der Warteschlange gespeichert ist, also bevor diese Runs ausgeführt werden. Gespeicherte Runs überstehen Worker-Neustarts und Deployments. Schlägt das Speichern fehl, bleibt der Offset offen und der Consumer stellt dieselbe Nachricht nach kurzer Verzögerung erneut zu. Spätere Nachrichten der Verbindung warten, bis das gelingt. Nachrichten, deren Value oder Key kein gültiges UTF-8 ist, werden als fehlgeschlagene Verbindungs-Runs protokolliert und committet. Nach einem Broker- oder Netzwerkfehler verbindet sich der Consumer mit exponentiellem Backoff neu und setzt am zuletzt committeten Offset fort.

Ein Run, der nach dem Speichern fehlschlägt, löst weder eine erneute Zustellung noch eine Dead-Letter-Nachricht aus. Der fehlgeschlagene Run behält seine vollständige Eingabe unter Runs, wo du ihn erneut ausführen kannst. Wiederholungen innerhalb eines Runs folgen den retry_options des Agenten.

Wird eine Nachricht nach einem Teilfehler, einem Neustart vor dem Offset-Commit oder einem Rebalance der Consumer Group erneut zugestellt, verwendet sie den bereits gespeicherten Run. Jeder Agent erhält pro Nachricht und Consumer Group genau einen Run, auch über Verbindungen mit derselben Group ID hinweg.

Runs werden in Offset-Reihenfolge gespeichert, laufen aber parallel bis zu max_concurrent_runs. Zwei Nachrichten mit demselben Key können also gleichzeitig laufen. Für eine Reihenfolge pro Key ergänze eine Concurrency-Regel auf dem Message Key:

concurrency:
  key: _kafka.key
  on_conflict: queue

Runs mit demselben Key starten dann einzeln in Eingangsreihenfolge, Runs mit unterschiedlichen Keys laufen weiter parallel. Ein fehlgeschlagener Run gibt seinen Key frei, die nächste Nachricht für diesen Key läuft also trotzdem. Nachrichten ohne Key werden nicht serialisiert.

Einrichtung

1. 1

   Kafka-Cluster vorbereiten

   Stelle sicher, dass der Cluster läuft und das Ziel-Topic existiert, oder aktiviere die automatische Erstellung auf dem Broker.

2. 2

   Verbindung erstellen

   Öffne den Agenten, klicke auf Add outbound connector, dann auf Create New Connector und wähle Apache Kafka.

3. 3

   Konfigurieren und erstellen

   Wähle den Mode Outbound (Producer) und gib Bootstrap Servers sowie Topic ein. Klicke auf Create; Ergebnisse verknüpfter Agenten werden anschließend im Topic veröffentlicht.

So funktioniert Outbound

Automatische Outbound-Verbindungen veröffentlichen die Wrapper abgeschlossener Runs im konfigurierten Topic und können auf ausgewählte Inputs begrenzt werden. Agent-Tool- und Middleware-Outbound-Verbindungen veröffentlichen nur bei einem Aufruf.

Nur Runs mit dem Status completed werden veröffentlicht. Fehlgeschlagene und abgebrochene Runs werden übersprungen. Runs, die durch StopProcessing vorzeitig beendet wurden, schließen regulär ab und werden daher ebenfalls veröffentlicht; die StopProcessing-Antwort dient dabei als Ausgabe. Löse StopProcessing mit publish_outbound=False aus, um die automatische Outbound-Verbindung für diesen Run zu überspringen. Ein bereits erfolgter Aufruf einer Agent-Tool- oder Middleware-Outbound-Verbindung wird dadurch nicht rückgängig gemacht.

{
  "run_id": "550e8400-e29b-41d4-a716-446655440000",
  "agent_name": "order-processor",
  "status": "completed",
  "output": "Order processed successfully. Total: $234.56",
  "error": null,
  "started_at": "2024-01-15T10:30:00Z",
  "ended_at": "2024-01-15T10:30:05Z",
  "token_usage": {
    "input_tokens": 150,
    "output_tokens": 50,
    "thinking_tokens": 0,
    "cached_input_tokens": 0,
    "total_tokens": 200
  }
}

Enthält run_id, agent_name, status, output, error, Zeitstempel und token_usage.

Agent-Tool- und Middleware-Outbound-Verbindungen

Eine Agent-Tool-Outbound-Verbindung stellt einen bearbeitbaren Tool-Namen bereit, standardmäßig send_to_<connector_name>. Rufe eine Middleware-Outbound-Verbindung über send_connector mit ihrem konfigurierten Namen auf. Beide verwenden dieses von der Verbindung definierte Payload-Schema:

{
  "payload": {
    "order_id": "12345",
    "status": "approved"
  },
  "key": "order-12345"
}

payload wird zum Wert der Kafka-Nachricht. key ist optional. connic verwendet das konfigurierte Topic, die Verbindung, Serialisierung, Sicherheit, Wiederholungsstrategie und das Bridge-Routing, ohne dem Modell Zugangsdaten offenzulegen.

Bei einer automatischen Outbound-Verbindung verwendet ein durch eine Inbound-Kafka-Verbindung gestarteter Run erneut den Message Key der Quelle; andere Runs verwenden die Run-ID. Agent-Tool- und Middleware-Outbound-Verbindungen können key ausdrücklich angeben.

# Eingehende Nachricht mit dem Key "order-123"
Nachricht empfangen → Agent ausgelöst → Run abgeschlossen

# Ausgehende Nachricht verwendet denselben Key "order-123"
Ergebnis mit Key veröffentlicht: "order-123" (Quelle: ursprüngliche Nachricht)

# Ohne eingehenden Key wird run_id verwendet
Ergebnis mit Key veröffentlicht: "550e8400-e29b..." (Quelle: run_id)

• Bestätigung nach vollständiger Replikation und automatische Wiederholungsversuche