High-Throughput Async & Streaming Pipelines

Why async for AI systems motivation

Picture your evaluation script the night before a launch. It has 500 golden test cases to run through the model, each call takes about two seconds, and you kick it off expecting to grab a coffee. Twenty minutes later it's still going, one call at a time, each one sitting and waiting for the network while your laptop does absolutely nothing. That dead waiting is the whole problem async solves — and once you see it, you can't unsee it in every AI system you touch.

Here's the mental picture that makes it click. Imagine a chef with one pot of pasta boiling. The naive chef stands and stares at the pot for ten minutes, then chops the vegetables, then stares at the oven. A good chef puts the pasta on, and while it boils chops the vegetables, sets the table, plates the starter. Nothing cooks faster — but the whole meal lands in a fraction of the time, because the chef fills the waiting with other useful work. Async is exactly that chef, and each "waiting for the API to answer" is a pot on the boil. Your program isn't working harder; it's just refusing to stand and stare.

The reason this matters so much for AI specifically is that LLM work is I/O-bound — you spend the overwhelming majority of every request just waiting for bytes to travel to a data center and back. A normal web function might be CPU-bound (crunching numbers), but an LLM call is 99% waiting, 1% doing. When almost all your time is waiting, overlapping that waiting is the single biggest lever you have — bigger than a faster CPU, bigger than more RAM.

The mistake nearly everyone makes first is assuming "faster" means "more machines" or "more threads." You reach for a bigger box, or you spawn 500 threads, and you're surprised when it barely helps or falls over. The truth is you don't need more workers to do more — you need one worker that stops waiting. A single async thread can hold thousands of half-finished requests in the air at once, each parked at its own "waiting for the network" line, resuming the instant its answer arrives. That's a different shape of solution than "add hardware," and getting the shape right is what this whole part teaches.

Say each API call takes 2 seconds. Evaluating 500 golden cases sequentially is ~17 minutes; with async concurrency (respecting rate limits) it's under a minute. Async also powers streaming — showing tokens as the model produces them, which transforms perceived latency in a chat UI. Same CPU, dramatically better throughput and UX, because the thread does useful work instead of blocking on the network.

1 · The event loop — the core idea intermediate

Everything in async rests on one gear that's always turning, and if you understand it, nothing else surprises you. That gear is the event loop. Skip it and async feels like magic keywords you sprinkle and hope; grasp it and every await, every "why did my task stall" bug, every design choice becomes obvious. So spend a minute here — this is the section the rest of the page depends on.

The clearest analogy is a waiter working several tables in a restaurant. The waiter takes table one's order, walks it to the kitchen, and — crucially — does not stand at the pass waiting for the food. They go take table two's order, refill table three's water, clear table four. When the kitchen bell dings that table one's food is ready, the waiter picks it up and delivers it. One waiter, many tables, never idle — because the moment any table needs waiting, the waiter goes and does something else useful. The event loop is that waiter, your coroutines are the tables, and "waiting for the kitchen" is waiting for the network.

