Streaming-Ausgabe

Ein Nutzer stellt eine Frage, auf dem Bildschirm passiert nichts, drei Sekunden später erscheint schlagartig die ganze Antwort. Erscheint bei denselben drei Sekunden das erste Zeichen nach einer halben Sekunde und dann eins nach dem anderen, fühlt sich das viel besser an: Er weiß, dass das Programm arbeitet, und kann schon beim Warten lesen.

Das ist Streaming-Ausgabe. Wie Modul 01, Lektion 2 gezeigt hat, erzeugt das Modell ohnehin Token für Token. Streaming schickt dir jedes kleine Stück sofort, statt alles zu sammeln und dann zusammen zu senden.

Streaming einschalten

Füg der Anfrage stream=True hinzu, dann bekommst du keine vollständige Antwort mehr, sondern einen Datenstrom, den man mit einer for-Schleife durchlaufen kann:

stream = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "用大约 200 字介绍 httpx 和 requests 的主要区别。"}],
    stream=True,
    extra_body={"thinking": {"type": "disabled"}},
)
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

Jedes chunk ist ein kleines Datenstück; der neu erzeugte Text steht in chunk.choices[0].delta.content. delta heißt „Zuwachs“: Es enthält nur den in diesem Stück neu hinzugekommenen Inhalt, nicht alles bisher. Zusammensetzen musst du selbst.

end="" bei print hängt die Stücke ohne Zeilenumbruch aneinander, flush=True zeigt sie sofort an, statt zu warten, bis der Puffer voll ist. Ohne flush=True erscheint der Text in Schüben, und der Streaming-Effekt ist weg.

Gemessen: wie viel schneller

code/03-llm-apps/streaming.py stellt dieselbe Frage einmal mit und einmal ohne Streaming und misst die Zeit bis zum ersten Zeichen und bis zum Ende, jeweils mit und ohne Denken:

def streaming(thinking, show=False):
    start = time.time()
    first_content = None
    stream = client.chat.completions.create(
        model=MODEL, messages=QUESTION, stream=True,
        stream_options={"include_usage": True},  # 让最后一个数据块带上 usage
        extra_body={"thinking": {"type": "enabled" if thinking else "disabled"}},
    )
    usage = None
    for chunk in stream:
        if chunk.usage:
            usage = chunk.usage
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta
        if delta.content:
            if first_content is None:
                first_content = time.time() - start
            if show:
                print(delta.content, end="", flush=True)
    if show:
        print()
    return first_content, time.time() - start, usage.completion_tokens

Mein Ergebnis (zuerst wurde einmal die Streaming-Ausgabe gezeigt, hier ist nur die Zeitmessung übrig):

不思考 非流式：第一个字 1.82 秒，全部完成 1.82 秒，输出 153 词元
不思考 流式：  第一个字 0.67 秒，全部完成 1.71 秒，输出 202 词元
开思考 非流式：第一个字 2.48 秒，全部完成 2.48 秒，输出 330 词元
开思考 流式：  第一个字 1.82 秒，全部完成 2.75 秒，输出 379 词元

Ohne Denken erschien das erste Zeichen beim Streaming nach 0,67 Sekunden, ohne Streaming war erst nach 1,82 Sekunden überhaupt etwas zu sehen.

Achte auf die Zeit bis „alles fertig“: Beide Varianten liegen nah beieinander (1,71 gegen 1,82 Sekunden, und die zwei Antworten waren auch unterschiedlich lang). Streaming lässt das Modell nicht schneller erzeugen; die Gesamtzeit bleibt gleich, es ändert nur, wann der Nutzer die ersten Inhalte sieht. Je länger die Antwort, desto deutlicher der Unterschied: Bei einer Antwort, die 20 Sekunden zum Erzeugen braucht, starrt der Nutzer ohne Streaming 20 Sekunden auf einen leeren Bildschirm.

