One piece at a time: how data flows through Python

How Python runs concurrent code drove a coroutine by hand with send() and wrote a 40-line event loop. Inside CPython looked at functions, frames and bytecode. This post puts the two together around one keyword: yield. It's what sits behind generators, coroutines, async generators, and the way FastAPI sends a 1 GB file or a chatbot's tokens to the browser without holding it all in RAM.

The route: why stream at all, what yield does inside CPython, the two kinds of pause in an async generator, the most dangerous part (cleanup), backpressure from yield all the way down to the socket, asyncio.Queue, then up to FastAPI's StreamingResponse, all of it put together into a chatbot answering token by token, and measurements.

Every number in this post was measured on a real machine: Linux (WSL2), 12 cores, Python 3.12 and 3.14, FastAPI 0.143, Starlette 1.8, uvicorn 0.54; the chatbot example uses transformers 5.20 and PyTorch 2.14 on the CPU.

Returning a million rows

An endpoint exports a CSV of a million rows (about 45 MB). The first version builds the whole file in memory, then returns it. The second sends batches of 1,000 rows as soon as they're ready:

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse, StreamingResponse

app = FastAPI()


def row(i: int) -> str:
    return f"{i},user{i},user{i}@example.com,{i * 7 % 1000}\n"


@app.get("/csv/list")
async def csv_list(n: int = 1_000_000):
    return PlainTextResponse("".join([row(i) for i in range(n)]))  # the whole file in memory first


async def rows(n: int, batch: int):
    buf = []
    for i in range(n):
        buf.append(row(i))
        if len(buf) == batch:
            yield "".join(buf)  # hand one chunk out, then pause here
            buf.clear()
    if buf:
        yield "".join(buf)


@app.get("/csv/stream")
async def csv_stream(n: int = 1_000_000, batch: int = 1000):
    return StreamingResponse(rows(n, batch), media_type="text/csv")

Each run starts a fresh server, then reads the process's peak memory (VmHWM in /proc/<pid>/status):

How it's returned	First byte after	Done after	Server peak memory
Build the whole file, then return it	371–1,142 ms	0.40–1.17 s	192 MiB
Stream batches of 1,000 rows	9–11 ms	0.32–0.36 s	51 MiB

The idle server uses 50 MiB. Building the whole file costs about 140 MiB more for one request: a list of a million strings, plus the joined string, plus its encoded bytes. Ten people downloading at once is over 1 GB. The streaming version costs almost nothing extra, the user gets the first byte after 10 ms, and the total time is even shorter because producing the data and sending it overlap.

All thanks to one keyword: yield.

What `yield` does inside CPython

If a function has a yield anywhere in its body, calling it runs none of it; it returns a generator:

import inspect


def numbers():
    total = 0
    for n in range(3):
        total += n
        sent = yield n
        print("  got back:", sent)
    return total


g = numbers()
print(inspect.getgeneratorstate(g))
print("first:", next(g))
print(inspect.getgeneratorstate(g), g.gi_frame.f_locals)
print("send:", g.send("hello"))
print("locals now:", g.gi_frame.f_locals)
try:
    g.send(None); g.send(None)
except StopIteration as stop:
    print("finished, return value:", stop.value)

GEN_CREATED
first: 0
GEN_SUSPENDED {'total': 0, 'n': 0}
  got back: hello
send: 1
locals now: {'total': 1, 'n': 1, 'sent': 'hello'}
  got back: None
  got back: None
finished, return value: 3

The previous post said that every call makes a frame holding the local variables and the current instruction, and a normal frame disappears when the function returns. A generator differs in exactly one way: its frame lives inside the generator object, on the heap, so it survives across runs. The bytecode shows it:

  6           RETURN_GENERATOR       # create the generator, move the frame into it, return it
              POP_TOP
      L1:     RESUME                   0
  ...
 10           LOAD_FAST_BORROW         1 (n)
              YIELD_VALUE              0   # hand n out, remember this spot, leave the frame
              RESUME                   5   # the next send() continues here
              STORE_FAST               2 (sent)

