{"title":"Async Python","description":"","content":"AI applications spend a lot of time waiting: model APIs, vector databases, object storage, webhooks, queues, and internal services all add network latency. Async Python lets one process keep many I/O operations in flight without dedicating a thread to each one.\n\n`async` and `await` do not make a single model call faster. They reduce wall-clock time when your program has many independent I/O-bound tasks. This chapter covers the patterns that matter in AI systems: async HTTP calls, concurrent model requests, structured concurrency, streaming responses, cancellation, and concurrency limits.\n\n---\n\n# The Restaurant Analogy\n\nBefore the code, here is the intuition. Async is about not standing still while another system is working.\n\n#### **Synchronous waiter**\n\nConsider a waiter who takes one table's order, walks to the kitchen, stands there watching the chef cook, waits until the food is ready, brings it back to the table, and only then moves to the next table.\n\nIf there are 10 tables and each meal takes 15 minutes to prepare, the last table waits 150 minutes. That is wasteful, but it is exactly how synchronous code handles I/O-bound tasks.\n\n#### **Asynchronous waiter**\n\nA real waiter takes Table 1's order, hands it to the kitchen, immediately walks to Table 2, takes their order, hands it to the kitchen, moves to Table 3, and so on. When any table's food is ready, the waiter picks it up and delivers it. \n\nAll 10 tables get served in roughly 15-20 minutes because the waiter never stands around waiting. The kitchen (the external API) is the bottleneck, not the waiter (your program).\n\nAsync programming works the same way. Your API calls do not get faster individually. But by overlapping the waiting time, the total wall-clock time drops.\n\n\n```mermaid\nflowchart TD\n subgraph SYNC[\"Synchronous Execution\"]\n direction LR\n S1[\"Call API 1
Wait 5s\"]:::primary --> S2[\"Call API 2
Wait 5s\"]:::primary --> S3[\"Call API 3
Wait 5s\"]:::primary\n end\n\n subgraph ASYNC[\"Asynchronous Execution\"]\n direction LR\n A1[\"Call API 1\"]:::green\n A2[\"Call API 2\"]:::green\n A3[\"Call API 3\"]:::green\n A1 --> W[\"Wait ~5s
(all three
in parallel)\"]:::orange\n A2 --> W\n A3 --> W\n W --> R[\"All results
ready\"]:::teal\n end\n\n SYNC ~~~ ASYNC\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef teal fill:#38d9a9,stroke:#000,color:#000\n```\n\n\nIn the synchronous case, three 5-second calls take 15 seconds total. In the async case, those same three calls overlap and finish in about 5 seconds total.\n\n---\n\n# asyncio Fundamentals\n\nPython's async programming is built on the `asyncio` module. The core concepts are small, but they must be used precisely.\n\n### The Event Loop\n\nThe event loop is the engine that makes async work. It maintains a queue of tasks, runs them until each one hits a waiting point (like an API call), then switches to another task that is ready to make progress.\n\nYou almost never interact with the event loop directly. You just write async functions and let the loop handle the scheduling.\n\n### async def and await\n\nTwo keywords are all you need to get started.\n\n- `async def` declares a coroutine function. Calling it does not execute the function immediately. It returns a coroutine object that the event loop can schedule.\n- `await` pauses the current coroutine and hands control back to the event loop. When the awaited operation finishes, the coroutine resumes from where it left off.\n\nHere is a simple example that simulates API calls using `asyncio.sleep`:\n\n\n**main.py**\n\n```python\nimport asyncio\nimport time\n\nasync def call_llm(prompt: str, delay: float) -> str:\n \"\"\"Simulate an LLM API call that takes `delay` seconds.\"\"\"\n print(f\"Sending: {prompt}\")\n await asyncio.sleep(delay) # Simulates network waiting\n print(f\"Received response for: {prompt}\")\n return f\"Response to: {prompt}\"\n\nasync def main():\n start = time.perf_counter()\n\n # Sequential: one after another\n result1 = await call_llm(\"What is Python?\", 0.2)\n result2 = await call_llm(\"What is async?\", 0.3)\n result3 = await call_llm(\"What is an LLM?\", 0.1)\n\n elapsed = time.perf_counter() - start\n print(f\"Sequential: {elapsed:.1f} seconds\") # ~0.6 seconds\n\nasyncio.run(main())\n```\n\n\nThis code runs three simulated API calls sequentially. Even though we are using async syntax, the three `await` calls happen one after another, so the total time is 0.2 + 0.3 + 0.1 = 0.6 seconds. We are not getting any concurrency benefit yet. To overlap these calls, we need `asyncio.gather()`, which we will cover shortly.\n\nThe important thing to understand: `await asyncio.sleep(delay)` is non-blocking. It tells the event loop \"I am going to be idle for this long, go do something else.\" If we had scheduled other tasks, the event loop would run them during this waiting period.\n\nCompare this to `time.sleep(delay)`, which blocks the thread and prevents the event loop from running other coroutines. Do not use `time.sleep()` inside async code.\n\n### asyncio.run()\n\n`asyncio.run()` is the entry point. It creates an event loop, runs your top-level coroutine, and cleans up when done. You call it once, from regular (non-async) code:\n\n`asyncio.run(main())`\n\nThis is the standard way to launch async programs. One caveat: you cannot call `asyncio.run()` from inside an event loop that is already running, which is common in notebooks and web frameworks.\n\n---\n\n# Async HTTP Calls with httpx\n\n`asyncio.sleep` is great for learning, but real AI engineering means making real HTTP calls. The `requests` library does not support async. Instead, use `httpx`, which provides both sync and async clients with an almost identical API.\n\n\n**main.py**\n\n```python\nimport asyncio\nimport httpx\nimport os\n\nasync def call_openai(client: httpx.AsyncClient, prompt: str) -> str:\n \"\"\"Send a prompt to OpenAI and return the response text.\"\"\"\n response = await client.post(\n \"https://api.openai.com/v1/responses\",\n headers={\"Authorization\": f\"Bearer {os.environ['OPENAI_API_KEY']}\"},\n json={\n \"model\": \"gpt-5.4-mini\",\n \"input\": prompt,\n \"max_output_tokens\": 200,\n },\n timeout=30.0,\n )\n response.raise_for_status()\n body = response.json()\n return \"\".join(\n part[\"text\"]\n for item in body.get(\"output\", [])\n for part in item.get(\"content\", [])\n if part.get(\"type\") == \"output_text\"\n )\n\nasync def main():\n async with httpx.AsyncClient() as client:\n result = await call_openai(client, \"Explain async in one sentence.\")\n print(result)\n\nasyncio.run(main())\n```\n\n\nThis small example shows the key patterns for async HTTP:\n\n- `httpx.AsyncClient` is used as an async context manager (`async with`). This ensures the connection pool is properly opened and closed.\n- `await client.post(...)` is the async version of `requests.post(...)`. The syntax is nearly identical.\n- Always set a `timeout`. LLM APIs can hang, and you do not want your program waiting forever.\n- Reuse a single `AsyncClient` across calls. It maintains a connection pool internally, which avoids the overhead of establishing a new TCP connection for every request.\n\n---\n\n# asyncio.gather() for Concurrent Tasks\n\nThis is where async starts paying off. `asyncio.gather()` takes multiple coroutines and runs them concurrently. It returns a list of results in the same order as the input coroutines.\n\n\n**main.py**\n\n```python\nimport asyncio\nimport time\n\nasync def call_model(model_name: str, prompt: str) -> dict:\n \"\"\"Simulate calling different LLM APIs with varying latencies.\"\"\"\n latencies = {\"gpt-5.5\": 0.3, \"claude-sonnet\": 0.4, \"gemini-pro\": 0.2}\n delay = latencies.get(model_name, 2)\n\n await asyncio.sleep(delay) # Simulate API latency\n return {\"model\": model_name, \"response\": f\"Answer from {model_name}\", \"latency\": delay}\n\nasync def main():\n prompt = \"Explain distributed systems in 2 sentences.\"\n\n start = time.perf_counter()\n\n # Run all three calls concurrently\n results = await asyncio.gather(\n call_model(\"gpt-5.5\", prompt),\n call_model(\"claude-sonnet\", prompt),\n call_model(\"gemini-pro\", prompt),\n )\n\n elapsed = time.perf_counter() - start\n print(f\"Total time: {elapsed:.1f} seconds\") # ~0.4 seconds, not 0.9\n\n for r in results:\n print(f\"{r['model']}: {r['response']}\")\n\nasyncio.run(main())\n```\n\n\nWithout `gather`, calling three models sequentially takes 0.3 + 0.4 + 0.2 = 0.9 seconds. With `gather`, all three calls start at the same time, and the total time is determined by the slowest call: about 0.4 seconds. That is the benefit of overlapping I/O.\n\n---\n\n# TaskGroup for Structured Concurrency\n\n`asyncio.gather()` is still common, especially when you want ordered results. In modern Python, `asyncio.TaskGroup` (added in Python 3.11) is often the better default when tasks belong to the same unit of work.\n\nA task group gives the tasks a clear lifetime: they start inside the `async with` block, and the block does not exit until they finish. If one task fails, the group cancels the remaining tasks and raises an exception group. That behavior is usually what you want for a batch where partial work should not continue after the parent operation has failed.\n\n\n**main.py**\n\n```python\nimport asyncio\n\nasync def score_document(document_id: str) -> dict:\n await asyncio.sleep(0.1)\n return {\"document_id\": document_id, \"score\": 0.82}\n\nasync def main():\n async with asyncio.TaskGroup() as group:\n tasks = [\n group.create_task(score_document(document_id))\n for document_id in [\"doc-1\", \"doc-2\", \"doc-3\"]\n ]\n\n results = [task.result() for task in tasks]\n print(results)\n\nasyncio.run(main())\n```\n\n\nUse `gather(return_exceptions=True)` when partial failure is acceptable and you want each result or exception. Use `TaskGroup` when the tasks are part of one operation and failure should cancel the rest.\n\n### Error Handling with return_exceptions\n\nBy default, if any coroutine in `gather` raises an exception, the exception propagates to the caller and you do not get a normal results list. This is often not what you want when calling multiple providers, because one timeout should not discard useful responses from the others.\n\nThe `return_exceptions=True` flag changes this behavior. Instead of raising, exceptions are returned as regular items in the results list:\n\n\n**main.py**\n\n```python\nimport asyncio\n\nasync def flaky_model(name: str) -> str:\n if name == \"unstable-model\":\n raise TimeoutError(f\"{name} timed out\")\n await asyncio.sleep(0.3)\n return f\"Response from {name}\"\n\nasync def main():\n results = await asyncio.gather(\n flaky_model(\"gpt-5.5\"),\n flaky_model(\"unstable-model\"),\n flaky_model(\"claude-sonnet\"),\n return_exceptions=True,\n )\n\n for i, result in enumerate(results):\n if isinstance(result, Exception):\n print(f\"Task {i} failed: {result}\")\n else:\n print(f\"Task {i}: {result}\")\n\nasyncio.run(main())\n```\n\n\nTwo out of three succeeded, and the failure is captured without crashing the program. This pattern matters in production AI systems where you call multiple models or process large batches and need to handle partial failures deliberately.\n\n---\n\n# Semaphores for Rate Limiting\n\nA problem that comes up quickly in production is provider limits. Model APIs usually cap requests per minute, tokens per minute, concurrent requests, or some combination of the three.\n\nIf you fire off 100 concurrent requests with `gather`, you can trigger HTTP 429 errors or saturate your own connection pool. `asyncio.Semaphore` is the simplest concurrency limiter.\n\nA semaphore holds an internal counter. Each time a task acquires the semaphore, the counter decreases. When it reaches zero, any further tasks wait until a running task releases the semaphore.\n\n\n**main.py**\n\n```python\nimport asyncio\nimport time\n\nasync def call_api(semaphore: asyncio.Semaphore, task_id: int) -> str:\n async with semaphore: # Acquire (blocks if limit reached)\n print(f\"[{time.perf_counter():.1f}] Task {task_id} started\")\n await asyncio.sleep(0.1) # Simulate API call\n print(f\"[{time.perf_counter():.1f}] Task {task_id} done\")\n return f\"Result {task_id}\"\n\nasync def main():\n semaphore = asyncio.Semaphore(3) # Max 3 concurrent requests\n\n tasks = [call_api(semaphore, i) for i in range(10)]\n results = await asyncio.gather(*tasks)\n print(f\"Got {len(results)} results\")\n\nasyncio.run(main())\n```\n\n\nWith a semaphore of 3, only 3 tasks run at any given moment. As soon as one finishes, the next one in the queue starts. This controls concurrency, not requests-per-minute by itself. For strict RPM or TPM limits, combine a semaphore with backoff, provider headers, and a rate limiter.\n\nHere is a more realistic version tailored to LLM API calls:\n\n\n**main.py**\n\n```python\nimport asyncio\nimport httpx\nimport os\n\nclass RateLimitedClient:\n def __init__(self, max_concurrent: int = 5, timeout: float = 30.0):\n self.semaphore = asyncio.Semaphore(max_concurrent)\n self.client = httpx.AsyncClient(timeout=timeout)\n\n async def __aenter__(self) -> \"RateLimitedClient\":\n return self\n\n async def __aexit__(self, exc_type, exc, tb) -> None:\n await self.client.aclose()\n\n async def call(self, prompt: str, model: str = \"gpt-5.4-mini\") -> str:\n async with self.semaphore:\n response = await self.client.post(\n \"https://api.openai.com/v1/responses\",\n headers={\"Authorization\": f\"Bearer {os.environ['OPENAI_API_KEY']}\"},\n json={\n \"model\": model,\n \"input\": prompt,\n },\n )\n response.raise_for_status()\n body = response.json()\n return \"\".join(\n part[\"text\"]\n for item in body.get(\"output\", [])\n for part in item.get(\"content\", [])\n if part.get(\"type\") == \"output_text\"\n )\n\n```\n\n\nThis `RateLimitedClient` wraps the semaphore and HTTP client together. Use it as an async context manager so the connection pool closes cleanly. Every call through it respects the concurrency limit, and `max_concurrent` lets you tune parallelism to your provider tier and workload.\n\n\n```mermaid\nflowchart LR\n subgraph Tasks[\"10 Pending Tasks\"]\n\t\tdirection LR\n T1[\"Task 1\"]:::primary\n T2[\"Task 2\"]:::primary\n T3[\"Task 3\"]:::primary\n T4[\"Task 4\"]:::primary\n T5[\"...\"]:::primary\n end\n\n SEM[\"Semaphore
Limit: 3\"]:::orange\n\n subgraph Active[\"Active (max 3)\"]\n\t\tdirection LR\n A1[\"Task A\"]:::green\n A2[\"Task B\"]:::green\n A3[\"Task C\"]:::green\n end\n\n API[\"LLM API\"]:::teal\n\n Tasks --> SEM --> Active --> API\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n classDef teal fill:#38d9a9,stroke:#000,color:#000\n```\n\n\nTasks queue up behind the semaphore. Only 3 pass through at a time. When one completes and releases its semaphore slot, the next queued task enters.\n\n---\n\n# asyncio.as_completed() for Streaming Results\n\n`asyncio.gather()` waits for all tasks to finish and then returns all results at once. But sometimes you want to process results as they arrive, especially when tasks have very different latencies.\n\n`asyncio.as_completed()` returns an iterator of futures that yield results in the order they finish, not the order they were submitted:\n\n\n**main.py**\n\n```python\nimport asyncio\nimport time\n\nasync def call_model(name: str, delay: float) -> dict:\n await asyncio.sleep(delay)\n return {\"model\": name, \"delay\": delay}\n\nasync def main():\n tasks = [\n call_model(\"gemini-flash\", 0.1),\n call_model(\"gpt-5.5\", 0.2),\n call_model(\"claude-haiku\", 0.1),\n call_model(\"gpt-5.4-mini\", 0.3),\n ]\n\n start = time.perf_counter()\n\n for coro in asyncio.as_completed(tasks):\n result = await coro\n elapsed = time.perf_counter() - start\n print(f\"[{elapsed:.1f}s] Got result from {result['model']}\")\n\nasyncio.run(main())\n```\n\n\nThis is useful when you want to display partial results to a user, update a progress bar, or implement a \"fastest model wins\" pattern where you return the first successful response and cancel the rest.\n\n---\n\n# Async Generators for Streaming LLM Responses\n\nMany LLM APIs support streaming, where the model sends tokens back incrementally as they are generated. In Python, async generators let you consume these token streams using `async for`.\n\nAn async generator is a function declared with `async def` that uses `yield` instead of `return`. The caller iterates over it with `async for`:\n\n\n**main.py**\n\n```python\nimport asyncio\nimport httpx\nimport json\nimport os\n\nasync def stream_completion(\n client: httpx.AsyncClient,\n prompt: str,\n model: str = \"gpt-5.4-mini\",\n) -> None:\n \"\"\"Stream tokens from an LLM and print them as they arrive.\"\"\"\n async with client.stream(\n \"POST\",\n \"https://api.openai.com/v1/responses\",\n headers={\"Authorization\": f\"Bearer {os.environ['OPENAI_API_KEY']}\"},\n json={\n \"model\": model,\n \"input\": prompt,\n \"stream\": True,\n },\n ) as response:\n async for line in response.aiter_lines():\n if line.startswith(\"data: \") and line != \"data: [DONE]\":\n chunk = json.loads(line[6:])\n if chunk.get(\"type\") == \"response.output_text.delta\":\n print(chunk[\"delta\"], end=\"\", flush=True)\n\n print() # Newline after streaming completes\n\nasync def main():\n async with httpx.AsyncClient() as client:\n await stream_completion(client, \"Write a haiku about Python.\")\n\nasyncio.run(main())\n```\n\n\nYou can also write your own async generators for custom streaming pipelines. For instance, here is a generator that simulates token streaming:\n\n\n**main.py**\n\n```python\nimport asyncio\n\nasync def fake_token_stream(text: str, tokens_per_second: int = 10):\n \"\"\"Simulate streaming by yielding one word at a time.\"\"\"\n words = text.split()\n delay = 1.0 / tokens_per_second\n for word in words:\n await asyncio.sleep(delay)\n yield word + \" \"\n\nasync def main():\n async for token in fake_token_stream(\"Async generators make streaming natural in Python\"):\n print(token, end=\"\", flush=True)\n print()\n\nasyncio.run(main())\n```\n\n\nAsync generators are the foundation for building streaming UIs, real-time logging, and progressive response display in AI applications.\n\n---\n\n### References\n\n- [Python asyncio documentation](https://docs.python.org/3/library/asyncio.html)\n- [httpx async documentation](https://www.python-httpx.org/async/)\n- [Real Python: Async IO in Python](https://realpython.com/async-io-python/)\n- [PEP 492: Coroutines with async and await syntax](https://peps.python.org/pep-0492/)\n- [OpenAI Streaming Guide](https://platform.openai.com/docs/api-reference/streaming)\n\n---\n\n# Quiz","pageType":"ai-engineering"}

Async Python

Get Premium