Mit Denken erschien das erste Zeichen auch beim Streaming erst nach 1,82 Sekunden, weil das Modell zuerst denkt und erst danach die eigentliche Antwort schreibt. Der Denkinhalt wird ebenfalls gestreamt, in delta.reasoning_content; willst du dem Nutzer „denkt nach …“ zeigen, kannst du ihn anzeigen oder nur einen Hinweis einblenden.

Wie man beim Streaming an usage kommt

Beim Aufruf ohne Streaming sagt dir response.usage direkt, wie viele Tokens verbraucht wurden. Beim Streaming gibt es diese Information standardmäßig nicht.

Mit stream_options={"include_usage": True} schickt der Server am Ende des Stroms ein zusätzliches Datenstück mit usage, dessen choices aber eine leere Liste ist. Deshalb steht im Code oben if not chunk.choices: continue; ohne diese Zeile gibt der Zugriff auf chunk.choices[0] einen IndexError.

Worauf man beim Streaming achten muss

finish_reason steht im letzten Stück. Beim Streaming ist finish_reason in allen früheren Stücken None, erst das letzte Stück mit Inhalt hat Werte wie stop oder length. Um zu prüfen, ob die Antwort abgeschnitten wurde, merkst du es dir in der Schleife.

Fehler können mittendrin passieren. Ein Aufruf ohne Streaming klappt oder scheitert. Ein gestreamter Aufruf kann abreißen, nachdem schon die Hälfte ausgegeben wurde, und der Nutzer hat die halbe Antwort schon gesehen. Ein neuer Versuch erzeugt dann von vorn eine neue Antwort, die womöglich nicht zur schon gesehenen ersten Hälfte passt. Üblich ist: Fehler beim Verbindungsaufbau darf man wiederholen; Fehler nach Beginn der Ausgabe meldet man dem Nutzer als „Antwort abgebrochen“ und lässt ihn entscheiden, ob er neu fragt. RepoBot in Lektion 5 macht es so.

Die vollständige Antwort muss man selbst zusammensetzen. In mehrstufigen Gesprächen wird die Antwort des Modells als assistant-Nachricht im Verlauf gespeichert. Beim Streaming gibt es keine fertige vollständige Antwort; man muss alle delta.content aneinanderhängen.

Streaming an den Browser senden

Auf der Kommandozeile genügt print. Als Webseite muss die Ausgabe des Modells über dein Backend in Echtzeit an den Browser weitergeleitet werden. Am gängigsten ist SSE (Server-Sent Events): ein vom Browser nativ unterstützter Weg, bei dem der Server fortlaufend Nachrichten an den Browser schickt. Das Format ist sehr einfach: Jede Nachricht ist eine Zeile data: Inhalt, gefolgt von einer Leerzeile.

Ein minimales Beispiel mit FastAPI (code/03-llm-apps/streaming_web.py):

import json
import os

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse
from openai import AsyncOpenAI

client = AsyncOpenAI(  # 网页服务要同时应付很多请求，用异步客户端
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
)
MODEL = os.environ.get("LLM_MODEL", "deepseek-flash")
app = FastAPI()


@app.get("/chat")
async def chat(q: str):
    async def events():
        stream = await client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "user", "content": q}],
            stream=True,
            extra_body={"thinking": {"type": "disabled"}},
        )
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                # SSE 的格式：每条消息以 "data: " 开头，以空行结尾
                yield f"data: {json.dumps(chunk.choices[0].delta.content, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

Ein paar Punkte:

• AsyncOpenAI statt OpenAI. Ein Webdienst bedient viele Nutzer gleichzeitig; ein synchroner Client blockiert den ganzen Dienst, während er auf das Modell wartet, ein asynchroner kann in der Wartezeit andere Anfragen bearbeiten.
• StreamingResponse nimmt einen Generator; jedes yield schickt ein Stück Daten an den Browser.
• Jedes Stück wird mit json.dumps kodiert. Die Ausgabe des Modells kann Zeilenumbrüche enthalten, und SSE trennt Nachrichten durch Zeilenumbrüche; direkt eingefügt würde das Format zerbrechen. Als JSON-String kodiert, gibt es das Problem nicht.
• Zum Schluss ein [DONE], das dem Browser das Ende meldet.