next(g) or g.send(x) runs the frame up to the next YIELD_VALUE. That instruction saves the position, hands the value out and leaves the frame without destroying it. Next time, the value passed to send becomes the result of the yield expression. When the function returns, CPython raises StopIteration carrying the return value; that's how a for loop knows it's over. A generator takes only 232 bytes on 3.14 (224 on 3.12), so keeping thousands of paused ones around is negligible.

Besides send, there are two more ways for the outside to step in:

g.throw(exc) raises an exception right at the paused yield.
g.close() raises GeneratorExit there, so the finally and with blocks inside run.

def lines():
    try:
        yield "a"
        yield "b"
    finally:
        print("  finally: cleanup")


g = lines()
next(g)
g = None  # nobody holds it any more
print("--- after the drop")

  finally: cleanup
--- after the drop

Nobody called close(), yet finally ran right away: when the reference count hits 0, CPython closes the generator itself. Keep this in mind, because for async generators it is not true any more.

Generators, coroutines, async generators: one machine

The first post in this series drove a coroutine with send(None). That's no coincidence: a coroutine is a renamed generator. The three kinds of function differ only by a flag in the code object:

Written as	Flag in `co_flags`	Calling it gives
`def` with `yield`	`CO_GENERATOR`	a generator: used with `for`
`async def`	`CO_COROUTINE`	a coroutine: used with `await`
`async def` with `yield`	`CO_ASYNC_GENERATOR`	an async generator: used with `async for`

All three share RETURN_GENERATOR, the frame on the heap, YIELD_VALUE and RESUME. Even the bytecode of await x is just a familiar yield loop:

GET_AWAITABLE                 # get the awaitable's iterator
LOAD_CONST       None
L2: SEND        (to L5)       # pass the value down; if the inner one finished, jump out
L3: YIELD_VALUE               # the inner one paused: pause us too, pass its value up
L4: RESUME
    JUMP_BACKWARD_NO_INTERRUPT (to L2)
L5: END_SEND                  # the result of the await

That's a generator's yield from. It explains the line from part 1, "await passes the request outwards, through every layer": each await layer is one SEND/YIELD_VALUE loop, passing the value up to the layer above until it reaches the event loop.

Async generators: two kinds of pause

An async generator has one tricky thing about it: a single function has two kinds of pause, going in two different directions.

await: waiting for I/O. The request has to go out to the event loop.
yield: a value is ready. It has to go to the async for loop that's reading.

The best way to see the two paths is to drive an async generator by hand, without asyncio, just like part 1:

import types


@types.coroutine
def wait_for(what):
    """Pause and hand `what` to whoever drives us (in asyncio: the event loop)."""
    reply = yield what
    return reply


async def rows():
    for page in (1, 2):
        batch = await wait_for(f"SELECT ... PAGE {page}")  # pause kind 1: goes out to the driver
        for row in batch:
            yield row  # pause kind 2: goes to whoever runs `async for`


agen = rows()
step = agen.__anext__()  # an awaitable: "work out the next item"
print("anext() gives:", type(step).__name__)
print("driver got request:", step.send(None))  # runs until the await inside rows()
try:
    step.send(["an", "binh"])  # answer the query: rows() runs on until its yield
except StopIteration as item:
    print("item:", item.value)

step = agen.__anext__()
try:
    step.send(None)  # the next row is already in memory: no request, straight to the yield
except StopIteration as item:
    print("item:", item.value)

step = agen.__anext__()
print("driver got request:", step.send(None))
try:
    step.send([])  # empty page 2: rows() ends
except StopAsyncIteration:
    print("StopAsyncIteration: the async for loop ends")

anext() gives: async_generator_asend
driver got request: SELECT ... PAGE 1
item: an
item: binh
driver got request: SELECT ... PAGE 2
StopAsyncIteration: the async for loop ends

async for row in rows() does exactly these steps: each round it calls __anext__() to get an awaitable, then awaits it. Inside that awaitable:

rows() hits an await: the value passes straight through, up to the event loop, as in any coroutine. The async for loop never sees it.
rows() hits a yield: the awaitable finishes with StopIteration(row), so to the async for loop, the await gives row.

So how does CPython tell them apart, when both are YIELD_VALUE? Look at the bytecode of yield row:

 16           LOAD_FAST_BORROW         2 (row)
              CALL_INTRINSIC_1         4 (INTRINSIC_ASYNC_GEN_WRAP)
              YIELD_VALUE              0

