The async generator pattern

Welcome! I'm Param. In this workshop we take a FastAPI AI app from a demo on your laptop to a service you can deploy with confidence. We start with the piece that breaks first in production: streaming.

LLMs generate tokens one at a time. If you buffer the full response before sending it back, your users stare at a spinner for ten seconds. Server-Sent Events push each token to the browser as soon as it is generated, over a regular HTTP response with a special content type.

SSE streaming flow

Client opens one HTTP connection, server emits events until the stream finishes.

service.py
python
import asyncio
import json
from datetime import datetime
from typing import AsyncGenerator

async def generate_chat_stream(request) -> AsyncGenerator[str, None]:
    prompt = request.prompt
    if request.system:
        prompt = request.system + "\n\n" + prompt

    yield f"data: {json.dumps({'status': 'connected'})}\n\n"

    heartbeat_task = asyncio.create_task(_heartbeat_ticker())
    try:
        async for chunk in llm_provider.generate_stream(prompt):
            payload = {"content": chunk, "timestamp": datetime.utcnow().isoformat()}
            yield f"data: {json.dumps(payload)}\n\n"
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e)})}\n\n"
    finally:
        heartbeat_task.cancel()

    yield f"data: {json.dumps({'done': True})}\n\n"

An async generator that yields SSE-formatted chunks. Each yield is one event the browser picks up immediately. The finally block cancels the heartbeat task so it does not leak past the request.

Because time-to-first-token is the metric users feel. With a generator the client starts rendering after a few hundred milliseconds. Buffering forces a ten-second wait on a long answer, and the response size is unbounded so you cannot stream it at all for long generations.

Quiz: Quiz

Loading practice…