Now the precise version, which will feel natural after the analogy. The event loop runs one task at a time (it's single-threaded — there is genuinely only one waiter). It runs that task forward until the task hits an await on something not yet ready — a network response, a timer, another coroutine. At that point the task politely yields: it says "I'm going to be waiting a while, go run someone else." The loop picks another ready task and runs it forward to its next await, and so on. When an awaited result finally arrives, the loop schedules that task to resume exactly where it paused. Round and round the gear turns, always giving the one thread to whoever can make progress.

The word "cooperative" is the catch that trips people, so plant it now. This scheduling is cooperative, not preemptive: the loop can only switch tasks at an await point. Nothing forcibly interrupts a running coroutine. That's a gift — you never get a task switch in the middle of a delicate update, so most async code is free of the locking headaches threads have. But it's also a loaded gun: if one task does something slow without awaiting — a heavy CPU loop, a blocking call — it never yields, the waiter is frozen at that one table, and every other task starves. That single failure mode is behind most "why is my async program hanging?" tickets, and it's exactly what the warning box a little further down forbids. Hold onto "the loop only switches at await" and you'll debug async faster than most engineers ever do.

pythonimport asyncio

async def main():           # a coroutine function
    print("start")
    await asyncio.sleep(1)   # yields control for 1s instead of blocking
    print("done")

asyncio.run(main())          # create loop, run main() to completion, close loop

2 · Coroutines & await intermediate → advanced

Here's a bug that has cost every async newcomer at least one baffled afternoon. You write an async function, you call it, the program runs clean with no error — and the work inside simply never happened. No crash, no traceback, just silence where your API call should have been. The cause is a single missing keyword, and understanding why it fails so quietly is the fastest way to make async stop feeling slippery.

The key idea is that an async def function is not like a normal function. A normal function does the work the moment you call it. An async def function, when you call it, does nothing but hand you back a coroutine — think of it as a sealed recipe card, not a cooked meal. The card lists every step, but writing the card down doesn't cook anything. Someone still has to take that card to the stove. That "someone" is either an await or the event loop scheduling it as a task. Calling complete("hi") writes the card; await complete("hi") actually cooks it.

And await is the other half of the pair — it's the verb to the coroutine's noun. Saying await x means two things at once: "cook this recipe and hand me its result" and "while it's cooking, I don't need the stove — let the event loop run someone else." That second half is the whole point. Every await is a voluntary yield point, a place you've marked as "safe to pause me here." This is why you can only use await inside an async def: it's the language making you declare, at each waiting spot, that you're happy to step aside so the one thread stays busy.

So the classic mistake — creating a coroutine and never awaiting it — is really "writing the recipe card and leaving it on the counter." The work is fully described and completely un-done. Python does try to help: at the end of the program it notices the abandoned card and prints RuntimeWarning: coroutine '...' was never awaited. Learn to read that warning as "you forgot an await somewhere," because it's pointing straight at real work that silently didn't run. The lab below shows the correct path and the abandoned-card bug side by side so the difference burns in.

pythonimport asyncio

async def complete(prompt):
    await asyncio.sleep(0.5)          # pretend network latency
    return f"answer to: {prompt}"

async def main():
    result = await complete("hello")   # await = run & get the value
    print(result)
    coro = complete("oops")             # NOT awaited -> never runs (warning)
    coro.close()

asyncio.run(main())

3 · Running things concurrently — gather & TaskGroup advanced

This is the section where the seventeen-minute evaluation from the top of the page becomes under a minute — the moment async pays for itself. Everything so far has been setup: the loop, coroutines, await. But awaiting one call at a time is still one-at-a-time; you've learned to not-block, but you haven't yet launched many things at once. That launch is what this section is about, and it's the payoff you came for.

The analogy is putting all the pots on the stove before you start staring at any of them. Awaiting calls one after another is boiling pot one to done, then starting pot two, then pot three — the pots never overlap, so total time is the sum. Launching them concurrently is lighting all the burners together and then waiting; now the boils overlap and total time is just the longest single pot. One hundred calls that each wait one second, fired together, finish in about one second — not one hundred — because the hundred waits happen on top of each other instead of end to end. That collapse from sum to max is the entire economics of async fan-out.

Python gives you two tools for this, and knowing when to use which is the mark of someone who's shipped async. asyncio.gather is the classic: hand it a pile of coroutines, it launches them all and gives you back their results in the same order you submitted them — which matters enormously when you need to line each answer back up with its input. asyncio.TaskGroup (Python 3.11+) is the modern, safer form built on "structured concurrency": you create tasks inside an async with block, and the block simply won't exit until every task is done. Its real superpower shows up on failure — if one task blows up, the group automatically cancels its siblings and reports the error cleanly, instead of leaving orphaned tasks running in the background.

The common mistake here is treating a failure in one of many concurrent calls as an afterthought. With plain gather, when one coroutine raises, you get that one exception — but by default the others may keep running, half-finished and unwatched, quietly burning your rate limit or leaking connections. Beginners write a happy-path gather over 1,000 calls, one 500-errors, and they spend the evening puzzling over "zombie" tasks and mystery costs. This is precisely the problem TaskGroup was designed to end: fail one, cancel the rest, surface the error, exit clean. For anything production, prefer it — and when you must use gather, decide deliberately how failures are handled rather than discovering the answer in an incident.

pythonimport asyncio, time

async def complete(i):
    await asyncio.sleep(1)
    return i * 2

async def main():
    # gather: all at once, results in submission order
    results = await asyncio.gather(*[complete(i) for i in range(100)])
    print(len(results))              # 100, in ~1 second (not 100s)

    # TaskGroup (3.11+): structured — cleaner errors & cancellation
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(complete(i)) for i in range(100)]
    results = [t.result() for t in tasks]   # all done when the block exits

