Einen KI-Agenten zu streamen ist ein Protokollproblem

Die Demo-Version eines KI-Chat-Features baust du an einem Nachmittag. Du öffnest eine Streaming-Response, leitest Tokens weiter, der Text erscheint Buchstabe für Buchstabe, alle sind begeistert. Dann stellst du das Ganze vor echte Nutzer mit echter Arbeit, und du lernst: Das Streaming war nie das Feature. Das Feature ist ein Protokoll, und fast alles Interessante daran handelt davon, was passiert, wenn der Happy Path eben nicht passiert.

Normans In-App-Assistent ist ein Agent, der Tools benutzt. Er kann sich deine Transaktionen ansehen, eine Rechnung entwerfen, eine Zahlung vorbereiten, eine Umsatzsteuer-ID erklären. Über unseren Produktions-Traffic haben wir Zehntausende Chat-Turns protokolliert und knapp hunderttausend getracte Agent-Runs über alle unsere KI-Workflows hinweg. Die Chat-Turns sind die langsamsten und sichtbarsten davon, und sie haben eine Form, mit der ein naiver Token-Stream schlecht umgeht:

• Der Median-Turn ist nach rund zwei Sekunden fertig.
• Das 90. Perzentil liegt bei etwa fünfzehn Sekunden.
• Das 99. Perzentil liegt bei etwa dreißig Sekunden, und die schlimmsten Turns, die wir gemessen haben, laufen über eine Minute.

Ungefähr jede fünfte Antwort ruft mindestens ein Tool auf. Genau das sind die langsamen Turns, und genau da hängt der Nutzer am meisten an der Antwort, weil er den Assistenten gebeten hat, etwas zu tun, statt etwas zu erklären.

Diese Verteilung ist der ganze Design-Auftrag. Ein Protokoll, das sich bei zwei Sekunden wohlfühlt und bei dreißig auseinanderfällt, reicht nicht, denn die Dreißig-Sekunden-Turns sind die, auf die es ankommt.

Der Transport ist absichtlich langweilig

Unser Backend ist eine synchrone Django-Anwendung. Die Agent-Schleife ist asynchron. Diese beiden Tatsachen wollen keine Freunde sein, und es gibt die bekannte Versuchung, den halben Stack umzuschreiben, damit sie es werden.

Haben wir nicht. Die Streaming-View startet die asynchrone Agent-Schleife in einem Daemon-Thread, und diese Schleife schiebt fertige Zeilen in eine gewöhnliche thread-sichere Queue. Der Response-Generator läuft im normalen synchronen Worker, leert die Queue und gibt jede Zeile an die HTTP-Response weiter, bis er ein Sentinel sieht:

def generate_streaming_content():
    q: sync_queue.Queue = sync_queue.Queue()

    async def run_async() -> None:
        try:
            async for chunk in wrapper.process_assistant_by_prompt_streamed(...):
                q.put(chunk + "\n")
        except Exception:
            logger.exception("Streaming view error")
            q.put(json.dumps({"type": "error", "message": "..."}) + "\n")
        finally:
            q.put(None)  # sentinel

    threading.Thread(target=lambda: asyncio.run(run_async()), daemon=True).start()

Die Payload ist zeilenweise getrenntes JSON. Ein Event pro Zeile, keine Framing-Kunststücke, keine halben Objekte. Der Client behält die letzte unvollständige Zeile in einem Buffer und parst nur ganze. Das sind vier Zeilen Client-Code, und es ist die einzige Framing-Regel im gesamten Protokoll.

Ich mag diese Schicht gerade deshalb, weil es nichts über sie zu sagen gibt. Sie war nie das, was kaputtging.

Jeder Event-Typ ist additiv

Der Stream trägt eine kleine Menge typisierter Events: Text-Deltas, einen Tool-Start, ein Bild, eine Action Card, ein Completion, einen Fehler und eine nachgereichte Speicherbestätigung. Der Client verzweigt über type mit einer schlichten if-Kette und ignoriert stillschweigend alles, was er nicht kennt.

Genau diese eine Eigenschaft, unbekannte Events werden übersprungen statt fatal, erlaubt uns, das Protokoll ohne koordiniertes Release weiterzuentwickeln. Web, iOS und Android konsumieren denselben Stream und aktualisieren sich nach völlig verschiedenen Zeitplänen. Wäre ein neuer Event-Typ ein Parse-Fehler, würde jede Verbesserung am Assistenten zu einem Release-Zug.

Beide Events, die wir diese Woche ausgeliefert haben, sind der Beweis, dass es funktioniert. Keines brauchte eine Client-Änderung, um sicher deploybar zu sein, sondern nur, um nützlich zu sein.

Das Problem ist die Stille, nicht die Langsamkeit

Das sagt dir niemand, bevor du einem Chat-Agenten Tools gibst: Ein Tool-Aufruf streamt nichts. Das Modell entscheidet sich für ein Tool, und dann wird es im Turn komplett still, so lange dieses Tool eben braucht. Keine Tokens, kein Fortschritt, nichts auf der Leitung.

