Iterators, generators, and async (how LLM streaming and concurrency actually work)

Two clusters live in this chapter, and you will meet both on your first day of reading real LLM code. Lazy evaluation explains why for chunk in stream: gives you tokens one at a time instead of one giant string. Async explains how a single agent fires 50 model calls without spawning 50 threads. Learn both and most GenAI code stops looking like magic.

Part A — Lazy evaluation

Iterables vs iterators

Python draws a line JavaScript blurs. An iterable is anything you can loop over (a list, a string, a file). An iterator is the one-shot cursor that actually walks through it, remembering where it is.

nums = [10, 20, 30]          # a list — iterable, but NOT an iterator
it = iter(nums)               # iter() asks the iterable for a fresh iterator
print(next(it))               # => 10
print(next(it))               # => 20
print(next(it))               # => 30
print(next(it, "DONE"))       # => DONE   (2nd arg = default instead of raising)

Under the hood two dunder methods define the protocol:

• __iter__(self) returns an iterator (a list returns a new cursor each time).
• __next__(self) returns the next value, or raises StopIteration when exhausted.

So this for loop:

for n in nums:
    print(n)

is exactly sugar for: call iter(nums) once, then call next() repeatedly, stopping silently when StopIteration fires. That is the whole mechanism. StopIteration is a normal exception the for statement catches for you — it is a signal, not an error.

JS → Python: This is JS's iteration protocol with renamed parts. Symbol.iterator ↔ __iter__; .next() returning {value, done} ↔ __next__() returning a value or raising StopIteration. JS signals completion with a done: true flag; Python signals it by raising. for…of ↔ for…in (note: Python's for…in is the value loop, not JS's key loop).

Generators: yield

Writing __iter__/__next__ by hand is tedious. A generator function — any def containing yield — builds the iterator for you. Calling it runs no body code; it hands back a generator object. Each next() runs until the next yield, pauses there (freezing all local state), and resumes on the following next().

def countdown(n: int):
    while n > 0:
        yield n          # pause here, hand out n, remember everything
        n -= 1

gen = countdown(3)
print(next(gen))         # => 3
print(list(gen))         # => [2, 1]   (drains the rest)

The payoff is laziness and memory. A generator holds one item at a time, so you can process data far larger than RAM. Here is the pattern you will see everywhere — streaming pages from a paginated API without ever building the full list:

def fetch_all_documents(client, batch_size: int = 100):
    """Yield documents one page at a time — memory stays flat."""
    cursor = 0
    while True:
        page = client.get_documents(offset=cursor, limit=batch_size)
        if not page:                 # empty page = we're done
            return                   # bare `return` ends a generator
        for doc in page:
            yield doc                # caller sees a flat stream of docs
        cursor += len(page)

# You loop as if it's one endless list; only ~100 rows exist in memory at once.
# for doc in fetch_all_documents(client):
#     embed_and_store(doc)

Generator expressions are comprehensions with () instead of []. A list comprehension builds everything now; a generator expression builds nothing until iterated.

squares_list = [x * x for x in range(1_000_000)]   # ~8 MB allocated immediately
squares_gen  = (x * x for x in range(1_000_000))   # near-zero memory, lazy
print(sum(x * x for x in range(5)))                # => 30  (no [] needed inside a call)

yield from delegates to a sub-iterable, flattening one generator into another:

def chain_streams(*streams):
    for s in streams:
        yield from s                 # re-yield every item of s
print(list(chain_streams([1, 2], [3, 4])))   # => [1, 2, 3, 4]

And generators can be infinite — legal precisely because they are lazy:

def ids():
    n = 0
    while True:
        yield n
        n += 1
# Safe only if the consumer stops pulling (e.g. with itertools.islice).

The key idea: token streaming from an LLM is an iterator. When you write for chunk in stream: against a chat completion, the SDK is a generator handing you deltas as bytes arrive over the wire — each yield is one token-ish chunk. You never hold the full response; you print pieces as they land, which is why streamed replies feel live.

# The shape you'll read in nearly every sync LLM streaming example:
# stream = client.chat.completions.create(model="...", messages=[...], stream=True)
# for chunk in stream:                       # <-- iterator protocol, nothing new
#     delta = chunk.choices[0].delta.content
#     if delta:
#         print(delta, end="", flush=True)   # print tokens as they arrive

itertools at read level

The standard itertools module supplies lazy iterator tools. You mostly need to recognize three:

• islice(it, n) — take the first n items of any iterator (the safe way to sip an infinite generator).
• chain(a, b, ...) — concatenate iterables lazily (the function form of yield from).
• tee(it, k) — split one iterator into k independent ones. Handy but it buffers whatever the slowest branch hasn't consumed.

from itertools import islice, chain
print(list(islice(ids(), 4)))          # => [0, 1, 2, 3]
print(list(chain("ab", "cd")))         # => ['a', 'b', 'c', 'd']

