High-Throughput Async & Streaming Pipelines

This is the smallest complete async program — the shape every later example follows. async code doesn't run by itself; something has to start the event loop (the scheduler that runs your tasks and decides who waits and who runs). asyncio.run(...) is that starter.

1. import asyncio brings in Python's built-in async toolkit — the event loop, sleep, gather, and the rest.
2. async def main(): defines a coroutine function. The word async is what makes it special: it is allowed to contain await, and calling it hands back a coroutine to run later rather than running now.
3. await asyncio.sleep(1) is the key line. Instead of freezing the program for a second, await yields control back to the event loop, which is free to run other tasks meanwhile. When the second is up, the loop resumes right here.
4. asyncio.run(main()) creates the event loop, runs main() until it finishes, then closes the loop. This is the one place a normal (non-async) program crosses into async.

What the output means: Prints start, waits about one second, then prints done. In a real program that one second of waiting is where dozens of other tasks could be making progress.

Try this: Change asyncio.sleep(1) to time.sleep(1) (after import time). It still works here, but time.sleep blocks the whole loop — the mistake the next warning box is about.

This shows the difference between defining async work, awaiting it, and the classic beginner bug: creating a coroutine but forgetting to await it, so it never runs.

1. async def complete(prompt): is a fake LLM call. The await asyncio.sleep(0.5) stands in for the half-second you'd really spend waiting on the network.
2. result = await complete("hello") is the normal path: await runs the coroutine, pauses main while it waits, and hands back the returned string.
3. coro = complete("oops") only creates a coroutine object — it does not run. Without an await in front, the work inside never happens. Python would warn "coroutine was never awaited"; coro.close() just tidies it up to silence that warning.

What the output means: Prints answer to: hello. The "oops" call produces nothing because it was never awaited — that silence is the bug.

Try this: Delete the coro.close() line and run it — Python prints a RuntimeWarning: coroutine 'complete' was never awaited. Recognising that warning saves hours later.

Here is the payoff of async: running many calls at the same time. One hundred calls that each wait one second finish in about one second total, not one hundred, because all the waiting overlaps. Two ways to do it are shown.

1. [complete(i) for i in range(100)] builds a list of 100 coroutines (still not running yet). The * in gather(*[...]) spreads that list out as separate arguments.
2. await asyncio.gather(...) launches all 100 at once and waits for every one to finish, returning their results as a list in the order you submitted them — handy when you need to line results back up with inputs.
3. async with asyncio.TaskGroup() as tg: is the newer (Python 3.11+) way. tg.create_task(...) schedules each call; when the async with block exits, all tasks are guaranteed done. If one fails, the group cancels the rest and reports the error cleanly — safer than gather for production.
4. [t.result() for t in tasks] then pulls each finished task's return value.

What the output means: len(results) is 100, and it appears after roughly one second — proof the calls overlapped instead of running one after another.

Try this: Wrap one complete call so it raises an error, and compare: gather surfaces just that one exception, while TaskGroup cancels the siblings and bundles the errors. That difference is why TaskGroup is preferred.

Launching 1000 calls at once (as in §3) would blow past the API's rate limit and your memory. A semaphore is a counter that only lets limit tasks run at a time; the rest wait their turn. This "backpressure" is the single most important production async pattern.

1. sem = asyncio.Semaphore(limit) creates a gate that allows limit holders (default 10) inside simultaneously.
2. async with sem: is how a worker acquires a slot. If limit workers already hold the gate, this line pauses until one of them exits the block and frees a slot — automatically throttling everything.
3. return await coro_fn(item) does the real work only while inside the gate, so no more than limit real calls are ever in flight.
4. return await asyncio.gather(*[worker(i) for i in items]) still submits all the items — but because each goes through the semaphore, only limit run at once.

What the output means: Nothing prints (the run line is commented out), but the design guarantees: 1000 items submitted, never more than 10 hitting the API at any instant.

Try this: Change limit to 1 and it becomes fully sequential; raise it and throughput climbs — until you hit the provider's rate limit. Tuning that number is real production work.