The value of a yield is wrapped in a special object before it goes out. The async_generator_asend object sees the wrapper and knows it's an item for async for; without the wrapper it's the event loop's business, and it passes it on up.

Async generators have asend(), athrow() and aclose(), the awaitable versions of send, throw and close. Two things from plain generators are banned here: yield from gives SyntaxError: 'yield from' inside async function, and a return with a value gives SyntaxError: 'return' with value in async generator.

The dangerous part: cleanup

Async generators often hold resources: a database connection, a file, a lock. So when the reader stops halfway (break, an exception, a client disconnecting), when does the finally block run? Three ways of writing it, under asyncio:

import asyncio
from contextlib import aclosing


async def rows(name):
    print(f"  [{name}] borrow a DB connection")
    try:
        for i in range(1_000):
            await asyncio.sleep(0)  # stands in for fetching the next row
            yield i
    finally:
        print(f"  [{name}] finally: give the connection back (task: {asyncio.current_task().get_name()})")


async def first_rows_held():
    stream = rows("held")  # a variable keeps the generator alive
    async for row in stream:
        if row == 2:
            break
    print("  handler: done with the rows, now doing slow work...")
    await asyncio.sleep(0.5)
    print("  handler: returns")


async def first_rows_inline():
    async for row in rows("inline"):  # no variable: its refcount drops to 0 at break
        if row == 2:
            break
    print("  handler: after the loop")
    await asyncio.sleep(0)
    print("  handler: after one more await")


async def first_rows_aclosing():
    async with aclosing(rows("aclosing")) as stream:
        async for row in stream:
            if row == 2:
                break
    print("  handler: after the loop")

first_rows_held
  [held] borrow a DB connection
  handler: done with the rows, now doing slow work...
  handler: returns
  [held] finally: give the connection back (task: Task-2)
first_rows_inline
  [inline] borrow a DB connection
  handler: after the loop
  handler: after one more await
  [inline] finally: give the connection back (task: Task-3)
first_rows_aclosing
  [aclosing] borrow a DB connection
  [aclosing] finally: give the connection back (task: first_rows_aclosing)
  handler: after the loop

Three things stand out:

held: the connection stays borrowed for the whole 0.5 seconds the handler spends on other work, and only goes back when the handler finishes and the stream variable disappears. With a pool of 10 connections and 10 such requests at once, the 11th request has to wait.
inline: even with nobody holding the generator, finally doesn't run right away as it would for a plain generator. It runs later, after the handler has moved on.
In both cases, finally runs in a different task (Task-2, Task-3), not the handler's.

The reason: an async generator's finally may need to await, say await conn.close(), and when the reference count hits 0 no event loop is running that generator. CPython can't await from inside the garbage collector. So it lets the event loop register two hooks with sys.set_asyncgen_hooks():

# asyncio/base_events.py (3.14)
def _asyncgen_firstiter_hook(self, agen):
    ...
    self._asyncgens.add(agen)  # remember every async generator that started

def _asyncgen_finalizer_hook(self, agen):
    self._asyncgens.discard(agen)
    if not self.is_closed():
        self.call_soon_threadsafe(self.create_task, agen.aclose())  # close it later, in a new task

When an unfinished async generator is collected, asyncio schedules aclose() in a new task. And when asyncio.run() ends, it calls loop.shutdown_asyncgens() to close whatever is left. In practice:

The resource goes back late, at a moment you can't predict.
If the finally uses something that was cleaned up before it (a database session closed along with the request), the error shows up in some unfamiliar task, and it's very hard to trace.

The fix is contextlib.aclosing (Python 3.10+). It calls await agen.aclose() as soon as the async with block exits, in your own task, as in the third case. The rule: read any async generator that holds a resource inside async with aclosing(...).

Backpressure: who makes whom wait

A generator only runs when someone asks for the next item. That gives streaming a built-in brake, called backpressure: when the reader is slow, the producer slows down with it. But it only works if every layer in between is willing to wait.

asyncio Streams and `drain()`

