Stream LLM responses

Beginner → Intermediate · 12 min · tested with openai 3.x, fastapi and streamlit

A 300-word answer can take 8 seconds to generate. Without streaming, the user stares at a spinner for 8 seconds. With streaming, the first words appear in under a second and the rest arrive as they're written. The total time is the same, but it feels many times faster. That's why every chat product streams.

---

1. Stream in a script

Add stream=True. Instead of one response you get an iterator of small chunks, each carrying a few characters:

from openai import OpenAI

client = OpenAI()      # reads OPENAI_API_KEY (see "Manage API keys and settings")

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "What does retrieval do in RAG? One sentence."}],
    stream=True,
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:     # some chunks carry no text
        print(chunk.choices[0].delta.content, end="", flush=True)
print()
# → Retrieval finds the right context.     (appears a few words at a time)

• delta (not message) holds the new text in this chunk.
• end="" stops print adding newlines between pieces; flush=True shows each piece immediately.
• The first and last chunks often have no text (the role, then the finish reason), hence the if.

---

2. Keep the full text and the token count

You usually need the whole answer afterwards too (to save the chat history), and the token usage for cost tracking. Wrap streaming in a generator that yields pieces, and ask for usage in the final chunk:

def stream_answer(question: str, usage: dict):
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
        stream=True,
        stream_options={"include_usage": True},      # adds one last chunk with token counts
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content
        elif chunk.usage:
            usage.update(prompt=chunk.usage.prompt_tokens, completion=chunk.usage.completion_tokens)

usage = {}
parts = []
for piece in stream_answer("What does retrieval do in RAG?", usage):
    print(piece, end="", flush=True)
    parts.append(piece)

full_answer = "".join(parts)
print("\n", usage)
# → Retrieval finds the right context.
# →  {'prompt': 12, 'completion': 5}

This generator is the building block for everything below: FastAPI and Streamlit both accept generators.

Anthropic (Claude) version

import anthropic

claude = anthropic.Anthropic()      # reads ANTHROPIC_API_KEY

with claude.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What does retrieval do in RAG? One sentence."}],
) as stream:
    for text in stream.text_stream:            # just the text pieces
        print(text, end="", flush=True)
    final = stream.get_final_message()          # the complete message, with usage
print("\n", final.usage.input_tokens, final.usage.output_tokens)

---

3. A streaming FastAPI endpoint

Return the generator in a StreamingResponse, and FastAPI sends each piece as soon as it's yielded:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

def text_pieces(question: str):
    stream = client.chat.completions.create(
        model="gpt-4o-mini", stream=True,
        messages=[{"role": "user", "content": question}],
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

@app.get("/chat")
def chat(q: str):
    return StreamingResponse(text_pieces(q), media_type="text/plain; charset=utf-8")

uvicorn main:app --reload
curl -N "http://localhost:8000/chat?q=What+is+RAG"      # -N: print as it arrives, don't buffer

Use a plain def generator with the sync client (FastAPI runs it in a thread), or an async def generator with AsyncOpenAI and async for. Don't call the sync client inside an async def: it blocks the whole server.

Server-Sent Events for browsers

Plain text is fine for curl. For a web frontend, send Server-Sent Events (SSE): each message is a data: ... line followed by a blank line, so the browser knows where pieces start and end, and you can send JSON:

import json

@app.get("/chat/sse")
def chat_sse(q: str):
    def events():
        for piece in text_pieces(q):
            yield f"data: {json.dumps({'text': piece})}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(events(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

What the client receives:

data: {"text": "Retrieval "}

data: {"text": "finds "}

...
data: [DONE]

Reading it from Python (another service, or a test):

import httpx

with httpx.stream("GET", "http://localhost:8000/chat/sse", params={"q": "What is RAG?"}, timeout=60) as r:
    for line in r.iter_lines():
        if line.startswith("data: ") and line != "data: [DONE]":
            print(json.loads(line[6:])["text"], end="", flush=True)

---

4. Streamlit: one line

st.write_stream takes a generator, shows the text as it arrives, and returns the full string:

app.py

import streamlit as st

if question := st.chat_input("Ask about RAG"):
    st.chat_message("user").write(question)
    with st.chat_message("assistant"):
        answer = st.write_stream(text_pieces(question))     # displays live, returns the full text
    st.session_state.setdefault("history", []).append((question, answer))

It even accepts the raw OpenAI stream object directly: st.write_stream(stream). See Chat apps for a full chat UI with history.

---

5. Gotchas

Problem	Cause	Fix
Text arrives all at once at the end	A proxy (nginx, some PaaS load balancers) buffers the response	Send X-Accel-Buffering: no; disable proxy buffering; curl -N to test
Whole server freezes while streaming	Sync client used inside async def	Use AsyncOpenAI with async for, or a plain def endpoint
No token usage when streaming	Usage isn't sent by default	stream_options={"include_usage": True}
Error halfway through an answer	The connection dropped or the provider failed mid-stream	Wrap the loop in try/except; tell the user the answer was cut off and offer a retry
Can't validate the answer before showing it	That's the trade-off of streaming	For JSON or guardrailed output, don't stream: wait, validate, then show
Request times out on long answers	Default HTTP timeouts in proxies	Raise read timeouts; streaming keeps the connection busy, which helps

---

Checklist

• The stream is wrapped in one generator that the CLI, API and UI all reuse.
• The full text is assembled for chat history; token usage is recorded.
• Async endpoints use the async client.
• Proxy buffering is turned off in production, tested with curl -N.
• Mid-stream errors are caught and shown to the user.

Next: Get structured JSON from an LLM →