Optimized streaming
On this page
What streaming actually changes
Streaming doesn’t make the model faster: it makes the wait bearable. For a response that takes five seconds to generate, a user without streaming stares at a blank screen for five seconds before receiving a block of text. With streaming, they see the first words after a few hundred milliseconds and read along as the model writes. The total duration is identical, but the perception is completely different. That’s why, in production, skipping streaming is almost always a design flaw rather than a conscious choice.
However, the stream must flow through your entire chain. A stream generated correctly by the API but accumulated by your application server or buffered by your reverse proxy reaches the user in a single block: you’ve paid the complexity price for streaming without reaping the benefits. This lesson therefore follows the complete path, from the API call to the browser.
Consuming the stream on the client side
With the Responses API, setting stream=True transforms the response into a sequence of typed events that you iterate through as they arrive. Each text fragment arrives in a response.output_text.delta event; the end of the stream is signaled by response.completed, which carries the full response object, useful for retrieving token usage.
import openai
client = openai.OpenAI()
def generate_streaming(prompt: str):
"""Streaming generation with the 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() # New line at the end
return event.responseThe flush=True flag isn’t cosmetic: without it, Python buffers the output itself, and you end up with the block behavior you were trying to avoid. This detail repeats at every layer of the chain, and it’s the main difficulty of streaming.
The mechanism also covers function calls. Tool arguments are built progressively, and you can display them as they generate to signal to the user that a search is being formulated, rather than leaving them staring at a frozen interface while the model prepares its call.
tools = [
{
"type": "function",
"name": "search_product",
"description": "Searches for a product in the catalog.",
"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":
# Function arguments currently being generated
print(f"[arg: {event.delta}]", end="")
case "response.function_call_arguments.done":
print(f"\n[Function call complete]")Relaying the stream to the browser
Your frontend doesn’t talk to the OpenAI API directly—it talks to your server, which must therefore relay the stream. Server-Sent Events (SSE) are the natural format for this: an HTTP response that stays open, where each data: line carries a JSON fragment. The following example relays text as it arrives, then emits a final message containing the token count, allowing you to bill or log without a second call.
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", # Important for nginx
},
)Then there’s the reverse proxy, the usual suspect when “streaming doesn’t work in production even though it worked locally.” By default, nginx accumulates the backend response before transmitting it. The configuration below disables this behavior for the specific route only.
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;
}Sustaining the connection over time
A stream open for several tens of seconds is more exposed than a short call: a network drop, a quota hit, and generation stops mid-sentence. Your retry strategy must distinguish between these two cases, because they require different approaches. A connection error should be retried with exponential backoff, giving the network time to recover without hammering the API; a quota exceedance should be retried after a fixed wait, since the rate limit window resets at regular intervals.
import openai
def resilient_streaming(prompt: str, max_attempts: int = 3):
"""Streaming with error handling and 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 # Success, exit
except openai.APIConnectionError:
if attempt < max_attempts - 1:
import time
time.sleep(2 ** attempt) # Exponential backoff
else:
raise
except openai.RateLimitError:
import time
time.sleep(5) # Fixed wait for rate limitOne final tweak improves reading comfort. Emitting every token produces jerky text, especially if your interface applies an animation or if the result goes to text-to-speech. Accumulating up to the end of a sentence gives a rhythm closer to human reading, at the cost of a slight additional delay on the first fragment.
def streaming_by_sentence(prompt: str):
"""Accumulates and yields by complete sentence."""
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
# Look for end of sentence
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()Note the final if buffer.strip(): without it, a response that doesn’t end with strong punctuation would lose its last words. This kind of oversight goes unnoticed in testing and manifests in production on responses truncated by max_output_tokens.
Key takeaways
- Always use streaming in production to reduce perceived latency
- Configure nginx with
proxy_buffering offfor SSE - Handle errors with retry and exponential backoff
- Accumulate by sentence for more natural display