The lowest layer of asyncio you'd use directly is Streams: asyncio.open_connection() and asyncio.start_server() give you a StreamReader/StreamWriter pair. Part 1 used them to write its client. The thing to know: writer.write() never waits. It only copies the data into the transport's buffer and returns at once. To wait, you have to call await writer.drain().

Here's a server sending 200 MiB to a client that reads nothing for the first second:

CHUNK = b"x" * 64 * 1024


async def serve(reader, writer, use_drain):
    for i in range(3_200):  # 200 MiB
        writer.write(i.to_bytes(4) + CHUNK[4:])  # never waits: it only appends to the buffer
        if use_drain:
            await writer.drain()  # waits while the buffer is above the high-water mark
    await writer.drain()
    writer.close()


async def slow_client(port):
    reader, writer = await asyncio.open_connection("127.0.0.1", port)
    await asyncio.sleep(1)  # the client doesn't read for a second
    while await reader.read(1 << 20):
        pass

	Data waiting in the buffer (peak)	Peak memory
Without `drain()`	197.5 MiB	218 MiB
With `drain()`	0.1 MiB	21 MiB

Without drain(), the server piles all 200 MiB into RAM while the client isn't reading. With drain(), the buffer never grows past a few dozen KiB. writer.transport.get_write_buffer_limits() returns (16384, 65536): when the buffer goes over 64 KiB, the transport calls pause_writing() and drain() starts waiting; when it drops below 16 KiB, resume_writing() lets it go on.

From `yield` down to the socket

uvicorn does exactly this. Each time the app calls send() with a piece of body, it checks this flag first:

# uvicorn/protocols/http/httptools_impl.py
async def send(self, message):
    if self.flow.write_paused and not self.disconnected:
        await self.flow.drain()  # the transport said its buffer is over 64 KiB: wait
    ...

Put together, here is a StreamingResponse's chain of waiting:

The client reads slowly, so its receive buffer in the kernel fills up.
The server's send buffer in the kernel fills up too.
The transport's buffer goes over 64 KiB, and uvicorn sets write_paused.
Starlette's await send(...) waits in drain().
The async for loop stops calling __anext__().
Your generator stays put at its yield and produces nothing more.

Measured: an endpoint streaming 625 MiB in 64 KiB pieces, and a client that reads the headers and then stops reading:

client not reading for 0.5 s: generator produced 9.75 MiB of 625 MiB
client not reading for 1 s: generator produced 9.75 MiB of 625 MiB
client not reading for 2 s: generator produced 9.75 MiB of 625 MiB

The generator stops at 9.75 MiB and stays there. That's roughly the sum of the buffers along the way: on this machine the kernel allows up to 4 MiB of send buffer (tcp_wmem) and up to 6 MiB of receive buffer (tcp_rmem), plus uvicorn's 64 KiB.

`asyncio.Queue`: when producer and reader are two tasks

A generator ties the producer and the reader to the same beat: one item asked for, one item made. Often it isn't like that:

Data is pushed to you, through a callback or from a thread, rather than pulled.
Several sources pour into one place, or one source is shared among several workers.
The producer runs in the background, independent of any request.

Then the two sides are two tasks, joined by an asyncio.Queue.

An unbounded queue is a trap

A producer pushes a million events, and a consumer spends 1 ms on each one:

async def producer(queue, stats):
    for i in range(1_000_000):
        await queue.put(f"event {i}")  # with maxsize, waits here while the queue is full
        stats["produced"] += 1


async def consumer(queue):
    while True:
        await queue.get()
        await asyncio.sleep(0.001)  # 1 ms per event: a slow client, a slow API


queue = asyncio.Queue(maxsize=0)  # 0 = unbounded; then the same with maxsize=100

Python 3.12	After 1 second: produced	Waiting in the queue	Memory
`maxsize=0` (unbounded)	1,000,000	999,561	90 MiB
`maxsize=100`	933	100	31 MiB

The unbounded queue causes two problems:

Memory grows with the gap between the two sides.
The event loop is blocked. The producer made all million events in 0.68 s on 3.12 (0.34 s on 3.14), and the consumer didn't run once in that time. await queue.put() on a queue that isn't full returns at once without giving up the loop. await only gives up the loop when it actually has to wait.

