Async and concurrency

Calling a model API is mostly waiting. Make a hundred calls one after another and your program spends nearly its whole life doing nothing, so the fix is to have many in flight at once. This section covers how Python expresses that, how to start many and collect the results, how to cap the number running so you do not trip a rate limit, and the one mistake that quietly turns concurrent code back into sequential code.

The problem async solves#

A call to a model API takes a few seconds, and your program spends nearly all of that time doing nothing but waiting for the reply. If you make fifty calls one after another, you wait fifty times.

# 50 calls, 2 seconds each, about 100 seconds total
for doc in docs:
    summarise(doc)

Async lets you start all fifty and wait for them together. The total becomes roughly the time of one call. This is the single biggest speed improvement available in most code that talks to an API.

The syntax#

Mark a function async def, and use await when calling something that takes time. A function marked async does not run on its own, it has to be started by asyncio.run or awaited from another async function.

import asyncio

async def summarise(doc: str) -> str:
    return await call_model(doc)

async def main():
    result = await summarise("some text")
    print(result)

asyncio.run(main())          # starts everything

Running many at once#

gather starts everything together and waits for all of it. Results come back in the order you passed them in.

async def main(docs):
    results = await asyncio.gather(*(summarise(d) for d in docs))
    return results

By default, if one of them raises, gather cancels the rest and raises. In a batch job that is usually wrong: one bad document should not lose the other forty-nine. Pass return_exceptions=True and the failures come back as values you can inspect.

results = await asyncio.gather(*tasks, return_exceptions=True)

for doc, result in zip(docs, results):
    if isinstance(result, Exception):
        log(f"failed: {doc}", result)
    else:
        save(result)

Not too many at once#

Firing five hundred requests simultaneously will get you rate limited. A semaphore is a counter that only lets a fixed number through at a time, and queues the rest.

async def run_all(docs, limit=8):
    sem = asyncio.Semaphore(limit)

    async def one(doc):
        async with sem:                       # waits for a free slot
            async with asyncio.timeout(60):   # give up after 60s
                return await summarise(doc)

    return await asyncio.gather(*(one(d) for d in docs), return_exceptions=True)

Put a timeout on anything that touches the network. Without one, a single stalled connection holds a slot forever and the job never finishes.

The mistake that undoes everything#

import requests, httpx, asyncio

# Blocks the whole event loop
async def bad(url):
    return requests.get(url).text

# Correct: an async client
async def good(url):
    async with httpx.AsyncClient() as client:
        r = await client.get(url)
        return r.text

# When you have no async option
async def fallback(path):
    return await asyncio.to_thread(heavy_sync_read, path)

When async does not help#

Async only helps when you are waiting. If the work is calculation rather than waiting, there is still only one thread doing all of it, so async adds complexity and speeds up nothing. For heavy computation you want multiprocessing instead.