Async / Await (asyncio)

Essential Python Skills for AI Engineers (added — needed in practice, not in original outline)

How This Lesson Fits the Module

Many AI workloads are I/O-bound: waiting on LLM APIs, vector databases, file reads, and webhooks. Running these sequentially wastes time. Asyncio with async/await lets one thread interleave many concurrent network operations—essential for responsive agents, batch embedding jobs, and FastAPI services.

Async does not speed up CPU-bound model training (use GPUs and batching for that). It shines when your code spends most of its time waiting on external systems.

Learning Objectives

By the end of this lesson, students should be able to:

Distinguish synchronous blocking code from asynchronous concurrent code.
Define coroutines with async def and await them with await.
Run multiple coroutines with asyncio.gather().
Understand when asyncio helps (I/O-bound) versus when it does not (CPU-bound).
Recognize async patterns in FastAPI route handlers and HTTP clients (httpx).
Avoid calling blocking code directly inside async functions without offloading.

Introduction: Concurrency Without Threads

A normal function runs start to finish, blocking while waiting for a network response. An async function (coroutine) can pause at await, letting the event loop run other coroutines until the I/O completes.

Core Syntax

async def — defines a coroutine function
await — pauses until the awaited coroutine or I/O completes
asyncio.run(main()) — entry point to start the event loop
asyncio.gather(a(), b()) — run coroutines concurrently

import asyncio
import httpx

async def fetch_title(client: httpx.AsyncClient, url: str) -> str:
    resp = await client.get(url, timeout=10)
    resp.raise_for_status()
    return url

async def main():
    urls = ["https://example.com", "https://httpbin.org/get"]
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch_title(client, u) for u in urls])
    print(results)

asyncio.run(main())

Async Fits

Many parallel API calls
WebSocket streaming chat
FastAPI request handlers
Concurrent DB / cache lookups

Async Does Not Fit

Heavy NumPy / PyTorch compute
Large matrix multiplication
CPU-bound preprocessing without offload

FastAPI Async Route

@app.post("/chat")
async def chat(body: ChatRequest) -> ChatResponse:
    reply = await llm_client.complete(body.message)
    return ChatResponse(text=reply)

Misconception 1: “Async makes PyTorch training faster.”

Why people believe it: Async is marketed as “faster concurrency.”

Reality: GPU training is compute-bound. Async helps orchestrate I/O around the model—fetching prompts, writing logs, calling tools—not the forward/backward pass itself.

Parallel API Calls with `asyncio.gather`

Embedding pipelines often call an API once per document. Sequential calls waste wall-clock time waiting on network latency. asyncio.gather schedules many coroutines and resumes each as responses arrive.

import asyncio
import httpx

async def embed_text(client: httpx.AsyncClient, text: str) -> list[float]:
    resp = await client.post(
        "/v1/embeddings",
        json={"model": "text-embedding-3-small", "input": text},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["data"][0]["embedding"]

async def embed_batch(texts: list[str]) -> list[list[float]]:
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        tasks = [embed_text(client, t) for t in texts]
        return await asyncio.gather(*tasks)

Pattern	When to Use	ML / AI Example
`asyncio.gather`	Run independent I/O tasks in parallel	Batch embedding 100 chunks
`asyncio.Semaphore(n)`	Limit concurrency (rate limits)	Max 10 parallel LLM calls
`asyncio.create_task`	Fire-and-forget background work	Log metrics while streaming
`asyncio.to_thread`	Offload blocking CPU/library code	Run `pandas.read_csv` without blocking loop

Streaming LLM Responses

Chat completions can stream tokens as they are generated. Async iterators let your server forward chunks to clients without holding the full response in memory.

Definition — Async Streaming

Streaming delivers data incrementally as it becomes available. With async for, each chunk is awaited individually—the event loop can serve other connections between chunks.

import httpx

async def stream_chat(prompt: str):
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "POST",
            "https://api.example.com/v1/chat/completions",
            json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}], "stream": True},
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    yield line[6:]   # forward SSE chunk to browser

In FastAPI, return a StreamingResponse wrapping an async generator so users see tokens appear in real time—lower perceived latency than waiting for the full completion.

Rate-Limited Concurrent Fetching

sem = asyncio.Semaphore(5)   # max 5 in flight

async def fetch_with_limit(client, url):
    async with sem:
        return await client.get(url)

async def fetch_all(urls):
    async with httpx.AsyncClient() as client:
        return await asyncio.gather(*[fetch_with_limit(client, u) for u in urls])

Misconception 2: “Adding async to a function makes it run in parallel automatically.”

Why people believe it: async def sounds like “automatic parallelism.”

Reality: A coroutine does nothing until awaited and scheduled on an event loop. Calling fetch() without await returns a coroutine object, not a result.

Misconception 3: “You can call blocking requests.get inside async def safely.”

Why people believe it: It “still works” in small demos.

Reality: Blocking calls freeze the entire event loop. Use httpx.AsyncClient, or offload with await asyncio.to_thread(requests.get, url).

Quick Knowledge Check

Short Answer: What keyword pauses a coroutine until I/O finishes? Answer: await.
True/False: asyncio.gather runs coroutines concurrently on one thread. Answer: True (cooperative concurrency).
Multiple Choice: Best use case: (a) 50 embedding API calls, (b) training ResNet, (c) sorting a list in memory, (d) matrix inverse. Answer: (a).
Short Answer: Entry point to run async main() from a script? Answer: asyncio.run(main()).
True/False: Async speeds up GPU matrix multiplication. Answer: False — compute-bound work needs different strategies.
Short Answer: Why use Semaphore with gather? Answer: Limit concurrent requests to respect rate limits.
Multiple Choice: Stream tokens from an LLM API: (a) blocking for-loop, (b) async for over chunks, (c) threads only, (d) multiprocessing.Pool. Answer: (b).
Short Answer: Safe way to call blocking pandas code in async route? Answer: asyncio.to_thread(...) or run in executor.
True/False: FastAPI supports async def route handlers. Answer: True.
Multiple Choice: HTTP client matching async style: (a) requests only, (b) httpx AsyncClient, (c) urllib without await, (d) os.system. Answer: (b).

Key Takeaways

async/await enables concurrent I/O without threads on a single event loop.
Use asyncio.gather for parallel API and DB operations; add Semaphore for rate limits.
Stream LLM responses with async iterators and async for for responsive UIs.
Do not block the event loop with synchronous requests or heavy CPU work.
Async orchestrates I/O around models; it does not replace GPU training.
Next: Regular Expressions, the module capstone bridging to Volume 04 data engineering.

Trainer’s Guide

Timing demo: Fetch five URLs sequentially vs with asyncio.gather. Wall-clock difference makes the value concrete.

Streaming lab: Wrap a mock async generator that yields one word per second; connect to a simple FastAPI StreamingResponse and observe incremental output in the browser.

Discussion prompt: Your agent calls search, calculator, and LLM tools—which should be async and why?

What’s Next Continue to Regular Expressions (re) for pattern-based text extraction before Volume 04 data pipelines.