With maxsize=100, put() has to wait when the queue is full, so the producer runs at the consumer's pace. Nearly every queue joining two sides that run at different speeds should have a maxsize.

Turning "push" data into a "pull" async generator

A very common case: a synchronous SDK calls a callback for each token, from its own thread, while you need an async generator to hand to StreamingResponse:

import asyncio
import threading
import time


def sdk_stream(on_token, on_done):
    """A blocking SDK that pushes tokens to a callback from its own thread."""
    for token in "Xin chào , mình là chatbot".split():
        time.sleep(0.1)
        on_token(token)
    on_done()


async def tokens():
    """Turn the push-style callbacks into a pull-style async generator."""
    loop = asyncio.get_running_loop()
    queue = asyncio.Queue(maxsize=16)

    def on_token(token):  # runs in the SDK's thread: never touch the queue directly from there
        asyncio.run_coroutine_threadsafe(queue.put(token), loop).result()  # blocks the thread while full

    def on_done():
        loop.call_soon_threadsafe(queue.shutdown)  # 3.13+: wakes the reader with QueueShutDown

    threading.Thread(target=sdk_stream, args=(on_token, on_done), daemon=True).start()
    while True:
        try:
            yield await queue.get()
        except asyncio.QueueShutDown:
            return


async def main():
    start = time.perf_counter()
    async for token in tokens():
        print(f"  {time.perf_counter() - start:.1f} s  {token}")


asyncio.run(main())

  0.1 s  Xin
  0.2 s  chào
  0.3 s  ,
  0.4 s  mình
  0.5 s  là
  0.6 s  chatbot

Three details matter:

asyncio.Queue isn't safe to use from another thread. From the SDK's thread, ask the loop to do it for you, with run_coroutine_threadsafe or call_soon_threadsafe.
.result() makes the SDK's thread wait while the queue is full. Backpressure crosses the thread boundary.
queue.shutdown() (Python 3.13+) is the clean way to say "that's all". Every get() that is waiting or called afterwards raises QueueShutDown, instead of you inventing an end marker.

Several workers, one queue

shutdown() also solves an old problem of the worker pattern: how to tell workers that are waiting for jobs that it's over, so they can exit.

async def worker(name, queue):
    try:
        while True:
            job = await queue.get()
            await asyncio.sleep(0.01)
            print(f"  {name} did {job}")
            queue.task_done()
    except asyncio.QueueShutDown:
        print(f"  {name}: queue shut down, stopping")


async def main():
    queue = asyncio.Queue()
    async with asyncio.TaskGroup() as tg:
        for n in range(3):
            tg.create_task(worker(f"worker-{n}", queue))
        for job in range(5):
            queue.put_nowait(f"job {job}")
        await queue.join()  # every job got task_done()
        queue.shutdown()  # wake idle workers so the TaskGroup can finish

  worker-0 did job 0
  worker-1 did job 1
  worker-2 did job 2
  worker-0 did job 3
  worker-1 did job 4
  worker-2: queue shut down, stopping
  worker-0: queue shut down, stopping
  worker-1: queue shut down, stopping

Back up to FastAPI: `StreamingResponse` from the inside

Here's the core of StreamingResponse in Starlette 1.8, shortened:

class StreamingResponse(Response):
    def __init__(self, content, ...):
        if isinstance(content, AsyncIterable):
            self.body_iterator = content
        else:
            self.body_iterator = iterate_in_threadpool(content)  # a plain generator: each next() in a thread

    async def stream_response(self, send):
        await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
        async for chunk in self.body_iterator:
            await send({"type": "http.response.body", "body": chunk, "more_body": True})
        await send({"type": "http.response.body", "body": b"", "more_body": False})

    async def __call__(self, scope, receive, send):
        # uvicorn speaks ASGI 2.3: run the stream and a disconnect listener side by side
        async with create_collapsing_task_group() as task_group:
            async def wrap(func):
                await func()
                task_group.cancel_scope.cancel()  # whichever finishes first cancels the other

            task_group.start_soon(wrap, partial(self.stream_response, send))
            await wrap(partial(self.listen_for_disconnect, receive))