asyncio.run(main())

4 · Backpressure — bounding concurrency with a semaphore expert intermediate

You just learned to fire a thousand calls at once. Now learn why doing exactly that will get your API key throttled, your process OOM-killed, and your on-call phone ringing at 2am. The last section handed you a firehose; this one hands you the nozzle. If you remember only one production pattern from this entire page, make it this one — it's the difference between a demo and a system that survives real load.

Think of a motorway on-ramp with a metering light — that red-green signal that releases one car every few seconds during rush hour. It feels annoying when you're the car waiting, but it's the reason the motorway keeps flowing instead of collapsing into a jam. Dumping all the cars on at once doesn't get anyone there faster; it grinds everyone to a halt. A semaphore is that metering light for your coroutines: it lets a fixed number through at a time and makes the rest wait their turn at the line. The technical name for "make the source slow down so the destination can keep up" is backpressure, and the semaphore is your simplest, sturdiest way to apply it.

Concretely, a semaphore is just a counter with a cap. You create it with a limit — say 10 — and each coroutine must acquire a slot before doing its work and release it after (Python's async with sem: does both for you automatically). When all 10 slots are taken, the eleventh coroutine's async with line simply pauses — yielding to the loop like any other await — until someone finishes and frees a slot. You can still submit all 1,000 items up front; the semaphore quietly guarantees only 10 are ever truly in flight. Submission and execution get decoupled, and you get a smooth, steady stream instead of a thundering herd.

The mistake this prevents is the most seductive one in all of async, because the broken version works perfectly in testing. On your laptop with 20 test items, a naked gather over everything is instant and flawless — so it ships. Then it meets 10,000 real inputs in production: it opens 10,000 sockets simultaneously, blows straight through the provider's requests-per-minute limit into a wall of 429 Too Many Requests, and balloons memory holding 10,000 half-finished responses until the kernel kills the process. Nothing in the happy-path code hinted at this. The fix isn't more hardware — it's the metering light. Pair the semaphore with retry-and-backoff on those 429s (next tip box) and you have the canonical, boring, bulletproof shape every serious LLM fan-out uses.

pythonimport asyncio

async def bounded_map(coro_fn, items, limit=10):
    sem = asyncio.Semaphore(limit)        # at most `limit` inside at once
    async def worker(item):
        async with sem:                   # acquire; blocks if `limit` are running
            return await coro_fn(item)
    return await asyncio.gather(*[worker(i) for i in items])

async def embed(text):
    await asyncio.sleep(0.2); return len(text)

# 1000 items but never more than 10 in flight -> respects rate limits
# asyncio.run(bounded_map(embed, ["doc"] * 1000, limit=10))

5 · Timeouts & cancellation advanced

Somewhere out there is a request that will never come back — and if you don't plan for it, it takes your whole service down with it. A network blip, an overloaded provider, a connection that half-opens and then goes silent: any of these can leave a coroutine awaiting a response that simply never arrives. Without a deadline, that one stuck call waits forever, and worse, everything depending on it waits forever too. Timeouts are how you refuse to wait forever, and they are not optional in anything real.

The everyday version of this is calling a support line and being told "please hold." You'll happily hold for two minutes; you will not hold for two hours. At some point you hang up and try another approach, because your time is worth more than one stuck call. A timeout gives your code that same self-respect: "I'll wait two seconds for this answer; if it hasn't come, I'm hanging up and moving on." asyncio.timeout(2) (Python 3.11+) — or the older wait_for — wraps a block of async work in exactly that deadline. Beat the clock and everything proceeds normally; overrun it and the wait is abandoned.

What actually happens on overrun is worth understanding, because it's the second half of this section's title: cancellation. When the timeout fires, asyncio doesn't just walk away and leave the stuck task running in the background — it cancels it. Cancellation works by raising a special CancelledError exception inside the paused coroutine, right at its await point. That's a deliberate design: because it surfaces as a normal exception, your coroutine gets a chance to clean up — close a file, release a connection, roll back — in a finally block before it unwinds. Cancellation is cooperative and tidy, not a violent kill.

The trap is treating a timeout as a failure to be swallowed and forgotten. The naive instinct is to wrap everything in try/except, catch the TimeoutError, log nothing, and carry on with a silently missing result — which turns a loud, diagnosable problem into a quiet data hole nobody notices for weeks. And a subtler error: catching the CancelledError yourself and suppressing it, which defeats the cancellation entirely and leaves a zombie task alive. The right posture is to catch the timeout deliberately — record it, decide whether to retry, return a sentinel — and to let CancelledError propagate after any cleanup. A timeout is information: it's your system telling you a dependency is unhealthy, and you want to hear it, not mute it.

pythonimport asyncio

async def slow():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        async with asyncio.timeout(2):      # give it 2 seconds
            await slow()
    except TimeoutError:
        print("gave up after 2s")          # task is cancelled for us

asyncio.run(main())

6 · Streaming LLM tokens with async for advanced

Two chat apps take the exact same eight seconds to write a paragraph. One feels broken; the other feels alive. The only difference is streaming. The first sits on a blank screen for eight seconds and then dumps the whole reply at once — an eternity of nothing that makes users reload the page. The second starts typing words within a few hundred milliseconds and keeps flowing. Same model, same total time, wildly different experience. That gap is why every LLM chat interface you've ever admired streams, and this section is how.

The insight is about perceived latency, not real latency. Think of a waiter who brings out each dish the moment it's plated — starter, then soup, then main — versus one who hides in the kitchen until the entire multi-course meal is ready and carries it all out on one enormous tray. The total cooking time is identical, but the first table is happily eating and chatting the whole while, and the second is staring at an empty table wondering if they've been forgotten. Streaming serves your user each token as it's cooked instead of making them wait for the full tray. It doesn't make the model faster; it makes the wait disappear because something useful is happening the entire time.

The Python mechanism that models "a sequence that arrives over time" is the async generator: an async def function that uses yield instead of return. Where a normal function hands back one value and is done, and a normal generator yields values on demand, an async generator yields values that each may involve awaiting — perfect for tokens trickling in over a network one at a time. You consume it with async for, a loop that pulls the next item the instant it's ready and runs the loop body immediately, without blocking the event loop between items. The pairing is exact: async generator produces, async for consumes, and the loop stays free to do other work in the gaps.

The mistake that quietly cancels all the benefit is collecting the whole stream before showing any of it. It's tempting to write tokens = [t async for t in stream()] and then print the joined result — and it runs, and it's wrong, because you've reintroduced the exact "wait for the full tray" delay you were trying to kill. Streaming only helps if you render each token inside the loop, as it arrives — hence the print(token, end="", flush=True) in the lab, where flush=True forces each piece to the screen instantly rather than sitting in a buffer. A useful pattern is to do both at once: render each token live and append it to a list, so the user sees the typewriter effect while you still assemble the complete final string for logging or further processing. Get that loop body right and you have the identical shape real Anthropic SDK streaming uses.

pythonimport asyncio

async def stream_completion(prompt):        # async generator
    for tok in ["The", " answer", " is", " 42."]:
        await asyncio.sleep(0.1)             # each token arrives over the wire
        yield tok

async def main():
    parts = []
    async for token in stream_completion("q"):   # consume as it streams
        print(token, end="", flush=True)         # render immediately
        parts.append(token)
    return "".join(parts)

asyncio.run(main())

7 · Async producer/consumer pipeline expert intermediate

So far every pattern has had a start and an end — fire 1,000 calls, collect 1,000 results, done. But real systems often have no end: a feed of documents to ingest, a stream of tasks that never stops arriving. When work flows in continuously and the thing producing it runs at a different speed than the thing processing it, you need a shape that lets both run at their own pace without one crushing the other. That shape is the producer/consumer pipeline, and its heart is a queue.

Picture a kitchen at dinner rush. Waiters (producers) keep clipping new order tickets onto the rail above the line; cooks (consumers) pull tickets off and cook them. The ticket rail is the queue. Its beauty is that waiters and cooks never have to look at each other or move in lockstep — a waiter clips a ticket and immediately goes back to the floor; a cook grabs the next ticket whenever they finish the last dish. The rail absorbs the mismatch in their rhythms. An asyncio.Queue is that rail: producers put items on one end, consumers get them off the other, and neither has to know anything about the other's speed or even how many of each there are.

The subtle, powerful part is what happens when you give the rail a maximum size. A bounded queue is a self-regulating system: when consumers fall behind and the rail fills up, a producer's await q.put(...) simply pauses until a consumer frees a slot. The fast side is automatically forced to wait for the slow side — that's backpressure again, the same principle as the semaphore in §4, but now emerging naturally from the queue's size limit rather than an explicit counter. Set maxsize and you've capped how much unprocessed work can pile up in memory, which is exactly what stops a fast ingest from OOM-ing your process while a slow embedding step catches up.

The classic bug in this pattern is telling the consumers when to stop — or rather, forgetting to. A consumer loops on await q.get() forever; nothing in "the queue is empty" tells it the work is truly finished versus just momentarily drained. The standard fix is a sentinel: after the real items, the producer puts a special "you're done now" marker (often None) — and here's the catch that bites everyone — one sentinel per consumer, because each consumer only consumes one before breaking. Three consumers need three sentinels; send two and one worker waits at an empty rail forever, hanging your program with no error. Count your sentinels to match your consumers and the pipeline shuts down cleanly. Get this right and you can chain stages — ingest → embed → index — each its own producer/consumer pool, which is how production RAG ingestion actually flows.

pythonimport asyncio

async def producer(q, n):
    for i in range(n):
        await q.put(i)               # blocks if queue is full -> backpressure
    for _ in range(3):
        await q.put(None)            # one sentinel per consumer to stop it

async def consumer(q, out):
    while True:
        item = await q.get()
        if item is None:
            break
        await asyncio.sleep(0.05)      # process (e.g. embed) the item
        out.append(item * 2)

async def main():
    q = asyncio.Queue(maxsize=10)      # bounded = backpressure
    out = []
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(q, 50))
        for _ in range(3):
            tg.create_task(consumer(q, out))   # 3 workers drain concurrently
    print(len(out))                  # 50

