Streaming-Antworten in Python

Das Problem, das Streaming löst

Mit chat.complete() warten Sie darauf, dass das Modell die gesamte Antwort generiert hat, bevor Sie auch nur einen einzigen Zeichen erhalten. Bei einer kurzen Antwort bleibt diese Verzögerung unbemerkt. Bei einer detaillierten Analyse, einem langen Codeausschnitt oder einer mehrstufigen Argumentation starrt der Nutzer jedoch mehrere Sekunden lang auf einen leeren Bildschirm. In einem Chat-Interface kann selbst eine kurze Stille von wenigen Sekunden dazu führen, dass die Anwendung eingefroren wirkt.

Streaming ändert diese Wahrnehmung: Die Antwort kommt inkrementell, Chunk für Chunk, und wird schrittweise angezeigt, genau wie in den Chat-Schnittstellen, die Sie kennen. Die gesamte Generierungszeit sinkt nicht um einen Millisekunde, aber das erste Wort erscheint fast sofort. Diese anfängliche Verzögerung ist es, die der Nutzer tatsächlich spürt.

Verwendung von chat.stream()

Die Mistral SDK V2 stellt die Methode chat.stream() bereit, die einen Iterator von Chunks zurückgibt, anstatt eines einzelnen Objekts:

import os
from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

stream = client.chat.stream(
    model="mistral-small-latest",
    messages=[
        {
            "role": "user",
            "content": "Beschreiben Sie die Schritte zum Bereitstellen einer Flask-Anwendung auf einem Linux-Server."
        }
    ]
)

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

# Endzeilenzeichen
print()

Zwei Details machen in diesem Loop den entscheidenden Unterschied. Die Prüfung if content ist unerlässlich, da einige Chunks ohne Text eintreffen und sonst None ausgeben würden. Und flush=True zwingt Python, sofort in die Ausgabe zu schreiben, anstatt darauf zu warten, dass sich der Puffer füllt. Ohne diese Option würde der gesamte Text in einem Block am Ende erscheinen, was den gewünschten Nutzen vollständig zunichtemacht.

Was ein Chunk enthält

Jeder Chunk trägt ein Delta, das bedeutet einen Fragment der im Aufbau befindlichen Antwort. Durch die Inspektion des Chunks haben Sie Zugriff sowohl auf den Text als auch auf die Metadaten am Ende des Streams:

for chunk in stream:
    delta = chunk.data.choices[0].delta

    # Textinhalt (kann für einige Chunks None sein)
    text = delta.content

    # Stop-Grund (außer beim letzten Chunk None)
    end = chunk.data.choices[0].finish_reason

    if text:
        print(text, end="", flush=True)

    if end:
        print(f"\n[Stream-Ende: {end}]")

Der Ablauf folgt immer derselben Choreografie: Der erste Chunk trägt in der Regel die Rolle assistant, mittlere Chunks liefern den Text in kleinen Stücken, und der letzte liefert den finish_reason. Sie bestimmen, ob die Antwort normal beendet wurde oder durch max_tokens abgeschnitten wurde, indem Sie diesen letzten Chunk prüfen.

Gleichzeitiges Anzeigen und Speichern

Streaming hat einen Nachteil: Der Text läuft ab, aber Sie halten ihn nirgendwo fest. Eine reale Anwendung benötigt jedoch oft, um die Antwort in einer Datenbank für die Gesprächshistorie zu archivieren. Die Lösung besteht darin, die Fragmente während des Anzeigens zu akkumulieren:

import os
from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

stream = client.chat.stream(
    model="mistral-small-latest",
    messages=[
        {"role": "user", "content": "Listen Sie fünf Python-Web-Frameworks auf."}
    ]
)

full_response = []

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

print()

# Die rekonstruierte vollständige Antwort
final_text = "".join(full_response)
print(f"\nGesamtlänge: {len(final_text)} Zeichen")

Beachten Sie die Verwendung einer Liste gefolgt von einem finalen "".join() anstelle von String-Konkatenierung in jeder Iteration des Loops. Bei einer langen Antwort würde der letztere Ansatz den gesamten akkumulierten Text für jedes empfangene Fragment kopieren, was ineffizient ist.

Die asynchrone Variante

In einem Web-Server – FastAPI, aiohttp – blockiert die synchrone Version den Thread während der gesamten Generierung und friert andere laufende Anfragen ein. Das SDK stellt daher chat.stream_async() bereit:

import asyncio
import os
from mistralai.client import Mistral

async def generate_response():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    stream = await client.chat.stream_async(
        model="mistral-small-latest",
        messages=[
            {"role": "user", "content": "Erklären Sie das Konzept einer Coroutine in Python."}
        ]
    )

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

    print()

asyncio.run(generate_response())

Die Struktur ist identisch zur synchronen Version, abgesehen von den Schlüsselwörtern await und async for. Sobald Ihr Code in einem Web-Server ausgeführt wird, ist dies die Variante, die Sie verwenden müssen.

Wann zu streamen und wann darauf zu verzichten

In einem Chat-Interface ist Streaming immer zwingend erforderlich: Die Lesekomfort steht auf dem Spiel. Auf der Backend-API-Seite lohnt es sich, Streaming zu aktivieren, solange Ihr Frontend Server-Sent Events verarbeiten kann. Für jede Antwort, die länger als ein paar Sätze ist, rechtfertigt der Gewinn an wahrgenommener Latenz die zusätzliche Komplexität. Umgekehrt bleibt bei einem Batch-Skript, das tausende Dokumente verarbeitet und Ergebnisse in eine Datei schreibt, chat.complete() einfacher und völlig ausreichend, da niemand den Bildschirm beobachtet.

Kernaussagen

• chat.stream() gibt einen Iterator von Chunks für die schrittweise Anzeige zurück
• Der Inhalt befindet sich in chunk.data.choices[0].delta.content
• Verwenden Sie flush=True in Ihren print()-Aufrufen für die Echtzeit-Anzeige
• Rekonstruieren Sie die vollständige Antwort, indem Sie Chunks akkumulieren, falls erforderlich
• Die asynchrone Version chat.stream_async() wird für Webanwendungen empfohlen