Streaming in Python implementieren

Vom Prototyp zur Produktion

Die vorherige Lektion hat die Konzepte des SSE-Streamings behandelt. Der schwierigere Teil bleibt: Ein Demonstrations-Loop in einen Client umzuwandeln, der unter Last standhält, Netzwerkfehler übersteht und eine Web-Oberfläche versorgt. Diese Lektion folgt der realen Projektentwicklung: zuerst das Gerüst, dann Fehlerbehandlung, dann asynchrone Verarbeitung, dann die HTTP-Exposition und schließlich die Messung.

Das Grundmuster

Alles beginnt mit diesem zehnzeiligen Gerüst, das Sie für jedes Projekt anpassen werden. Es erledigt zwei Dinge gleichzeitig: Es zeigt Tokens an, sobald sie eintreffen, und sammelt sie für die Rückgabe der Funktion.

from mistralai.client import Mistral
import os

client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))

def stream_chat(messages: list, model: str = "mistral-large-latest") -> str:
    """Sendet eine Streaming-Anfrage und zeigt Tokens in Echtzeit an."""
    stream = client.chat.stream(
        model=model,
        messages=messages
    )

    full_text = ""
    for chunk in stream:
        content = chunk.data.choices[0].delta.content
        if content:
            full_text += content
            print(content, end="", flush=True)

    print()  # Neue Zeile
    return full_text

Diese Version funktioniert für ein lokales Skript. Sie hat einen fatalen Mangel in der Produktion: Sie druckt in das Terminal, sodass sie nichts anderes als Terminal-Arbeiten verrichten kann.

Umfassende Fehlerbehandlung

Netzwerkfehler, Timeouts und Rate Limits sind keine Hypothesen: Sie sind bei echtem Verkehrsaufkommen Gewissheiten. Der untenstehende Client adressiert beide Probleme. Er wiederholt 429-Fehler automatisch mit exponentiellem Backoff und druckt nichts mehr aus: Er emittiert typisierte Ereignisse über yield, sodass der Aufrufer selbst entscheiden kann, was er damit tut – sie anzeigt, sie in ein WebSocket schiebt oder sie in eine Datenbank schreibt.

from mistralai.client import Mistral
from mistralai.client.errors import SDKError
import os
import time

client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))

def stream_with_retry(
    messages: list,
    model: str = "mistral-large-latest",
    max_retries: int = 3,
    max_tokens: int = 2000
) -> dict:
    """Streaming mit automatischer Wiederholung und Metriken."""
    for attempt in range(max_retries):
        try:
            stream = client.chat.stream(
                model=model,
                messages=messages,
                max_tokens=max_tokens
            )

            full_text = ""
            finish_reason = None
            chunk_count = 0

            for chunk in stream:
                delta = chunk.data.choices[0].delta
                if delta.content:
                    full_text += delta.content
                    chunk_count += 1
                    yield {"type": "token", "content": delta.content}

                if chunk.data.choices[0].finish_reason:
                    finish_reason = chunk.data.choices[0].finish_reason

            yield {
                "type": "done",
                "full_text": full_text,
                "finish_reason": finish_reason,
                "chunks": chunk_count
            }
            return

        except SDKError as e:
            if e.status_code == 429 and attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponentieller Backoff
                yield {"type": "retry", "wait": wait_time, "attempt": attempt + 1}
                time.sleep(wait_time)
            else:
                yield {"type": "error", "message": str(e), "status": e.status_code}
                return

        except Exception as e:
            yield {"type": "error", "message": str(e), "status": None}
            return

Der exponentielle Backoff 2 ** attempt wartet 1 Sekunde, dann 2, dann 4. Diese Progression ist nicht dekorativ: Wenn alle Ihre Clients im gleichen festen Rhythmus wiederholen würden, würden sie die API in synchronisierten Stößen treffen und die sehr Sättigung verlängern, der sie bereits ausgesetzt sind. Beachten Sie auch, dass nur Fehler 429 eine Wiederholung auslösen – eine Wiederholung eines 401-Fehlers aufgrund eines ungültigen Schlüssels würde nur die Diagnose verzögern.

Auf der Seite des Aufrufers wird die Verarbeitung zu einer einfachen Ereignis-Leseoperation:

messages = [
    {"role": "system", "content": "Sie sind ein technischer Assistent."},
    {"role": "user", "content": "Erklären Sie Designmuster in Python."}
]

for event in stream_with_retry(messages):
    if event["type"] == "token":
        print(event["content"], end="", flush=True)
    elif event["type"] == "retry":
        print(f"\n[Wiederholung {event['attempt']}, warten {event['wait']}s...]")
    elif event["type"] == "done":
        print(f"\n\n--- Fertig ({event['chunks']} Blöcke, {event['finish_reason']})")
    elif event["type"] == "error":
        print(f"\n[Fehler: {event['message']}]")