asyncio.run(main())

Exercises advanced

🎯 Interview practice interview

The interview questions this topic gets asked — worked, with code. For the full pattern catalog see A9 · Big Tech AI-engineering patterns.

pythonimport asyncio
async def call(x): await asyncio.sleep(0.2); return x*2
async def bounded(items, limit=10):
    sem = asyncio.Semaphore(limit)
    async def w(x):
        async with sem: return await call(x)
    return await asyncio.gather(*[w(i) for i in items])

pythonimport asyncio

async def stream(prompt):
    for tok in ["The", " answer", " is", " 42."]:
        await asyncio.sleep(0.05); yield tok
async def main():
    parts = []
    async for t in stream("q"):
        print(t, end="", flush=True); parts.append(t)
    return "".join(parts)

Checkpoint advanced

import asyncio, time

async def task(name, secs):
    await asyncio.sleep(secs)
    return f"{name} done"

async def main():
    t = time.perf_counter()
    res = await asyncio.gather(task("a", 0.2), task("b", 0.3))
    print(res, "in %.2fs" % (time.perf_counter()-t))   # ~0.30s, not 0.50

asyncio.run(main())

import asyncio

async def worker(i, sem, state):
    async with sem:
        state["cur"] += 1
        state["peak"] = max(state["peak"], state["cur"])
        await asyncio.sleep(0.05)
        state["cur"] -= 1
        return i