Auf Browserseite empfängt man mit EventSource:

const source = new EventSource("/chat?q=" + encodeURIComponent(question));
source.onmessage = (e) => {
  if (e.data === "[DONE]") { source.close(); return; }
  out.textContent += JSON.parse(e.data);
};

Der vollständige Seitencode steht in streaming_web.py. Abhängigkeiten installieren und starten:

uv add fastapi uvicorn
uvicorn streaming_web:app --port 8000

Im Browser http://127.0.0.1:8000 öffnen und ausprobieren. Man kann die rohen SSE-Daten auch mit curl ansehen; die Option -N lässt curl jedes empfangene Stück sofort anzeigen:

curl -N "http://127.0.0.1:8000/chat?q=用一句话介绍httpx"

Die ersten Nachrichten bei meinem Lauf:

data: "HTTP"

data: "X"

data: " "

data: "是一个"

data: "功能"

Jede Nachricht ist ein oder zwei Tokens. Der Browser hängt jede empfangene Nachricht an die Seite an.

EventSource kann nur GET-Anfragen senden; die Frage muss in die URL, deren Länge begrenzt ist, und der Gesprächsverlauf lässt sich schlecht mitschicken. In echten Projekten sendet man meist per fetch eine POST-Anfrage und liest dann den zurückkommenden Datenstrom. Beim Deployment von RepoBot in Modul 06, Lektion 6 wird es so gemacht.

Wann man kein Streaming nutzt

• Das Ergebnis wird von einem Programm verarbeitet. Etwa bei der JSON-Extraktion aus Modul 02, Lektion 4: Das Programm braucht das vollständige JSON zum Parsen, Streaming bringt nichts.
• Batch-Aufgaben im Hintergrund. Niemand wartet vor dem Bildschirm; Streaming macht den Code nur komplizierter.

Überall, wo jemand vor dem Bildschirm auf eine Antwort wartet, sollte man streamen.

Übungen

1. Ändere in streaming.py die Frage in „Schreib einen Artikel mit 800 Zeichen über httpx“ und vergleiche erneut die Zeit bis zum ersten Zeichen mit und ohne Streaming. Ist der Abstand größer geworden?
2. Ändere die Funktion streaming so, dass sie bei eingeschaltetem Denken auch delta.reasoning_content in Grau oder anders markiert ausgibt, damit der Nutzer sieht, „woran das Modell denkt“.
3. Erweitere streaming_web.py: Am Ende des Stroms eine zusätzliche Nachricht senden, die dem Browser mitteilt, wie viele Tokens verbraucht wurden (denk an stream_options).

Selbsttest

1. Erzeugt das Modell mit Streaming eine vollständige Antwort schneller?

Nein. Die Gesamtzeit für die vollständige Antwort bleibt im Wesentlichen gleich. Streaming ändert, wann der Nutzer das erste Zeichen sieht: Der Inhalt wird beim Erzeugen angezeigt, und der Nutzer muss nicht warten, bis alles fertig ist.

2. Beim Streaming mit stream_options={"include_usage": True} wirft das Programm in der Zeile chunk.choices[0] einen IndexError. Warum?

Mit include_usage schickt der Server am Ende des Stroms ein Datenstück, das nur usage enthält und dessen choices eine leere Liste ist. Vor dem Zugriff auf choices[0] muss man prüfen, ob chunk.choices leer ist.

3. Warum sollte ein Web-Backend AsyncOpenAI statt OpenAI verwenden?

Ein synchroner Client blockiert, während er auf die Antwort des Modells wartet, und in dieser Zeit kann der Server keine Anfragen anderer Nutzer bearbeiten. Ein asynchroner Client kann in der Wartezeit zu anderen Anfragen wechseln, sodass ein Prozess viele Nutzer gleichzeitig bedienen kann.