Async Python for LLM apps

Why async matters for AI applications

Imagine your model takes 5 seconds to answer. If your server handles requests one at a time, the second user waits 5 seconds for the first user's answer to finish and then 5 more for their own, and the hundredth user waits more than 8 minutes. That is not a slow model, that is a server that sits idle while it waits.

Most of what an AI application does is waiting: for the model, for the vector database, for a web search, for a tool's API. During that waiting, the CPU is doing nothing. Asynchronous code lets the program use that time, so while one request waits for the model, another request can be retrieving documents, and a third can be sending its result back to the user.

The same idea works inside one request. If an agent needs to search the web, query a database and fetch a document, and none of those depends on the others, there is no reason to do them one after another.

Blocking and non-blocking code

The clearest way to see the difference is with a function that pretends to be a slow model call.

import time


def slow_task() -> str:
    time.sleep(5)
    return "done"


print("starting")
result = slow_task()
print(result)

time.sleep(5) blocks. For 5 seconds, the whole program stops, and nothing else can happen.

Here is the same idea written asynchronously.

import asyncio


async def slow_task() -> str:
    await asyncio.sleep(5)
    return "done"


async def main() -> None:
    print("starting")
    task = asyncio.create_task(slow_task())
    await asyncio.sleep(1)
    print("doing other work")
    result = await task
    print(result)


asyncio.run(main())

This prints "starting", then "doing other work" after 1 second, and "done" after 5 seconds in total. create_task starts the slow task in the background, and await is the point where a function says "I am waiting, so give control to something else until my result is ready." That handing over of control is the whole idea of async.

async and await, step by step

There are four things to understand.

• async def defines a coroutine, a function that can pause at an await and resume later.
• await pauses the current coroutine until the thing it waits for is ready, and lets other coroutines run in the meantime.
• The event loop is the scheduler that decides which coroutine runs next. asyncio.run(main()) starts it.
• Calling an async function doesn't run it. It creates a coroutine object, and it only runs when you await it or turn it into a task.

The most common beginner mistake is forgetting the await. The code doesn't crash, it just never runs the call, and Python warns that a coroutine was never awaited.

Running many calls at the same time

asyncio.gather runs several coroutines concurrently and returns their results in the same order you passed them.

import asyncio
import random


async def fake_llm(prompt: str) -> str:
    await asyncio.sleep(random.uniform(1, 2))
    return f"answer to: {prompt}"


async def main() -> None:
    prompts = ["summarise doc 1", "summarise doc 2", "summarise doc 3"]
    answers = await asyncio.gather(*(fake_llm(p) for p in prompts))
    print(answers)


asyncio.run(main())

Three calls that each take 1 to 2 seconds finish in about 2 seconds together, instead of 3 to 6 seconds one after another.

By default, if one call raises an error, gather raises it and you lose the other results. Pass return_exceptions=True to get the errors back as values alongside the successful results, and decide what to do with each one.

Limit concurrency with a semaphore

If you have 5000 documents to summarise, you can't send 5000 requests at the same moment. You will hit the provider's rate limit immediately, and the 429 errors will slow everything down more than sending them at a steady pace.

A semaphore limits how many coroutines can be inside a block at the same time.

import asyncio


async def summarise(doc: str, limit: asyncio.Semaphore) -> str:
    async with limit:
        await asyncio.sleep(1)
        return doc.upper()


async def main() -> None:
    docs = [f"doc {i}" for i in range(50)]
    limit = asyncio.Semaphore(10)
    results = await asyncio.gather(*(summarise(d, limit) for d in docs))
    print(len(results))


asyncio.run(main())

At most 10 calls run at any moment, and the other 40 wait their turn. Start with a small number, watch your rate limit errors and latency in your logs, and increase it slowly.

Timeouts in async code

Every awaited call that goes over the network needs a timeout, for the same reasons as on the API page. Use the SDK's own timeout setting where it has one, and wrap other calls with asyncio.wait_for or, in Python 3.11 and later, asyncio.timeout.

import asyncio


async def slow_tool() -> str:
    await asyncio.sleep(10)
    return "result"


async def main() -> None:
    try:
        result = await asyncio.wait_for(slow_tool(), timeout=3)
    except TimeoutError:
        result = "tool timed out"
    print(result)