async def main():
    sem = asyncio.Semaphore(3)
    state = {"cur": 0, "peak": 0}
    await asyncio.gather(*(worker(i, sem, state) for i in range(10)))
    print("peak concurrency:", state["peak"])   # 3

asyncio.run(main())

import asyncio

cancelled = {"hit": False}

async def slow():
    try:
        await asyncio.sleep(1.0)
        return "finished"
    except asyncio.CancelledError:
        cancelled["hit"] = True
        raise

async def main():
    try:
        await asyncio.wait_for(slow(), timeout=0.1)
    except asyncio.TimeoutError:
        print("timed out; cancelled =", cancelled["hit"])   # True

asyncio.run(main())

import asyncio

async def stream_tokens(prompt):
    # real client yields deltas from the network; here we simulate
    for tok in ["The", " GIL", " serial", "izes", " bytecode", "."]:
        await asyncio.sleep(0.02)
        yield tok

async def main():
    out = []
    async for tok in stream_tokens("explain the GIL"):
        out.append(tok)
        print("recv:", repr(tok))     # arrives one at a time
    print("final:", "".join(out))

asyncio.run(main())

import asyncio

async def producer(q, n):
    for i in range(n):
        await q.put(i)          # blocks when queue is full -> backpressure
    await q.put(None)           # sentinel

