Building a Streaming API With LLMs

Why Streaming Matters for UX

Without streaming, a 500-token response takes 5-15 seconds to appear. With streaming, the first token appears in under a second. The psychological difference is enormous - streaming makes AI tools feel responsive and interactive rather than slow and unpredictable.

Server-Side Streaming With FastAPI

pythonCopy
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic
import json

app = FastAPI()
client = anthropic.Anthropic()

@app.post("/chat/stream")
async def chat_stream(request: dict):
    query = request.get("message", "")

    def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": query}],
        ) as stream:
            for text_chunk in stream.text_stream:
                # SSE format: "data: <json>\n\n"
                yield f"data: {json.dumps({'chunk': text_chunk})}\n\n"
        # Signal stream end
        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Required for nginx
        }
    )

Client-Side Consumption (JavaScript)

javascriptCopy
async function streamChat(message) {
    const response = await fetch('/chat/stream', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({message}),
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
        const {done, value} = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, {stream: true});
        const lines = buffer.split('\n\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
            if (!line.startsWith('data: ')) continue;
            const data = JSON.parse(line.slice(6));
            if (data.done) return;
            if (data.chunk) {
                // Append to your UI element
                document.getElementById('response').textContent += data.chunk;
            }
        }
    }
}

Streaming for RAG Pipelines

In a RAG pipeline, retrieval must complete before generation begins. Stream only the generation step:

pythonCopy
@app.post("/rag/stream")
async def rag_stream(request: dict):
    query = request.get("query", "")

    # Retrieval (blocking  -  must complete before streaming)
    candidates = retrieve(query, top_k=20)
    top_chunks = rerank(query, candidates, top_k=5)
    context = assemble_context(top_chunks)

    def generate():
        # Optionally stream the sources first
        sources = [{"source": c["source"], "page": c.get("page")} for c in top_chunks]
        yield f"data: {json.dumps({'sources': sources})}\n\n"

        # Then stream the response
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"Answer using only this context:\n{context}\n\nQuestion: {query}"
            }],
        ) as stream:
            for chunk in stream.text_stream:
                yield f"data: {json.dumps({'chunk': chunk})}\n\n"
        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Token Counting During Streaming

pythonCopy
async def stream_with_cost_tracking(messages, **kwargs):
    total_input_tokens = 0
    total_output_tokens = 0

    with client.messages.stream(messages=messages, **kwargs) as stream:
        for text in stream.text_stream:
            yield text

    # Totals available after stream completes
    usage = stream.get_final_message().usage
    log_cost(usage.input_tokens, usage.output_tokens, kwargs.get("model"))

Handling Stream Interruption

Users may cancel mid-stream (navigate away, click stop). Handle this gracefully:

pythonCopy
from asyncio import CancelledError

async def interruptible_stream(query: str):
    try:
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": query}],
        ) as stream:
            for chunk in stream.text_stream:
                yield chunk
    except CancelledError:
        # Client disconnected  -  this is expected, not an error
        pass
    except Exception as e:
        yield f"\n\n[Error: {str(e)}]"

Latency Optimization

Time to first token (TTFT) is the metric that determines perceived responsiveness. Optimize it:

• Use smaller models for simple tasks (Haiku is 2-3x faster than Sonnet to first token)
• Keep system prompts short (longer prompts take longer to process)
• Use prompt caching for static system prompts - the first request after cache warm-up has TTFT < 500ms

The best investment for perceived performance: reduce TTFT to under 500ms for your most common query types.

Common Mistakes

Buffering the full stream before parsing instead of streaming to the client. If your server collects all SSE events into a buffer and only then writes to the client, you get zero benefit from streaming - the user still waits for full generation to complete. True streaming requires passing chunks to the response object immediately as they arrive from the LLM provider.

Not handling SSE reconnection. The SSE spec defines a Last-Event-ID header that clients send on reconnect so the server can replay missed events. Most LLM proxy implementations omit this, meaning a network blip during a long generation silently drops content. Either implement event IDs and a short replay window, or clearly document that reconnection is not supported and let the client restart the request.

Ignoring that streaming changes the time-to-first-token vs. total latency tradeoff. Streaming dramatically improves perceived responsiveness (first token appears quickly) but does not reduce total generation time - it may actually increase it slightly due to protocol overhead. Design your latency SLAs around time-to-first-token for streaming endpoints, not total request duration.

What to Practice Next

• Implement a FastAPI streaming endpoint that proxies an OpenAI or Anthropic streaming response directly to the HTTP client using StreamingResponse; verify the first token appears in the browser within a few hundred milliseconds.
• Simulate a mid-stream connection drop using a tool like tc or Toxiproxy and observe how your client handles partial output; implement a retry strategy that restarts cleanly.
• Measure time-to-first-token and total generation time for the same prompt with and without streaming and explain the difference to a teammate.