Asynchrones Streaming mit asyncio

Der synchrone Client blockiert seinen Thread für die gesamte Generierung. Auf einem Web-Server, der fünfzig gleichzeitige Gespräche bearbeiten muss, bedeutet dies, fünfzig Threads zu immobilisieren, nur um zu warten. Der asynchrone Client hebt diese Einschränkung auf:

from mistralai.client import Mistral
import asyncio
import os

client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))

async def async_stream_chat(messages: list) -> str:
    """Asynchrones Streaming für konforme Anwendungen."""
    stream = await client.chat.stream_async(
        model="mistral-large-latest",
        messages=messages
    )

    full_text = ""
    async for chunk in stream:
        content = chunk.data.choices[0].delta.content
        if content:
            full_text += content
            print(content, end="", flush=True)

    print()
    return full_text

# Ausführung
async def main():
    messages = [{"role": "user", "content": "Hallo, stelle dich vor."}]
    result = await async_stream_chat(messages)
    print(f"\nGesamt: {len(result)} Zeichen")

asyncio.run(main())

Die Logik ist identisch mit der synchronen Version; nur stream_async, das await beim Aufruf und das async for im Loop ändern sich.

Integration mit FastAPI

Als Nächstes müssen Sie diesen Stream aus Ihrem Python-Prozess in den Browser bekommen. FastAPI übernimmt dies mit StreamingResponse, sofern Sie jeden Fragment neu formatieren, um das SSE-Format zu entsprechen, das in der vorherigen Lektion gesehen wurde:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from mistralai.client import Mistral
import os
import json

app = FastAPI()
client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))

async def generate_stream(prompt: str):
    """SSE-Generator für FastAPI."""
    stream = await client.chat.stream_async(
        model="mistral-large-latest",
        messages=[{"role": "user", "content": prompt}]
    )

    async for chunk in stream:
        content = chunk.data.choices[0].delta.content
        if content:
            data = json.dumps({"content": content})
            yield f"data: {data}\n\n"

    yield "data: [DONE]\n\n"

@app.get("/chat/stream")
async def chat_stream(prompt: str):
    return StreamingResponse(
        generate_stream(prompt),
        media_type="text/event-stream"
    )

Zwei Details bestimmen, ob dieser Endpunkt korrekt funktioniert. Das doppelte Newline \n\n, das jedes yield beendet, ist das, was Ereignisse im SSE-Protokoll trennt: Ohne es wartet der Browser unendlich auf ein Ereignis, das er als unvollständig betrachtet. Und media_type="text/event-stream" ist das, was es dem JavaScript-Client ermöglicht, die Antwort mit einem EventSource statt mit einem Standard-fetch zu konsumieren.

Leistung messen

Sobald der Stream in place ist, werden Sie wissen wollen, wie er performt. Zwei Metriken reichen aus, um Optimierungen voranzutreiben:

import time

start = time.perf_counter()
first_token_time = None
token_count = 0

stream = client.chat.stream(
    model="mistral-large-latest",
    messages=[{"role": "user", "content": "Schreiben Sie einen Absatz über Paris."}]
)

for chunk in stream:
    if chunk.data.choices[0].delta.content:
        if first_token_time is None:
            first_token_time = time.perf_counter() - start
        token_count += 1

total_time = time.perf_counter() - start

print(f"Time to First Token (TTFT): {first_token_time:.3f}s")
print(f"Generierte Tokens: {token_count}")
print(f"Gesamtzeit: {total_time:.3f}s")
print(f"Tokens/Sekunde: {token_count / total_time:.1f}")

TTFT (Time to First Token) – die Zeit zwischen dem Senden der Anfrage und der Anzeige des ersten Zeichens – ist die wichtigste Metrik für das Benutzererlebnis. Sie hat Vorrang vor der Gesamtzeit: Eine Generierung von fünfzehn Sekunden, deren erstes Wort in drei Zehntelsekunden erscheint, wird sich schnell anfühlen, während eine fünfsekündige Generierung, die in einem Block geliefert wird, sich langsam anfühlt. Messen Sie sie an Ihren echten Prompts, einschließlich der Systemnachricht – eine Systemnachmittellänge von zwei Seiten verlängert die TTFT, bevor überhaupt das erste Token generiert wird.

Kernaussagen

• Verwenden Sie ein Generator-Muster (yield), um Streaming-Tokens zu propagieren
• Implementieren Sie exponentiellen Backoff für 429-Fehler (Rate Limit)
• Bevorzugen Sie stream_async für konforme Web-Anwendungen
• Messen Sie TTFT und Tokens/Sekunde, um die Leistung zu optimieren
• Mit FastAPI propagiert StreamingResponse + text/event-stream SSE an den Browser