It's just an async for loop around your generator, with one await send(...) per item and more_body: True. uvicorn sends each item as one piece of chunked encoding: a line with the length (in hex), the data, then \r\n. Three consequences are worth knowing.

A plain `def` generator: one thread switch per item

If content is a plain generator, Starlette wraps it in iterate_in_threadpool, so every next() runs in AnyIO's thread pool, like a def endpoint in part 1. That's not wrong, since a synchronous generator may block, but every thread switch costs something. The numbers are in the last section.

The client disconnects: cancellation can only get in at an `await`

uvicorn announces ASGI version 2.3, so Starlette runs two things side by side: streaming the body, and listening on receive(). When the client closes the connection, receive() returns http.disconnect and the task group cancels the streaming. But cancelling in asyncio means raising CancelledError where the task is waiting. If the task isn't waiting anywhere, there's nowhere to raise it.

The experiment: the 625 MiB endpoint from above, with a client that stops reading for 2 seconds and then cuts the connection. Two versions of the generator: one only builds data in memory, the other adds an await asyncio.sleep(0) per piece:

async def big_chunks(pause: bool):
    try:
        for i in range(10_000):
            if pause:
                await asyncio.sleep(0)  # a real await: the loop gets control back here
            yield i.to_bytes(4) + b"x" * (64 * 1024 - 4)
    finally:
        ...  # record how far we got

Generator	Produced when the client left	Produced in total	`CancelledError` in the generator
No real `await`	9.75 MiB	625 MiB	No
With `await asyncio.sleep(0)`	9.44 MiB	9.44 MiB	Yes, `finally` runs right away

The version without an await produced over 600 MiB more after the client had gone, sending it to nobody. Why: once the connection is closed, uvicorn's send() returns at once (if self.disconnected: return) without waiting, the generator doesn't wait on anything either, so the async for loop runs straight to the end and the cancellation has nowhere to get in. During all that, the event loop can't serve any other request either.

A real generator that reads from a database or an API usually has real awaits, so it rarely hits this. But a generator walking over data already in memory, or just computing, hits exactly this. One await asyncio.sleep(0) per piece is enough to make it stoppable.

A full example: a chatbot answering token by token

All the pieces above come together in the most familiar case: a chatbot answering word by word. This example runs a real language model, Qwen2.5-0.5B-Instruct, right on the CPU with the transformers library. It's a very small model, so its answers aren't good; what matters here is the timing and the path each token takes, not the content.

Why stream at all

A model produces one token at a time. The question "Viết một đoạn ngắn giới thiệu ngôn ngữ lập trình Python cho người mới bắt đầu" ("Write a short introduction to Python for beginners") gives 180 tokens:

	Time
First piece of text reaches the browser	0.69–0.71 s
Whole answer done	10.34–10.51 s

Without streaming, the user stares at an empty screen for over 10 seconds. With streaming, text starts appearing after 0.7 seconds and keeps flowing, each piece about 0.05–0.1 s after the last:

   0.69 s  'Tất '
   0.79 s  'nhiên, '
   0.84 s  'tôi '
   0.90 s  'sẽ '
   0.95 s  'giúp '
   1.00 s  'bạn '

APIs like OpenAI's or Anthropic's generate much faster than a model on a laptop CPU, but the proportion holds: a long answer still takes many seconds, while the first token arrives almost at once.

The model runs in a thread, tokens go through a queue

model.generate() is a synchronous function that runs for seconds or tens of seconds, so it must not run on the event loop. It runs in its own thread and sends text out through a "streamer": an object that generate() calls whenever it has more text. This is exactly the "push" case from the asyncio.Queue section: the model's thread pushes text into the queue, and an async generator pulls it out and sends it as SSE.

Server-Sent Events is the simplest streaming format browsers understand out of the box: Content-Type: text/event-stream, with each event made of data: ... lines and ended by a blank line.

import asyncio
import json
import threading

import torch
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from transformers import AutoModelForCausalLM, AutoTokenizer, StoppingCriteria, TextStreamer

MODEL = "Qwen/Qwen2.5-0.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(MODEL)
model = AutoModelForCausalLM.from_pretrained(MODEL)
app = FastAPI()


