Streaming responses

What this session covers

The first-call chapter waited for the whole answer before printing anything. That works for one sentence. For a paragraph it feels broken — the reader sits there watching nothing happen, then sees a blob. Streaming swaps that for a typewriter. The model ships tokens as it generates them, the terminal prints them as they arrive. Same model, same prompt, same total latency. What changes is how it feels, and that's most of what users notice.

OpenAI

One keyword argument. stream=True flips the response from "give me the final blob" to "give me a sequence of typed events". Iterating yields several event kinds — created, in-progress, output-text deltas, done, completed. Only the deltas carry text; the rest is metadata I ignore. Each delta gets printed with end="" and flush=True so the terminal types it out instead of buffering a whole line.

"""Week 3 - Streaming responses (OpenAI).

Same one API call as week 1, with stream=True. The SDK yields typed
events as tokens come back from the server; we filter for the
output-text delta events and print each delta as it arrives. The
terminal shows the answer take shape instead of waiting for the full
blob.
"""
import os
import sys

from dotenv import find_dotenv, load_dotenv
from openai import OpenAI

load_dotenv(find_dotenv())

if not os.environ.get("OPENAI_API_KEY"):
    sys.exit("OPENAI_API_KEY is not set. Put it in the course-root .env file.")

client = OpenAI()

# stream=True turns the call into a server-sent-event stream. Iterating
# the response yields one event per server-side delta; we only want the
# text deltas, so we filter by event.type.
stream = client.responses.create(
    model="gpt-5.4-nano",
    input="In one short paragraph, explain why streaming responses matter for chat UX.",
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        # end="" + flush=True is what turns this into a typewriter.
        # Without flush the terminal buffers a whole line at a time.
        print(event.delta, end="", flush=True)
print()

A few things to watch. The event types are stable across the Responses API, so the filter on response.output_text.delta holds across models. The deltas are server-side chunks, not single tokens — sometimes a word, sometimes a sentence. That's fine; the lesson is "don't wait", not "do something with each token". And flush=True is non-negotiable. Python buffers stdout by default. Without flush, the typewriter falls apart.

Gemini

Gemini picked the other ergonomic bet. Streaming gets its own method — generate_content_stream, the streaming twin of generate_content. The return type is plainer: an iterator of chunks, each chunk carrying a .text attribute that's either a string delta or empty.

"""Week 3 - Streaming responses (Gemini).

The Gemini SDK exposes streaming as a separate method
(`generate_content_stream`) that returns an iterator of chunks. Each
chunk carries a partial text delta. Same teaching point as the OpenAI
example: print as you go, no buffering.
"""
import os
import sys

from dotenv import find_dotenv, load_dotenv
from google import genai

load_dotenv(find_dotenv())

if not os.environ.get("GEMINI_API_KEY"):
    sys.exit("GEMINI_API_KEY is not set. Put it in the course-root .env file.")

client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

# generate_content_stream is the streaming twin of generate_content.
# Each iteration yields a partial response carrying a `.text` delta.
stream = client.models.generate_content_stream(
    model="gemini-3.1-flash-lite",
    contents="In one short paragraph, explain why streaming responses matter for chat UX.",
)

for chunk in stream:
    if chunk.text:
        print(chunk.text, end="", flush=True)
print()

Same UX, different SDK shape. OpenAI overloaded one method with a flag; Google split it into two. Both work. Pick whichever reads clearer in your codebase and stay consistent across the project.

Put it to work

Three showcases, three docker-compose projects. Each one is a browser tab with a form. bash bootstrap-secrets.sh, then docker compose up --build, then http://localhost:3000. The book's sidebar lists them as cards. Each card opens a landing page with the source code, a try-it widget you can hit live, and a one-command zip download.

Showcase 1 — Live explainer

Ask a question, watch the answer take shape. Same one-API-call shape as week 1, pointed at a Q&A prompt, with stream=True doing the work. The point is the feel: waiting for one blob vs. watching the response build. Same total latency. Only the experience changes.

Showcase 2 — Stream translate

Paste text, pick a target language, watch the translation arrive sentence by sentence. Translation is where streaming earns its keep — long-ish output, no value in batching, and you can start reading the first sentence while the model is still on the third. The input carries the target language in a small target: <language> header so the def run(input) contract stays one string.

Showcase 3 — Live rewrite

Paste any draft message. A tighter version streams in. Same facts, fewer words. This one is the closest to a daily-use tool — the streaming UI makes the rewrite feel like editing alongside someone, which is the experience you want in a writing assistant.

Across all three, the design that keeps the code honest is the two-callable shape. run_stream yields tokens as they arrive — that is the lesson. run is a thin accumulator over run_stream that returns the final string. The local frontend hits /api/ai/stream for the typewriter; the website's try-it widget hits /api/ai for the accumulated text. One streaming path, two surfaces, no duplicated logic.

Run it

README.md has the prereqs, the basic-CLI commands, and the three per-showcase run blocks. Short version: python code/openai/main.py for the simplest demo, or cd into any showcase, bash bootstrap-secrets.sh, and docker compose up --build.

Takeaways

Streaming is one keyword argument away from a better-feeling app, and the model never got faster. That trick — changing the experience of waiting without changing the actual latency — is one of the cheaper wins in AI work. Worth doing. Keep the provider logic in a run_stream generator so you control the loop, not the SDK. Accumulate the stream into a string when you need one at the edge. Don't reach for streaming on one-line answers; the overhead isn't worth it. Reach for it when the output is long enough that the reader would tab away otherwise.