Kafka

Einrichtung

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, Streaming Analytics 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"
  }
}

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.

• Stellt die Verbindung mit exponentiellem Backoff wieder her und bestätigt Offsets nach erfolgreicher Verarbeitung

Einrichtung

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