class QueueStreamer(TextStreamer):
    """Gets each new piece of text from the model's thread and puts it on an asyncio.Queue."""

    def __init__(self, queue, loop):
        super().__init__(tokenizer, skip_prompt=True, skip_special_tokens=True)
        self.queue, self.loop = queue, loop

    def on_finalized_text(self, text, stream_end=False):
        try:
            if text:
                # waits (in the model's thread) while the queue is full: backpressure on the model
                asyncio.run_coroutine_threadsafe(self.queue.put(text), self.loop).result()
            if stream_end:
                self.loop.call_soon_threadsafe(self.queue.shutdown)
        except asyncio.QueueShutDown:
            pass  # the reader is gone; StopWhen below ends generation at the next token


class StopWhen(StoppingCriteria):
    """Checked by generate() after every token: stop as soon as the event is set."""

    def __init__(self, event):
        self.event = event

    def __call__(self, input_ids, scores, **kwargs):
        return torch.full((input_ids.shape[0],), self.event.is_set(), dtype=torch.bool)


async def answer(question: str):
    loop = asyncio.get_running_loop()
    queue = asyncio.Queue(maxsize=16)
    stop = threading.Event()
    inputs = tokenizer.apply_chat_template(
        [{"role": "user", "content": question}], add_generation_prompt=True, return_tensors="pt", return_dict=True
    )

    def generate():
        model.generate(
            **inputs, max_new_tokens=180, do_sample=False,
            streamer=QueueStreamer(queue, loop), stopping_criteria=[StopWhen(stop)],
        )

    threading.Thread(target=generate, daemon=True).start()  # the model blocks: run it in its own thread
    try:
        while True:
            try:
                text = await queue.get()  # a real await: cancellation lands here if the user leaves
            except asyncio.QueueShutDown:
                yield "data: [DONE]\n\n"
                return
            yield f"data: {json.dumps(text, ensure_ascii=False)}\n\n"
    finally:
        stop.set()  # the user left (or we're done): stop generating
        queue.shutdown(immediate=True)  # and unblock the model's thread if it's waiting on a full queue


@app.get("/chat")
async def chat(q: str):
    return StreamingResponse(answer(q), media_type="text/event-stream")

Every piece of this code appeared above:

The model's thread never touches the queue directly: it asks the event loop through run_coroutine_threadsafe and call_soon_threadsafe.
maxsize=16 and .result(): if the user reads slowly, the model's thread waits. That's backpressure reaching all the way to the model.
await queue.get() is a real await, so when the user leaves, Starlette's cancellation has somewhere to get in.
queue.shutdown() says "that's all" when the model is done. shutdown(immediate=True) in finally frees the model's thread if it's waiting on a full queue.

In the browser

<p id="answer"></p>
<script>
const answer = document.querySelector("#answer");
const events = new EventSource("/chat?q=" + encodeURIComponent("Viết một đoạn ngắn giới thiệu Python."));
events.onmessage = (event) => {
  if (event.data === "[DONE]") return events.close();  // otherwise EventSource reconnects and asks again
  answer.textContent += JSON.parse(event.data);
};
</script>

The [DONE] event and events.close() aren't decoration. When the server ends the stream, EventSource treats it as a lost connection and reconnects by itself. Tried in Chromium with a 3-word stream: without close(), the page sent the request again 3–4 times in 10 seconds, about every 3 seconds, and the answer appeared 4 times in a row. With a real chatbot, each of those is another run of the model.

When the user leaves

The user closes the tab after receiving 5 pieces of text. Comparing the code above with a version without the finally:

	Tokens when the user left	2 seconds later	In the end
With `finally`: `stop.set()` + `shutdown(immediate=True)`	8	9	Stops at 9 tokens
Without `finally`	8	25	Stuck at 25 tokens forever; the thread never ends

With the finally, the cancellation gets in at await queue.get(), finally sets the stop flag, and generate() stops at the very next token. Without it, the model keeps producing text for nobody until the 16-slot queue is full. Then the model's thread waits in put() forever: a leaked thread, holding its memory and tensors, for every user who leaves halfway. With an unbounded queue it wouldn't get stuck; it would run all 180 tokens instead, spending CPU (or GPU) on an answer nobody sees.