Reading & judging: list(some_generator) at the top of a function quietly cancels the laziness — the whole stream is now in memory. That is fine for 100 rows, a bug for 10 million. If you see a generator immediately wrapped in list() or sorted() right before a for, ask whether streaming was the point. Also watch for tee on an unbounded stream: it can grow memory without limit if one branch lags.

Try it: Write def take(it, n) that yields the first n items of any iterator using only for/yield/break — no itertools. Confirm list(take(countdown(100), 3)) == [100, 99, 98] and that it never exhausts the source.

Part B — Async Python

The why: I/O-bound work

An LLM call spends ~2 seconds doing nothing locally — it sent bytes and is waiting for the model to respond. HTTP requests, database queries, file reads are all like this: I/O-bound. Blocking one at a time wastes the wait. You could use threads, but Python's GIL (Global Interpreter Lock) means only one thread runs Python bytecode at a time, so threads help I/O yet add locking headaches.

Async takes a different route: cooperative concurrency on a single thread. One event loop runs your coroutines; whenever a coroutine hits an await on I/O, it voluntarily steps aside so the loop can run another coroutine, resuming the first when its data is ready. Crucially, async is not parallelism — no two coroutines run Python at the same instant. It is one worker juggling many waits efficiently. For LLM/agent code, which is almost entirely "send request, wait, send another," this is exactly the right tool.

JS → Python: You already own this mental model. Python's event loop is Node's event loop. A coroutine (from async def) is a Promise/awaitable. await is await. asyncio.run(main()) is the top-level bootstrap Node does for you. The GIL-vs-single-thread story is basically Node's single-threaded event loop with the same "don't block me" rule.

async def, await, and the top beginner bug

import asyncio

async def get_answer(prompt: str) -> str:
    await asyncio.sleep(1)            # pretend this is an LLM call (non-blocking wait)
    return f"answer to: {prompt}"

# THE classic mistake:
maybe = get_answer("hi")
print(maybe)     # => <coroutine object get_answer at 0x...>   ← NOT the result!

Calling a coroutine function does not run it — it returns a coroutine object, inert until awaited or driven by the loop. (Python even warns coroutine '...' was never awaited if you drop it.) You must either await it inside another coroutine or hand it to asyncio.run at the top:

async def main():
    result = await get_answer("hi")   # now it actually runs
    print(result)                     # => answer to: hi

asyncio.run(main())                   # the ONE sync entry point that starts the loop

JS → Python: Same trap, sharper edge. In JS, calling an async function starts it and gives you a pending Promise; forgetting await often still runs the work. In Python the coroutine is completely inert until awaited — forget the await and nothing happens at all, silently. That silent no-op is the single most common async bug in GenAI code.

Running things concurrently

Awaiting in a plain loop is sequential — each await finishes before the next starts. To overlap the waits, schedule first, then await together.

import asyncio, time

async def call_llm(prompt: str) -> str:
    await asyncio.sleep(1)            # each "call" takes ~1s of waiting
    return prompt.upper()

async def sequential():
    t = time.perf_counter()
    out = []
    for p in ["a", "b", "c"]:
        out.append(await call_llm(p))    # wait fully before next — 3 x 1s
    print(out, f"{time.perf_counter() - t:.1f}s")
    # => ['A', 'B', 'C'] 3.0s

async def concurrent():
    t = time.perf_counter()
    out = await asyncio.gather(          # fire all three, wait for all — overlapped
        call_llm("a"), call_llm("b"), call_llm("c"),
    )
    print(out, f"{time.perf_counter() - t:.1f}s")
    # => ['A', 'B', 'C'] 1.0s

asyncio.run(sequential())
asyncio.run(concurrent())

asyncio.gather(*coros) runs them concurrently and returns results in argument order (not completion order). This 3×-to-N× speedup is the biggest real-world async win in GenAI code: fanning out many independent model calls.

JS → Python: asyncio.gather(a, b, c) ↔ Promise.all([a, b, c]) — same idea, same ordered-results guarantee. asyncio.create_task(coro) ↔ eagerly starting a Promise so it runs in the background while you do other work.

Modern Python (3.11+) prefers asyncio.TaskGroup over bare gather for structured concurrency — if one task raises, the group cancels the siblings and propagates the error cleanly:

async def fan_out(prompts: list[str]) -> list[str]:
    async with asyncio.TaskGroup() as tg:          # 3.11+
        tasks = [tg.create_task(call_llm(p)) for p in prompts]
    # on exiting the `async with`, all tasks are done (or the group raised)
    return [t.result() for t in tasks]

print(asyncio.run(fan_out(["x", "y"])))            # => ['X', 'Y']

Add a timeout so a hung API call cannot stall forever (asyncio.timeout is 3.11+):

async def guarded():
    try:
        async with asyncio.timeout(0.5):           # budget 0.5s
            return await call_llm("slow")          # needs ~1s → will blow the budget
    except TimeoutError:
        return "timed out"