async def consumer(q, out):
    while True:
        item = await q.get()
        if item is None:
            await q.put(None)   # let siblings see the sentinel too
            return
        await asyncio.sleep(0.01)
        out.append(item * item)

async def main():
    q = asyncio.Queue(maxsize=2)
    out = []
    await asyncio.gather(producer(q, 6), consumer(q, out), consumer(q, out))
    print(sorted(out))          # [0, 1, 4, 9, 16, 25]

asyncio.run(main())

import asyncio, random

async def call_llm(text):
    # real: await client.messages.create(...)  -- needs an API key to run
    await asyncio.sleep(0.01)
    if random.random() < 0.1:
        raise RuntimeError("transient 503")
    return text.upper()

async def one(text, sem, retries=3, timeout=2.0):
    async with sem:                       # bound concurrency
        for attempt in range(retries):
            try:
                val = await asyncio.wait_for(call_llm(text), timeout)
                return (text, True, val)
            except (asyncio.TimeoutError, RuntimeError):
                await asyncio.sleep(0.02 * 2**attempt)   # backoff
        return (text, False, "failed after retries")

async def fan_out(inputs, concurrency=50):
    sem = asyncio.Semaphore(concurrency)
    return await asyncio.gather(*(one(t, sem) for t in inputs))

async def main():
    res = await fan_out([f"row{i}" for i in range(20)], concurrency=5)
    ok = sum(1 for _, good, _ in res if good)
    print(f"{ok}/{len(res)} succeeded; failures isolated")

asyncio.run(main())

Knowledge check check yourself