Optimierter Streaming-Output
Auf dieser Seite
Was Streaming tatsächlich verändert
Streaming macht das Modell nicht schneller: es macht das Warten erträglich. Bei einer Antwort, die fünf Sekunden zur Generierung benötigen, starrt ein Nutzer ohne Streaming fünf Sekunden lang auf einen leeren Bildschirm, bevor ein Textblock erscheint. Mit Streaming sehen die Nutzer die ersten Wörter nach wenigen hundert Millisekunden und lesen mit, während das Modell schreibt. Die Gesamtdauer bleibt identisch, die Wahrnehmung ist jedoch völlig anders. Deshalb ist das Überspringen von Streaming in der Produktion fast immer ein Designfehler und keine bewusste Entscheidung.
Der Stream muss jedoch durch Ihre gesamte Kette fließen. Ein Stream, der zwar korrekt von der API generiert, aber von Ihrem Anwendungsserver gesammelt oder durch Ihren Reverse-Proxy gepuffert wird, erreicht den Nutzer als einzelner Block: Sie zahlen den Komplexitätspreis für Streaming, ohne die Vorteile zu ernten. Diese Lektion folgt daher dem kompletten Pfad, von der API-Aufruf bis zum Browser.
Verbrauch des Streams auf der Client-Seite
Mit der Responses API verwandelt das Setzen von stream=True die Antwort in eine Sequenz von typisierten Ereignissen, die Sie beim Eintreffen iterieren. Jedes Textfragment kommt in einem response.output_text.delta-Ereignis an; das Ende des Streams wird durch response.completed signalisiert, das das vollständige Antwortobjekt enthält und nützlich ist, um den Tokenverbrauch abzurufen.
import openai
client = openai.OpenAI()
def generate_streaming(prompt: str):
"""Streaming-Generierung mit der Responses API."""
stream = client.responses.create(
model="gpt-5.6-terra",
input=prompt,
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True)
elif event.type == "response.completed":
print() # Neue Zeile am Ende
return event.responseDie Flagge flush=True ist kein kosmetisches Detail: Ohne sie puffert Python den Output selbst, und Sie landen im Block-Verhalten, das Sie vermeiden wollten. Dieser Detailaspekt wiederholt sich auf jeder Ebene der Kette und ist die Haupterschwerung beim Streaming.
Der Mechanismus deckt auch Funktionsaufrufe ab. Tool-Argumente werden schrittweise aufgebaut, und Sie können sie anzeigen, während sie generiert werden, um dem Nutzer zu signalisieren, dass eine Suche formuliert wird, statt ihn auf eine eingefrorene Schnittstelle zu lassen, während das Modell seinen Aufruf vorbereitet.
tools = [
{
"type": "function",
"name": "search_product",
"description": "Sucht nach einem Produkt im Katalog.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"category": {"type": "string"},
},
"required": ["query"],
},
}
]
def streaming_with_tools(prompt: str):
stream = client.responses.create(
model="gpt-5.6-terra",
input=prompt,
tools=tools,
stream=True,
)
for event in stream:
match event.type:
case "response.output_text.delta":
print(event.delta, end="", flush=True)
case "response.function_call_arguments.delta":
# Aktuell generierte Funktionsargumente
print(f"[arg: {event.delta}]", end="")
case "response.function_call_arguments.done":
print(f"\n[Funktionsaufruf abgeschlossen]")Weiterleitung des Streams an den Browser
Ihr Frontend kommuniziert nicht direkt mit der OpenAI API, sondern mit Ihrem Server, der den Stream daher weiterleiten muss. Server-Sent Events (SSE) sind das natürliche Format hierfür: Eine HTTP-Antwort, die offen bleibt, wobei jede data:-Zeile einen JSON-Fragment enthält. Das folgende Beispiel leitet Text weiter, sobald er ankommt, und sendet abschließend eine Nachricht mit der Token-Anzahl, sodass Sie abrechnen oder protokollieren können, ohne einen zweiten Aufruf durchführen zu müssen.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import openai
import json
app = FastAPI()
client = openai.OpenAI()
@app.post("/api/chat")
async def chat_streaming(request: dict):
prompt = request["message"]
async def generate():
stream = client.responses.create(
model="gpt-5.6-terra",
input=prompt,
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
data = json.dumps({"text": event.delta})
yield f"data: {data}\n\n"
elif event.type == "response.completed":
usage = event.response.usage
data = json.dumps({
"end": True,
"input_tokens": usage.input_tokens,
"output_tokens": usage.output_tokens,
})
yield f"data: {data}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no", # Wichtig für nginx
},
)Dann gibt es noch den Reverse-Proxy, den üblichen Verdächtigen, wenn „Streaming in der Produktion nicht funktioniert, obwohl es lokal geklappt hat”. Standardmäßig sammelt nginx die Backend-Antwort, bevor sie sie übermittelt. Die folgende Konfiguration deaktiviert dieses Verhalten nur für die spezifische Route.
location /api/chat {
proxy_pass http://127.0.0.1:8000;
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection "";
chunked_transfer_encoding on;
}Aufrechterhaltung der Verbindung über die Zeit
Ein Stream, der über mehrere Zehntelsekunden offen bleibt, ist einem kurzen Aufruf stärker ausgesetzt: Ein Netzwerkausfall, ein Kontingent-Limit und die Generierung mitten im Satz stoppen. Ihre Retry-Strategie muss zwischen diesen beiden Fällen unterscheiden, da sie unterschiedliche Ansätze erfordern. Ein Verbindungsfehler sollte mit exponentiellem Backoff wiederholt werden, um dem Netzwerk Zeit zur Erholung zu geben, ohne die API zu überlasten; eine Überschreitung des Kontingents sollte nach einer festen Wartezeit wiederholt werden, da sich das Rate-Limit-Fenster in regelmäßigen Abständen zurücksetzt.
import openai
def resilient_streaming(prompt: str, max_attempts: int = 3):
"""Streaming mit Fehlerbehandlung und Retry."""
for attempt in range(max_attempts):
try:
stream = client.responses.create(
model="gpt-5.6-terra",
input=prompt,
stream=True,
)
full_text = ""
for event in stream:
if event.type == "response.output_text.delta":
full_text += event.delta
yield event.delta
return # Erfolg, beenden
except openai.APIConnectionError:
if attempt < max_attempts - 1:
import time
time.sleep(2 ** attempt) # Exponentielles Backoff
else:
raise
except openai.RateLimitError:
import time
time.sleep(5) # Feste Wartezeit für Rate LimitEine letzte Anpassung verbessert die Lesekomfort. Das Ausgeben jedes Tokens erzeugt ruckelnden Text, insbesondere wenn Ihre Schnittstelle eine Animation anwendet oder das Ergebnis in Text-zu-Sprache umgewandelt wird. Das Akkumulieren bis zum Ende eines Satzes ergibt einen Rhythmus, der dem menschlichen Lesen näher kommt, auf Kosten einer geringfügigen zusätzlichen Verzögerung beim ersten Fragment.
def streaming_by_sentence(prompt: str):
"""Akkumuliert und gibt vollständige Sätze aus."""
buffer = ""
stream = client.responses.create(
model="gpt-5.6-terra",
input=prompt,
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
buffer += event.delta
# Suche nach Satzende
for separator in [".\n", "!\n", "?\n", ". ", "! ", "? "]:
if separator in buffer:
parts = buffer.split(separator, 1)
yield parts[0] + separator.rstrip()
buffer = parts[1] if len(parts) > 1 else ""
if buffer.strip():
yield buffer.strip()Beachten Sie das finale if buffer.strip(): Ohne diesen würde eine Antwort, die nicht mit starkem Zeichensetzung endet, ihre letzten Wörter verlieren. Solche Übersehen werden im Test nicht bemerkt und manifestieren sich in der Produktion bei durch max_output_tokens abgeschnittenen Antworten.
Kernaussagen
- Nutzen Sie in der Produktion immer Streaming, um die wahrgenommene Latenz zu reduzieren.
- Konfigurieren Sie nginx mit
proxy_buffering offfür SSE. - Behandeln Sie Fehler mit Retry und exponentiellem Backoff.
- Akkumulieren Sie nach Sätzen für eine natürlichere Darstellung.