A single request that hangs forever must not freeze your whole service. A timeout puts a ceiling on how long you'll wait; if the work overruns, async cancels it for you.

1. async def slow(): simulates a stuck call by sleeping ten seconds.
2. async with asyncio.timeout(2): (Python 3.11+) starts a two-second clock around everything inside the block. If the block isn't done in time, the task is cancelled.
3. except TimeoutError: catches that expiry so you can react — here it just prints a message. Without the try/except, the timeout would bubble up as an error.

What the output means: After about two seconds (not ten), it prints gave up after 2s — the slow() call was cancelled the moment the clock ran out.

Try this: Lower asyncio.sleep(10) to 1 so the work finishes before the 2-second limit — now the except never fires. That's how you confirm the timeout only triggers on genuine overruns.

Streaming makes a slow answer feel fast: instead of waiting for the whole reply, you show each piece (token) the instant the model produces it. An async generator models the stream, and async for consumes it.

1. async def stream_completion(prompt): plus yield tok makes an async generator — a coroutine that hands back values one at a time instead of returning once. Each await asyncio.sleep(0.1) mimics the delay between tokens on the network.
2. async for token in stream_completion("q"): is the streaming loop: it pulls the next token as soon as it's ready and runs the loop body immediately — no waiting for the full message.
3. print(token, end="", flush=True) renders each token right away. flush=True forces it to the screen instantly so the user sees text appear live; parts.append(token) also collects them to rebuild the full answer at the end.

What the output means: You see The answer is 42. appear a word at a time rather than all at once — exactly the typing effect of a chat UI.

Try this: Raise the sleep to 0.5 to exaggerate the streaming, then set it to 0 so it prints instantly — same code, different perceived speed. Real Anthropic SDK streaming uses this identical async for shape.

For never-ending work (a feed of documents or tasks), a queue lets fast producers and slow consumers run independently. A bounded queue also gives free backpressure: if consumers fall behind, producers are forced to wait.

1. async def producer(q, n): puts n items on the queue with await q.put(i). If the queue is full, put pauses until a consumer removes something — that's the backpressure.
2. After the real items, it puts three None values — one sentinel ('stop now') for each of the three consumers, so every worker knows when to quit.
3. async def consumer(q, out): loops forever with await q.get(), breaks when it sees the None sentinel, and otherwise 'processes' the item (the sleep stands in for real work like embedding).
4. q = asyncio.Queue(maxsize=10) bounds the queue. The TaskGroup runs one producer and three consumers together; the block exits only when all are finished.

What the output means: Prints 50 — every one of the 50 produced items was drained and processed by the three consumers working in parallel.

Try this: Set maxsize=1 and watch throughput drop — the producer now stalls constantly. Then forget to send enough sentinels (range(2) instead of 3) and one consumer hangs forever waiting for a stop signal.

The classic interview question: make 1000 API calls without hitting rate limits. The expected answer is this exact pattern — a semaphore to cap concurrency, then gather to collect results. It's §3 and §4 compressed into the shape interviewers look for.

1. call(x) is the stand-in API call (waits, then returns x*2).
2. sem = asyncio.Semaphore(limit) caps how many calls run at once; the inner w(x) wraps each call in async with sem: so it must claim a slot first.
3. await asyncio.gather(*[w(i) for i in items]) submits every item but lets only limit proceed at a time — safe fan-out over a huge list.

Try this: Say the two knobs out loud in an interview: the semaphore bounds concurrency, and you'd pair it with retry-and-backoff on HTTP 429 to fully respect rate limits. Naming both is what earns the check-mark.

The streaming interview question: consume a token stream and render it live. This is §6 boiled down — an async generator plus an async for loop.

1. async def stream(prompt): with yield tok is the async generator producing tokens with a tiny delay each.
2. async for t in stream("q"): pulls each token the moment it's ready and prints it with flush=True so it shows immediately.
3. parts.append(t) then "".join(parts) reassembles the full answer, so you get both the live typing effect and the complete final string.

Try this: The interview point to make: streaming improves perceived latency (first token appears fast) even though total time is the same — and the consuming code is just an async for loop.