With OpenAI's or Anthropic's API the structure is the same, except the SDK already gives you an async iterator, so there's no thread and no queue. What you must keep is the finally (or async with) that closes the stream to the API as soon as the user leaves, instead of reading on to the end of an answer nobody is watching.

Through a proxy

There's usually a reverse proxy between the user and uvicorn, and a proxy may hold data back before sending it. To measure cleanly, this experiment replaces the model with a fake generator, exactly 0.1 s per token, 19 tokens, with the client sending Accept-Encoding: gzip:

Path	First token after	Tokens arrive
Straight into uvicorn	0.1 s	One at a time
Caddy 2.11, `encode gzip`	0.1 s	One at a time, compressed
nginx 1.31, `gzip on`	1.91 s	The whole sentence at once, in one read
nginx 1.31, `gzip on`, app sends `X-Accel-Buffering: no`	0.1 s	One at a time, compressed

With nginx and gzip on, the user stares at an empty screen for almost 2 seconds, then the whole sentence appears at once, exactly the "frozen chatbot" feeling, while the server is streaming correctly. The X-Accel-Buffering: no header tells nginx not to hold back that one response. Caddy (this website runs behind Caddy with encode gzip) sends each token right away. Starlette 1.8's GZipMiddleware is safe too: it skips text/event-stream, and for other streams it flushes after every piece.

The general lesson: when streaming "doesn't work", measure each hop (straight into the app, through the proxy, through the CDN) before changing code.

Measurements

What does one yield cost? Looping over 5 million items, best of 5 runs:

How it loops	Python 3.12	Python 3.14
Build a `list`, then `for`	32.0 ns/item	30.7 ns/item
`for` over a generator	22.6 ns/item	14.2 ns/item
`async for` over an async generator	99.6 ns/item	86.7 ns/item

yield itself is very cheap. An async generator costs about 4–6 times more because each item goes through an asend awaitable and a StopIteration, but it's still under 0.1 µs.

And inside StreamingResponse? The same CSV as at the top, changing the piece size and the kind of generator:

Generator	Each piece	Rows	Time
`async def`	1,000 rows	1,000,000	0.32–0.36 s
Plain `def`	1,000 rows	1,000,000	0.40–0.45 s
`async def`	1 row	1,000,000	3.08–3.17 s
`async def`	1 row	100,000	0.32–0.34 s
Plain `def`	1 row	100,000	9.71–9.76 s

The cost is per piece, not per yield. Each piece is one send() through Starlette, uvicorn and the transport, about 3 µs. Yielding row by row is about 10 times slower than batching 1,000 rows. It also adds 6 MB on the wire, because each chunked-encoding piece has its own header and trailer.
A plain def generator with small pieces is the worst: about 97 µs per row, because every row is a switch to the thread pool and back. For a million rows, that's an estimated minute and a half or more. Batch the pieces and it's nearly as fast as async def.

The rule: batch into pieces of a few dozen KiB (unless each part must go out at once, like a chatbot's tokens), and use async def when the data source allows it.

In short

Concept	What it really is
Generator	A frame living on the heap; `YIELD_VALUE` saves the position and leaves the frame, `send()` resumes it
Coroutine	The same machine; `await` is a `SEND`/`YIELD_VALUE` loop passing values up to the event loop
Async generator	Two kinds of pause: `await` goes out to the event loop, `yield` (wrapped) goes to `async for`
Cleanup	A plain generator closes as soon as nobody holds it; an async generator is closed later, in another task. Use `aclosing`
Backpressure	When the reader is slow, `drain()` waits, `send()` waits, and the generator stays at `yield`
`asyncio.Queue`	Joins two tasks; always set `maxsize`; `shutdown()` to say it's over (3.13+)
`StreamingResponse`	An `async for` loop around the generator; one `send()` per item
Cancelling when the client leaves	Can only get in at an `await` that really waits
Chatbot	Model in a thread → bounded queue → async generator → SSE; `finally` stops the model when the user leaves

Everything in this post comes back to two questions from part 1: where is the data sitting while it waits, and who is waiting for whom? A good stream is one where, at every layer, the answers are "only a small piece" and "the fast side waits for the slow one".