Startseite / Artikel / Wiederaufnahmefähiger LLM-Streaming mit Redis Streams als Treffpunkt pro Turnus

redis streaming llm agents gateway realtime

Wiederaufnahmefähiger LLM-Streaming mit Redis Streams als Treffpunkt pro Turnus

Überleben Sie Unterbrechungen und mehrminütige Pausen der Tools, indem Sie Agentenevents in einen turn-basierten Redis Stream veröffentlichen, den die Clients wieder aufnehmen können.

2049 Wörter

Was wir wollten

Die Antworten der Agenten können mehrere Sekunden dauern: Überlegungen anstellen, eine Fähigkeit laden, Tools aufrufen, warten und Tokens streamen. In Demos bleibt eine offene Verbindung bestehen. In der Produktion wird die Kommunikation mitten in einer Antwort unterbrochen, Gateways werden neu bereitgestellt und es entstehen Pausen bei Tools, die erst Minuten später antworten. Das Ziel ist fortsetzbares Echtzeit-Streaming, damit ein Client sich wieder verbinden und denselben Dialog fortsetzen kann.

Der Weg dorthin

Iteration 1: Der Client spricht direkt mit dem Agenten

Einfach, aber anfällig. Jeder Netzwerkfehler beendet den Stream. Horizontale Skalierung bedeutet entweder sticky Sessions oder verlorene Ereignisse.

Iteration 2: gRPC-Streaming zwischen Diensten

Bessere interne Schnittstellen, doch immer noch unpraktisch für Browser-Clienten und anfällig bei mehrminütigen Pausen durch Neustarts der Pods.

Warum nicht Kafka?

Sehr gut für langlebige Protokolle; schwerer, als es für ein Treffen pro Runde mit kurzer Speicherungsdauer sowie Konsumierergruppen nötig ist, die sich schlecht auf „eine Browserleiste“ abbilden lassen.

Iteration 3 (was funktioniert hat): benannte Treffen pro Runde

// Agent output
{
  "type": "tool_result",
  "tool": "product_search",
  "data": {
    "items": [...]
  }
}

// Gateway -> TV
{
  "type": "product_carousel",
  "items": [...]
}

// Gateway -> Mobile
{
  "type": "product_list",
  "items": [...]
}

cursor = last_event_id or "0-0"

while True:
    entries = xread({key: cursor}, block=30_000)

    if not entries:          # the only timeout check point
        check_timeouts()
        continue

    for entry_id, event in entries:
        # writes to the socket; not an ack that the client received it
        sse.send(id=entry_id, data=event.payload)
        cursor = entry_id
        if event.type in TERMINAL:
            return

id: 1755600000123-0
data: {"type":"tool_selected","tool":"search"}

id: 1755600000871-0
data: {"type":"response_block","block":{...}}

GET /sessions/{sid}/turns/{tid}/stream
Last-Event-ID: 1755600000871-0

# turn starts: one atomic step (MULTI/EXEC, or a Lua script)
xadd(key, first_event)
expire(key, GENEROUS_TTL)

# producer finishes: bring it in
expire(key, RECONNECT_TTL)

Jede Benutzer-Runde erhält einen Redis Stream (oder das Pattern Stream+Konsumierergruppe), der nach turn_id keyt wird. Der Agent veröffentlicht Token-/Tool-Ereignisse; das Gateway folgt vom letzten id des Clients. Die Wiederverbindung setzt an diesem Cursor fort. Pods können absterben; der Stream behält genügend Historie für die Runde.

Wichtige Details

Cursor-Disziplin – Clients bestätigen den zuletzt gesehenen Stream-ID; nach der ersten Verbindung wird niemals von 0 aus weitergemacht.

Heartbeat-Ereignisse – verhindern, dass Zwischenkomponenten inaktive Verbindungen während des Wartens auf das Tool schließen.

Turn-Lebenszyklus – explizite Werte für turn_started / turn_paused / turn_completed / turn_failed.

TTL – Streams werden nach Abschluss des Turns abgelaufen, damit Redis nicht zu einer endlosen Archivierung wird.

Authz – turn_id stellt keine Autorisierung dar; Turns müssen mit der authentifizierten Sitzung verknüpft werden.

Der Fall, der die Lösung brachte: eine mehrminütige Pause

Ein Tool übergab die Aufgabe an einen anderen Pipeline, der erst Minuten später antwortete. Die direkte HTTP-Streaming-Verbindung brach ab. Mit Redis Streams veröffentlichte der Agent ein Pause-Ereignis, der Client zeigte an, dass weitergearbeitet wird, und später wurden die Tokens bei der Neukonnektierung ohne Neuablauf des gesamten Plans wiederhergestellt.

Was wir an den Client senden

Eingetippte Ereignisse: Tokens, tool_start, Zusammenfassungen von tool_result (niemals vertrauliche Informationen), Fehler sowie Abschluss. Halten Sie die Payloads klein; lagern Sie umfangreiche Daten in Object Storage und senden Sie nur Referenzen.

