Praxiserfahrung mit Streaming in ChatGPT Anwendungen

Jeder, der schon einmal mit ChatGPT interagiert hat, kennt das Feature: ChatGPT baut seine Antworten Wort für Wort, oder genauer gesagt, Token für Token auf. Dieses Verhalten wird als “Streaming” bezeichnet.

Warum ist Streaming wichtig?

Das Generieren einer Antwort kann durchaus seine Zeit in Anspruch nehmen. Manchmal kann es sogar bis zu 1,5 Minuten dauern, bis eine Antwort vollständig generiert ist. Die Latenz, also die Zeit bis die Antwort generiert wurde, hängt hauptsächlich von der Länge der generierten Antwort ab. Die Rechenzeit wird pro generiertem Token berechnet, wobei in geringerem Maße auch die Länge des Inputs eine Rolle spielt.

Eine lange Wartezeit kann jedoch zu einer suboptimalen Benutzererfahrung führen. Selbst wenn man in einem Chat-Interface eine Animation integriert, die signalisiert, dass der Bot gerade schreibt, haben Nutzer wenig Geduld. Nach mehr als 10 Sekunden nehmen viele an, dass ein Systemfehler vorliegt. Interessanterweise ist die Toleranz der Nutzer noch geringer, wenn sie wissen, dass die Antwort von einer KI generiert wird.

Unsere Erfahrung mit Streaming

In den Softwarelösungen, die wir entwickelt haben, nutzen wir die APIs von OpenAI und anderen Anbietern. Hierbei ist es möglich, die Antwort von der API in einzelnen Teilen, sogenannten Tokens, zu streamen. Ohne diese Streaming-Fähigkeit könnten wir dem Nutzer erst eine Antwort liefern, wenn das Language Model das letzte Token generiert hat.

Unsere eigenen Lösungen, wie der “Lieblingskollege”, mehrphasige Coaching-Flows und semantische Suche, nutzen ebenfalls dieses Streaming-Feature.

Technologische Lösungen für Streaming

Traditionelle REST-APIs senden für jede Anfrage einen Request und erhalten eine Antwort. Für Streaming möchten wir jedoch einen Request senden und mehrere Antworten vom Server erhalten. Hierfür gibt es verschiedene Ansätze:

Polling

• Pro: Einfach zu implementieren.
• Con: Sehr ineffizient, da man in kurzen Abständen den Server abfragen muss. Dies kann zu einer Überlastung der Infrastruktur führen, insbesondere wenn man den aktuellen Zustand über alle Server-Instanzen synchronisieren muss. Damit mehrere Server-Instanzen auf Anfragen reagieren können, muss der Zustand zwischen den Servern synchronisiert werden (z.B. über Redis)

Websockets

• Pro: Wie gemacht für bidirektionale Echtzeitkommunikation ohne Polling.
• Con: Nicht trivial zu implementieren, da man den gesamten Kommunikationsinhalt selbst definieren muss. Bei Verbindungsabbrüchen, insbesondere bei mobilen Geräten, muss die Verbindung manuell wiederhergestellt werden, was zusätzlichen Implementierungsaufwand bedeutet.

Server Sent Events (SSE)

• Pro: Nutzt eine bestehende HTTP-Verbindung, automatische Wiederherstellung bei Verbindungsabbrüchen, einfache Implementierung in vielen Frameworks wie FastAPI.
• Con: Der Kommunikationsinhalt muss immer noch selbst definiert werden, obwohl die Handhabung von Verbindungsabbrüchen einfacher ist als bei Websockets.

Nach sorgfältiger Abwägung haben wir uns für die Implementierung mit Server Sent Events entschieden.

Implementierung in Python mit FastAPI

Unser bevorzugtes REST Framework in Python, FastAPI, bietet eine einfache Implementierung von SSE. Mit dem Paket sse-starlette konnten wir eine zuverlässige Lösung implementieren. Ein kleiner Stolperstein war das korrekte Encoding der Inhalte. Während FastAPI normalerweise das Encoding übernimmt, mussten wir bei der Implementierung von SSE selbst Hand anlegen. Mit dem Befehl json_message = message.json(ensure_ascii=False) konnten wir jedoch sicherstellen, dass die Inhalte korrekt und ohne unerwünschte ASCII-Konvertierung gesendet werden.

Herausforderungen in Flutter

Während unserer Arbeit mit Flutter stießen wir auf einen Bug im Dart SDK, der bereits seit vier Jahren offen war. Dieser Bug verhinderte eine optimale Implementierung von Server Sent Events. Erstaunlicherweise wurde dieser Bug erst im Mai 2023 wieder aktiv diskutiert. Der Grund? Viele Entwickler wollten nun OpenAI und andere generative KI-Dienste integrieren und stießen auf dasselbe Problem. Ein kleines Beispiel dafür, wie sich die Prioritäten in der Softwareentwicklung mit der Zeit ändern können!

In Flutter mussten wir eigene Implementierungen für SSE schreiben. Für Web-Anwendungen konnten wir die Browser-Funktionen nutzen, während wir für native Anwendungen die Flutter HTTP Library verwendeten. Beide Implementierungen waren überschaubar im Aufwand und funktionierten zuverlässig.

Da die HTTP Verbindung mit dem Server bestehen bleibt, können Exceptions nicht wie gewohnt durch FastAPI Handler und HTTP Status Code behandelt werden. Sie müssen als Payload über den Event Stream der Server Sent Events gesendet werden:

def wrap_streaming_response_exceptions(generator: Generator) -> Generator:
    try:
        yield from generator
    except TimeoutError as exc:
        logger.error(f"TimeoutError raised: {exc}")
        yield SSEError(code=status.HTTP_504_GATEWAY_TIMEOUT, message=f"Timeout error, {exc}").json()
        return
    # more exceptions

Die Fehlermeldungen kommen dann als normale Payload auf Client-Seite an, und müssen wie eine normale Payload behandelt werden.

Fazit

Server Sent Events bieten eine effiziente und zuverlässige Lösung für das Streaming in generativen KI-Anwendungen. Mit FastAPI in Python ist die Implementierung unkompliziert, und auch in Flutter lässt sich mit etwas Handarbeit eine solide Lösung realisieren. Das Ergebnis: Eine Benutzererfahrung, die sowohl uns als Entwickler als auch unsere Nutzer zufriedenstellt.