asyncio.run(main())

In an agent, give each tool call its own timeout and give the whole run a total time budget, so one slow tool can't hold the user forever.

Async clients

Async only helps if the library you call supports it. If you call a normal blocking function inside an async function, it still blocks the whole event loop.

Most model SDKs ship an async client next to the normal one. The OpenAI SDK has AsyncOpenAI, the Anthropic SDK has AsyncAnthropic, and Google's Gen AI SDK exposes async methods under client.aio. For plain HTTP calls, use httpx.AsyncClient instead of requests, because requests is blocking.

import os

from google import genai


async def ask(question: str) -> str:
    client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    response = await client.aio.models.generate_content(
        model=os.environ.get("MODEL_NAME", "gemini-2.5-flash-lite"),
        contents=question,
    )
    return response.text

Async in FastAPI

FastAPI supports both normal and async endpoints, and the difference matters.

If you write async def, FastAPI runs your function on the event loop, so inside it you must only use async calls. A blocking call like time.sleep, requests.get or a synchronous SDK call inside an async def endpoint freezes the event loop, and every other request waits.

If you write a normal def, FastAPI runs it in a thread pool, so blocking code is fine there, but you can handle fewer requests at once.

The rule is simple: use async def with async libraries, and plain def when you have to use blocking libraries.

import asyncio

from fastapi import FastAPI

app = FastAPI()


async def retrieve(question: str) -> list[str]:
    await asyncio.sleep(0.3)
    return ["chunk about refunds"]


async def search_web(question: str) -> list[str]:
    await asyncio.sleep(0.5)
    return ["page about refunds"]


@app.get("/answer")
async def answer(q: str) -> dict:
    chunks, pages = await asyncio.gather(retrieve(q), search_web(q))
    return {"question": q, "context": chunks + pages}

WebSockets for two-way, real-time communication

A normal HTTP request is one question and one answer. Some AI applications need a connection that stays open in both directions: a voice assistant that streams audio both ways, a chat where the user can interrupt the model, or a dashboard that receives updates while a long job runs.

A WebSocket keeps a two-way connection open between the browser and the server, so either side can send messages at any time, and async code is what lets one server hold many of these connections at once.

from fastapi import FastAPI, WebSocket

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    while True:
        message = await websocket.receive_text()
        await websocket.send_text(f"echo: {message}")

For simple token streaming from a model, server-sent events over normal HTTP are usually enough. Use WebSockets when the client also needs to send messages while the response is still coming.

Async, threads and processes

Async is one of three ways to do more than one thing at a time in Python, and each fits a different kind of work.

Approach	Best for	Example in AI work
Async (asyncio)	Many tasks that mostly wait on the network	LLM calls, vector DB queries, web requests
Threads	Waiting work with libraries that are not async	Calling a blocking SDK from async code with asyncio.to_thread
Processes	Heavy CPU work	Parsing thousands of PDFs, computing embeddings locally

If your work is waiting, use async. If your work is computing, async doesn't help, because the CPU is busy the whole time, and you need multiple processes or a job queue with separate workers.

When to use a job queue instead

Some work shouldn't happen inside a web request at all. If indexing a large document set takes 20 minutes, the user shouldn't wait for it, and your API shouldn't hold a request open for 20 minutes.

For that, the API accepts the job, puts it on a queue, returns a job ID immediately, and separate workers process the queue. The user checks the status later, or gets a notification when it's done. Async makes each worker efficient, and the queue makes the system reliable.

From the floor

In the first hands-on session of AI Engineer HQ cohort 1, after containerization, we moved to async Python. I explained it with the inference path of an LLM application: a query comes in, the system checks the cache, tokenizes the input and sends it to the model, and while the model is generating, which takes time, the system shouldn't stop, it should be searching the web, getting citations or preparing the next step in parallel. Then we wrote the blocking version with time.sleep, the async version with create_task, and a FastAPI WebSocket server, so builders could see the difference in their own terminal.

Related

• Working with LLM APIs
• The LLM inference pipeline
• LLMOps

Sources

• asyncio, Python documentation
• Concurrency and async / await, FastAPI documentation
• WebSockets, FastAPI documentation