In diesen zehn bis dreißig Sekunden zeigte unsere UI einen Tipp-Indikator. Ein Tipp-Indikator ist in diesem Moment eine Lüge. Das Modell tippt nicht, es liest drei Monate Transaktionen, und der Nutzer kann "arbeitet hart" nicht von "hängt" unterscheiden.

Die Lösung ist keine bessere Animation. Sie ist ein Event. Die Streaming-Schleife kannte den Tool-Namen ohnehin, weil sie damit Tool-Ausgaben auf Action Cards abbildet, also senden wir ihn jetzt:

if event.item.type == "tool_call_item":
    self.current_tool_id = _tool_name_of(event.item.raw_item)
    # Ein Turn kann zig Sekunden in Tool-Aufrufen verbringen, ohne dass
    # etwas gestreamt wird. Ohne das hier zeigt die UI nur einen Tipp-
    # Indikator, und der Nutzer weiß nicht, ob überhaupt etwas passiert.
    if self.current_tool_id:
        yield json.dumps({"type": "tool_start", "tool_name": self.current_tool_id})

Jetzt kann der Client "Rechnung wird erstellt" sagen statt drei animierte Punkte zu zeigen. Gleiche Latenz, völlig andere Erfahrung. Eine langsame Operation, die dir sagt, was sie tut, ist erträglich. Eine schnelle, die schweigt, nicht.

Die ID kommt nach dem Text, und das mit Absicht

Unsere Chat-Antworten haben Daumen-hoch- und Daumen-runter-Buttons. Reaktionen adressieren eine Nachricht über ihre gespeicherte ID.

Eine gestreamte Antwort hat keine. Die Datenbankzeile wird erst geschrieben, wenn der Stream fertig ist, weil es vorher keinen finalen Text zu schreiben gibt. Also rendert der Client eine Nachricht ohne ID, und die einzigen Antworten, auf die du wirklich reagieren konntest, waren die, die du durch Neuladen der Historie bekamst. Das Feedback-Signal, das wir am dringendsten wollten, war praktisch unerreichbar. Unsere Produktionsdaten sagen genau das: Reaktionen über die gesamte Produkthistorie sind nahezu keine, was eine sehr höfliche Art ist zu sagen, dass der Button dekorativ war.

Die Lösung ist ein nachgereichtes Event. Nach dem Completion speichern wir die Nachricht und senden ihre ID:

saved_message_id = await self._save_assistant_message_async(accumulated_message)
# Bewusst nach "completion" gesendet: die Zeile existiert erst, wenn der
# Stream durch ist, und Clients dürfen fürs Rendern nicht darauf warten.
yield json.dumps({"type": "message_saved", "message_id": saved_message_id})

Der Kommentar zur Reihenfolge ist der wichtige Teil. Es wäre aufgeräumter, die ID zuerst zu senden und ein einziges autoritatives Event zu haben, das heißt "die Antwort ist vollständig und hier ist alles dazu". Es würde aber auch bedeuten, dass ein fehlgeschlagener Schreibvorgang eine völlig gute Antwort daran hindern kann, angezeigt zu werden. Rendern darf nie von Persistenz abhängen. Also ist die Antwort fertig, und die ID holt einen Moment später auf, um eine sekundäre Funktion freizuschalten.

Retry ist nur sicher, bevor das erste Tool läuft

Streams brechen ab. Upstream-Verbindungen fallen weg, Anbieter liefern transiente Serverfehler, Netzwerke tun Netzwerkdinge. Die naheliegende Reaktion ist, den Turn zu wiederholen.

Die naheliegende Reaktion ist gefährlich. Unser Assistent legt Rechnungen, Transaktionen und Zahlungen an. Einen Turn zu wiederholen, der bereits create_transaction ausgeführt hat, liefert keine bessere Antwort, sondern zwei Transaktionen.

Die Retry-Regel hängt deshalb an zwei Zuständen, die die Streaming-Schleife ohnehin mitführt:

is_transient = isinstance(e, APIError) and getattr(e, "type", None) == "server_error"
if is_transient and attempt < MAX_STREAM_RETRIES and not tool_executed:
    await asyncio.sleep(attempt + 1)
    continue

tool_executed wird wahr, sobald irgendein Tool-Aufruf Output zurückgibt. Ab da ist der Turn keine reine Funktion seiner Eingabe mehr, und wir dürfen nicht so tun, als wäre er es. Dieselbe Regel gilt für eine abgebrochene Verbindung, mit einem Zusatz: Wir wiederholen auch dann nicht, wenn bereits Text gestreamt wurde, denn der Nutzer hat ihn gesehen, und die Antwort von vorne zu beginnen ist eine eigene Art von kaputt.

Das ist derselbe Instinkt, der sich durch den Rest unseres Systems zieht. Wenn etwas die Außenwelt schon berührt hat, ist der einzige sichere Zug nach vorne.

Eine Teilantwort schlägt einen roten Fehler

Wenn wir nicht wiederholen können, retten wir. Bricht ein Stream ab, nachdem echter Inhalt angekommen ist, werfen wir den Text nicht weg. Wir speichern, was wir haben, senden ein Completion-Event damit und schicken die Message-ID hinterher, genau wie bei einem erfolgreichen Turn. Der Nutzer sieht eine kurze Antwort statt eines Fehlers, und sie bleibt in seiner Historie wie jede andere Nachricht.