Kosten, Grenzen, zu beachtende Aspekte

Achten Sie auf den Redis-Speicherbedarf, die maximale Länge der Streams sowie auf das Ausbreitungsverhalten, falls viele Gateways an einer Aufgabe arbeiten. Begrenzen Sie die Anzahl der gleichzeitigen Aufgaben pro Benutzer. Führen Sie Lasttests durch, um Wiederverbindungsanfragen nach dem Deploy eines Gateways zu überprüfen.

Der Endzustand

Das Gateway ist ein lesender Prozess, der mit Fortschrittsinformationen umgehen kann; der Agent ist ein schreibender Prozess; Redis Streams dient als Treffpunkt. Eine Echtzeit-UX übersteht die langweiligen Fehler, die typischerweise Demo-Architekturen zum Scheitern bringen.

Operative Tipps: Speichern Sie die letzte Stream-ID im Session-Cookie oder in der Client-Memory sowie auch auf Serverseite, um das Wiedergeben zu ermöglichen. Wenn ein Benutzer angibt, dass alles eingefroren sei, sollte der Support den jeweiligen Schritt aus dem Stream rekonstruieren, anstatt von ihm zu verlangen, auf eine zehnminütige Wartezeit zu warten. Fügen Sie Dashboards hinzu, die die Wiederaufnahmerate, abgebrochene Schritte sowie die durchschnittliche Pausezeit anzeigen, damit das Produkt erkennen kann, ob die Agenten langsam arbeiten oder die Netzwerke instabil sind.

Operative Tipps: Speichern Sie die letzte Stream-ID im Session-Cookie oder in der Client-Memory sowie auch auf Serverseite, um das Wiedergeben zu ermöglichen. Wenn ein Benutzer angibt, dass alles eingefroren sei, sollte der Support den jeweiligen Schritt aus dem Stream rekonstruieren, anstatt von ihm zu verlangen, eine zehnminütige Wartezeit nachzuvollziehen. Fügen Sie Dashboards hinzu, die die Wiederaufnahmerate, abgebrochene Schritte sowie die durchschnittliche Pausezeit anzeigen, damit das Produkt erkennen kann, ob die Agenten langsam arbeiten oder die Netzwerke instabil sind.

Operative Tipps: Speichern Sie die letzte Stream-ID im Session-Cookie oder in der Client-Memory sowie auch auf Serverseite, um das Wiedergeben zu ermöglichen. Wenn ein Benutzer angibt, dass alles eingefroren sei, sollte der Support den jeweiligen Schritt aus dem Stream rekonstruieren, anstatt von ihm zu verlangen, eine zehnminütige Wartezeit zu wiederholen. Fügen Sie Dashboards hinzu, die die Wiederaufnahmerate, abgebrochene Schritte sowie die durchschnittliche Pausezeit anzeigen, damit das Produkt erkennen kann, ob die Agenten langsam arbeiten oder die Netzwerke instabil sind.

Operative Tipps: Speichern Sie die letzte Stream-ID im Session-Cookie oder in der Client-Memory sowie auch auf Serverseite, um das Wiedergeben zu ermöglichen. Wenn ein Benutzer angibt, dass alles eingefroren sei, sollte der Support den jeweiligen Schritt aus dem Stream rekonstruieren, anstatt von ihm zu verlangen, eine zehnminütige Wartezeit nachzuvollziehen. Fügen Sie Dashboards hinzu, die die Wiederaufnahmerate, abgebrochene Schritte sowie die durchschnittliche Pausezeit anzeigen, damit das Produkt erkennen kann, ob die Agenten langsam arbeiten oder die Netzwerke instabil sind.

Operative Tipps: Speichern Sie die letzte Stream-ID im Session-Cookie oder in der Client-Memory sowie auch auf Serverseite, um das Wiedergeben zu ermöglichen. Wenn ein Benutzer angibt, dass alles eingefroren sei, sollte der Support den jeweiligen Schritt aus dem Stream rekonstruieren, anstatt ihn bitten zu müssen, zehn Minuten lang auf das Tool zu warten. Fügen Sie Dashboards hinzu, die die Wiederaufnahmerate, abgebrochene Schritte sowie die durchschnittliche Pausezeit anzeigen, damit das Produkt erkennen kann, ob die Agenten langsam arbeiten oder die Netzwerke instabil sind.

Operative Tipps: Speichern Sie die letzte Stream-ID im Session-Cookie oder in der Client-Memory sowie auch auf Serverseite, um das Wiedergeben zu ermöglichen. Wenn ein Benutzer angibt, dass alles eingefroren sei, sollte der Support den jeweiligen Schritt aus dem Stream rekonstruieren, anstatt von ihm zu verlangen, eine zehnminütige Wartezeit nachzuvollziehen. Fügen Sie Dashboards hinzu, die die Wiederaufnahmerate, abgebrochene Schritte sowie die durchschnittliche Pausezeit anzeigen, damit das Produkt erkennen kann, ob die Agenten langsam arbeiten oder die Netzwerke instabil sind.

Alle Artikel Brief senden