print(asyncio.run(guarded()))                      # => timed out

asyncio.sleep(n) is the async-friendly pause — it yields to the loop, unlike time.sleep which freezes everything (more on that below).

Async iteration & streaming

Two async cousins of Part A power streaming LLM code:

• async for iterates an async iterator — each step may await.
• async with is an async context manager — setup/teardown that can await (opening/closing an HTTP connection).
• An async generator is a coroutine with yield: it produces values and can await between them.

async def token_stream(text: str):
    """Simulate an LLM streaming tokens over the network."""
    for word in text.split():
        await asyncio.sleep(0.1)         # network delay per token
        yield word                       # async generator: await + yield together

async def consume():
    async for tok in token_stream("streaming is just async iteration"):
        print(tok, end=" ", flush=True)  # => streaming is just async iteration
    print()

asyncio.run(consume())

This is precisely the async streaming shape in real SDKs — async for chunk in stream: where each chunk is a token delta arriving over the wire. Combine both wins — stream each response, but fan out N prompts concurrently:

async def summarize(prompt: str) -> str:
    pieces = []
    async for tok in token_stream(f"summary of {prompt}"):
        pieces.append(tok)
    return " ".join(pieces)

async def batch():
    return await asyncio.gather(
        summarize("doc1"), summarize("doc2"), summarize("doc3"),
    )                                    # 3 streamed summaries, all in flight at once

print(asyncio.run(batch()))
# => ['summary of doc1', 'summary of doc2', 'summary of doc3']

JS → Python: async for … in ↔ JS for await…of; an async generator (async def + yield) ↔ JS async function* with yield. The correspondence is nearly one-to-one — if you can read async iterators in Node, you can read them here.

Sync vs async SDK clients

Most GenAI libraries ship two clients, and spotting which one you are reading tells you the whole execution model:

Sync (blocking)	Async (awaitable)
OpenAI()	AsyncOpenAI()
Anthropic()	AsyncAnthropic()
httpx.Client()	httpx.AsyncClient()
client.chat.completions.create(...)	await client.chat.completions.create(...)
for chunk in stream:	async for chunk in stream:

Rule of thumb: an Async… class name, an await before the call, or async for/async with = async code that belongs inside asyncio.run. No await and a plain for chunk in stream: = sync code that blocks the current thread until done. The API surface is deliberately mirror-imaged, so the only reliable tell is the async/await keywords, not the method names.

Pitfalls

The event loop is a single cooperative thread, so anything that does not await hogs it:

import time

async def bad():
    time.sleep(2)          # ❌ BLOCKS the whole loop for 2s — nothing else runs
    # requests.get(url)    # ❌ sync HTTP also freezes every other coroutine
    # sum(range(10**8))    # ❌ heavy CPU: no await point, loop is stuck

async def good():
    await asyncio.sleep(2)               # ✅ yields; other coroutines run meanwhile
    # await httpx.AsyncClient().get(url) # ✅ async HTTP
    # await asyncio.to_thread(cpu_heavy) # ✅ offload blocking/CPU work to a thread

asyncio.to_thread(fn, *args) (3.9+) pushes a blocking function onto a worker thread and gives you an awaitable — the escape hatch when you must call a sync library from async code.

Reading & judging: Five red flags when reviewing async GenAI code:

• Un-awaited coroutine — a client.create(...) with no await. Silent no-op; the call never happens.
• Blocking the loop — time.sleep, requests/sync httpx, or a big CPU loop inside async def. One offender freezes every concurrent request. Fix: async equivalents or asyncio.to_thread.
• Sequential awaits that should be gathered — results = [await call(p) for p in prompts]. Each waits fully before the next; a 50-item loop is 50× too slow. Should be await asyncio.gather(*(call(p) for p in prompts)).
• Materializing where streaming was the point — collecting a whole async for into a list only to return it, when the caller could stream.
• Unbounded concurrency — gather over 10 000 prompts with no asyncio.Semaphore. This hammers the API, trips rate limits, and often gets you 429s. Production code caps it: sem = asyncio.Semaphore(10) and async with sem: inside each task.

# The rate-limit-safe fan-out pattern you WANT to see:
async def bounded_call(sem, prompt):
    async with sem:                      # at most N in flight at once
        return await call_llm(prompt)

async def safe_batch(prompts):
    sem = asyncio.Semaphore(5)           # cap concurrency at 5
    return await asyncio.gather(*(bounded_call(sem, p) for p in prompts))

Try it: Take a list of 8 prompts and time three versions: a sequential for … await loop, a raw asyncio.gather, and a Semaphore(3)-bounded gather. With ~1s fake latency you should see roughly 8s, ~1s, and ~3s — proving the perf bug (sequential awaits) and why unbounded concurrency, while fastest, is the one that will get you rate-limited in production.

Between lazy iterators (streaming tokens) and async concurrency (fanning out calls), you now hold the two threads that run through virtually every LLM and agent codebase you will read.

---