Streaming — Module 18.2

How This Lesson Fits the Module & Volume

Module 18.1 SDKs all expose token streams (OpenAI stream=True, Anthropic text_stream, Gemini generate_content_stream). Users expect that typing UX. This lecture is how your FastAPI forwards those tokens—usually Server-Sent Events (SSE)—without giving the browser a vendor key.

Contrast with Celery (background job) and WebSockets (bidirectional). Streaming keeps one HTTP request open and incremental. Next: authentication on the same routes.

Learning Objectives

By the end of this lesson, students should be able to:

Explain SSE vs chunked JSON vs WebSockets vs sync completion.
Proxy an OpenAI (or Anthropic/Gemini) token stream through FastAPI.
Define a small event schema (delta, usage, error, done).
Handle client disconnect and cancel upstream when possible.
Know proxy buffering pitfalls (X-Accel-Buffering, timeouts).
Meter tokens after the stream ends for monitoring.

Definition

Streaming here means delivering model output incrementally over an open connection instead of one JSON blob. SSE is an HTTP response with Content-Type: text/event-stream and data: lines; the browser EventSource API consumes it (GET historically; POST+fetch readers are common for chat). It is one-way (server → client). Vendor SDK streaming is the upstream source; SSE is the downstream product transport.

Transport Comparison

Transport	Direction	Chat typing	Cancel / barge-in
JSON POST	one shot	Poor (long wait)	Abort request
SSE	server →	Excellent	Abort fetch; limited up-channel
WebSocket	both	Excellent	Natural
Celery + poll	async job	Wrong tool	Task revoke

FastAPI SSE Proxy Sketch

# pip install fastapi uvicorn openai
# Client: fetch("/v1/chat/stream", {method:"POST", body, headers}) + ReadableStream
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()
client = OpenAI()

class StreamIn(BaseModel):
    prompt: str

def sse(obj: dict) -> bytes:
    return f"data: {json.dumps(obj)}\n\n".encode()

@app.post("/v1/chat/stream")
def chat_stream(body: StreamIn):
    def gen():
        try:
            stream = client.chat.completions.create(
                model="gpt-4.1-mini",
                messages=[{"role": "user", "content": body.prompt}],
                stream=True,
                stream_options={"include_usage": True},  # if supported by SKU
            )
            for chunk in stream:
                if chunk.usage:
                    yield sse({"type": "usage", "input": chunk.usage.prompt_tokens,
                               "output": chunk.usage.completion_tokens})
                delta = chunk.choices[0].delta.content if chunk.choices else None
                if delta:
                    yield sse({"type": "delta", "text": delta})
            yield sse({"type": "done"})
        except Exception:
            yield sse({"type": "error", "code": "upstream_llm_failed"})

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # hint nginx not to buffer
        },
    )

# Anthropic/Gemini: same SSE envelope; different upstream iterators (Module 18.1)

Product Event Envelope

Your events

delta text fragments
usage once at end
error / done

Do not leak

Raw OpenAI chunk JSON
Anthropic block internals
Vendor request ids as auth

Ops

Gateway idle timeouts
Disable proxy buffering
Cancel upstream on disconnect

SSE wins

Simple, HTTP-native, cache-bustable
Same auth headers as POST
Enough for most chat UIs

SSE limits

One-way; use WS for barge-in
Some proxies buffer or kill idle
Classic EventSource is GET-only

Related Lectures

Lecture	Role
OpenAI SDK	Upstream `stream=True`
WebSockets	Bidirectional alternative
Celery	Not for token typing
Authentication	Same JWT on stream routes
Monitoring	TTFT + tokens after `done`

Common Misconception

“Streaming means Celery.” Celery is jobs; streaming is an open HTTP/WS connection. Second: printing tokens in Uvicorn logs is not a product stream. Third: buffering nginx will make SSE look like sync JSON. Fourth: you still must auth—a public /stream is a wallet-draining endpoint. Fifth: Anthropic/Gemini chunks are not OpenAI deltas; wrap them in your envelope.

Knowledge Check

Short Answer: What Content-Type does SSE use? Answer: text/event-stream.
True/False: SSE is full-duplex like WebSockets. Answer: False—server to client only.
Multiple Choice: Token typing should use: (a) SSE/WS, (b) only Celery poll, (c) t-SNE. Answer: (a).
Short Answer: Why X-Accel-Buffering: no? Answer: Hint nginx/proxies not to buffer the stream into one blob.
True/False: The browser should call OpenAI with stream=True using the org key. Answer: False—proxy via your API.
Multiple Choice: A good product event type is: (a) delta, (b) DDIM, (c) LoRA rank. Answer: (a).
Short Answer: Name one metric to record at stream end. Answer: Any of: TTFT, total latency, input/output tokens, error rate.
True/False: You should leak raw vendor chunk objects to the SPA. Answer: False—normalize to your envelope.
Multiple Choice: Next lecture: (a) Authentication, (b) PCA, (c) ControlNet. Answer: (a).
Short Answer: When prefer WebSockets over SSE? Answer: Need client→server mid-stream (cancel, barge-in, collab, audio up).

Key Takeaways

Streaming = incremental tokens; SSE is the default product transport.
Proxy vendor streams; emit your delta/usage/done/error envelope.
Disable proxy buffering; watch timeouts; auth the route.
Celery ≠ typing UX; WS when you need uplink.
Next: Authentication.

Trainer’s Guide

Lab: Wire the sketch, consume with fetch + reader in a tiny HTML page. Turn nginx buffering on/off and compare UX. Map Anthropic text_stream into the same envelope.

Whiteboard: Browser → FastAPI SSE → OpenAI stream. Mark TTFT. Cross out Celery in this picture.

Recap: Stream tokens through your API via SSE. Lock the door next: Authentication.