Nur ein Abbruch ohne jeden Inhalt wird zu einem sichtbaren Fehler, und selbst dann ist es ein schlichter Satz mit der Bitte, es gleich noch einmal zu versuchen, kein Stacktrace.

Das Prinzip: Ein Turn, der etwas produziert hat, endet als Nachricht, und ein Turn, der nichts produziert hat, endet als Entschuldigung. Es gibt keinen dritten Fall, in dem der Nutzer raten muss.

Die Verbindung, an die du nicht gedacht hast

Mein Lieblingsbug in diesem System hatte nichts mit dem Modell zu tun.

Nachrichten aus einer asynchronen Schleife zu persistieren heißt, dass das ORM in Executor-Threads läuft. Django recycelt seine Datenbankverbindungen anhand von Signalen zu Request-Start und Request-Ende, und diese Signale feuern nie für selbst gestartete Threads. Ein langer Turn kann eine gepoolte Verbindung also über die gesamte Dauer des Modellaufrufs im Leerlauf lassen, lange genug, dass die Gegenseite sie schließt. Die nächste Query scheitert dann daran, dass die Verbindung weg ist, in einem ansonsten völlig gesunden Request.

Die Lösung ist klein und konkret:

def _retry_on_stale_connection(func):
    """Erholt sich von einer im Leerlauf verlorenen Postgres-Verbindung.

    Jeder umschlossene Helper macht genau ein create oder save ohne zuvor
    committete Seiteneffekte, deshalb ist es idempotent, die tote
    Verbindung zu verwerfen und exakt einmal zu wiederholen.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except (OperationalError, InterfaceError):
            close_old_connections()
            return func(*args, **kwargs)
    return wrapper

Beachte die Begründung im Docstring, nicht nur den Mechanismus. Der Retry ist erlaubt, weil die umschlossene Operation ein einzelner Schreibvorgang ohne vorher committete Änderungen ist. Das ist der Unterschied zwischen einem sicheren und einem hoffnungsvollen Retry, und es gehört dorthin geschrieben, wo die nächste Person es liest.

Jedes Timeout im Stack muss den langsamsten Turn überleben

Zurück zur Latenzverteilung: p99 bei rund dreißig Sekunden. Und jetzt stell dir ein Worker-Request-Timeout von fünfundzwanzig Sekunden vor, ein völlig vernünftiger Default für eine transaktionale API.

Diese Kombination ist eine Maschine zur Herstellung mysteriöser Fehler. Der Worker wird mitten im Turn beendet, die asynchrone Schleife stirbt ohne Aufräumen, verwaiste Tool-Subprozesse bleiben liegen, und der Nutzer sieht einen generischen Fehler genau bei den Requests, die am meisten geleistet haben. In den Anwendungslogs steht nichts von "Timeout", weil der Prozess, der es geloggt hätte, nicht mehr da ist.

Die Lehre daraus ist eine Regel, die wir jetzt bewusst anwenden:

1. Das Request-Timeout muss den langsamsten Turn übertreffen, den du ausliefern willst, nicht den mittleren.
2. Jedes innere Timeout, besonders das für Tool-Aufrufe, muss kürzer sein als das Request-Timeout, damit ein langsames Tool als fangbare Exception scheitert und nicht als getöteter Prozess.
3. Streaming-Endpunkte gehören nicht in dasselbe Worker-Budget wie transaktionale. Ein Request, der einen Worker dreißig Sekunden hält, und einer, der in achtzig Millisekunden zurückkommt, sollten nicht um dieselben Slots konkurrieren.

Es gibt eine vierte Regel, die wir gerade ausrollen, und sie ist die am wenigsten offensichtliche: sende früh ein Byte. Proxies vor einer API wiederholen üblicherweise einen Request, der noch keine Response-Bytes produziert hat. Für einen normalen Endpunkt ist das ein nützliches Sicherheitsnetz. Für einen Streaming-Endpunkt, der zwanzig Sekunden nachdenken kann, bevor das erste Token kommt, ist es eine Gefahr der Doppelausführung. Ein trivialer erster Chunk, geflusht bevor der Agent startet, schließt dieses Fenster.

Das Modell ist der einfache Teil

Nichts davon handelt von Prompting. Jede schwierige Entscheidung hier ist eine alte Entscheidung aus verteilten Systemen in neuen Kleidern: Was ist idempotent, was darf wiederholt werden, was darf umsortiert werden, was darf das Rendern nie blockieren, und welches Timeout ist das kürzeste in der Kette.

Wenn du so etwas baust, ist die kürzeste nützliche Zusammenfassung, die ich geben kann: Schreib deinen Event-Vertrag auf, bevor du die Streaming-Schleife schreibst, entscheide für jedes Event, ob es Pflicht oder additiv ist, und entscheide für jeden Fehlerfall, ob er eine Nachricht oder eine Entschuldigung ergibt. Das Token-Streaming funktioniert beim ersten Versuch. Alles andere ist das eigentliche Produkt.