Using Asynchronous Python With OpenAI SDK

Unit 12, Lesson 6 introduced AsyncOpenAI and showed a basic asyncio.gather example for running requests concurrently. That lesson focused on the SDK's async interface. This lesson goes one level deeper: it explains why asynchronous Python is the right execution model for large-volume AI workloads, how the event loop actually behaves while your requests are in flight, and how to structure a pipeline's Submit stage around async/await correctly — including the mistakes that quietly turn "async" code back into sequential code.

Why Async Fits AI Request Workloads Specifically

Every call to the OpenAI API is I/O-bound: your program sends a request over the network and then does nothing but wait for a response. During that wait — which for a language model call can be anywhere from a few hundred milliseconds to tens of seconds — your CPU is completely idle. If you make requests one at a time, sequentially, you are paying that entire wait time for every single item, multiplied by the number of items. For ten thousand requests averaging two seconds each, sequential processing takes over five and a half hours of pure waiting.

Asynchronous programming solves exactly this problem: it lets a single thread hold many requests "in flight" at once, switching between them whenever one is waiting on the network, so the waiting time overlaps instead of stacking up. This is different from using multiple threads or processes, which add CPU and memory overhead to get true parallelism — overhead that isn't needed here, because the bottleneck is network waiting, not CPU work. Async concurrency gets you the throughput benefit of parallelism for I/O-bound work with far less overhead per concurrent task.

The Event Loop, in Practical Terms

Python's asyncio runs a single-threaded event loop: a scheduler that keeps a list of tasks and, whenever the currently running task hits an await on something that isn't ready yet (like a network response), pauses that task and runs a different one that's ready to make progress. Nothing in this model executes literally at the same instant — it's cooperative multitasking, not parallel execution — but because the tasks spend almost all their time waiting rather than computing, the effect is that many requests appear to be "in progress" simultaneously, and the total wall-clock time to finish all of them approaches the time for the slowest single request, not the sum of all of them.

This has a critical, easy-to-miss consequence: cooperative multitasking only works if every task actually yields control at its await points. A coroutine that runs a long, CPU-bound loop of pure Python computation between its await calls blocks the entire event loop for that duration — every other pending request, no matter how ready it is to proceed, must wait. Async gives you concurrency for waiting, not for computing; if your pipeline needs heavy CPU work (like re-encoding large images before sending them, or complex text processing), that work belongs in a separate thread or process pool, not inline in an async coroutine.

Coroutines, async def, and await

A function defined with async def is a coroutine function. Calling it does not run its body immediately — it returns a coroutine object, which is a paused computation that only advances when something drives it forward (await-ing it, or scheduling it as a task). The await keyword is how a coroutine says "I am waiting on this other awaitable; pause me here and let the event loop run something else in the meantime."

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()


async def summarize(text: str) -> str:
    response = await client.responses.create(
        model="gpt-5.6-terra",
        input=f"Summarize in one sentence: {text}",
    )
    return response.output_text


async def main():
    result = await summarize("Async programming lets I/O-bound code overlap waiting time.")
    print(result)


asyncio.run(main())

Three details in this example matter more than they look:

• AsyncOpenAI() is a distinct client class from OpenAI(). It exposes the same method names (client.responses.create) but every method returns a coroutine that must be awaited, because internally it uses a non-blocking HTTP client rather than a blocking one.
• asyncio.run(main()) is the entry point that creates an event loop, runs main() to completion, and tears the loop down. This should appear exactly once, at the top level of your program — you don't nest asyncio.run calls inside other async code.
• A single await client.responses.create(...) by itself gains you nothing over the synchronous client — the concurrency benefit only appears once you have multiple such calls in flight at the same time, which is the subject of Lesson 4.

Note: Confirm the exact async client class name and constructor signature against the current OpenAI Python SDK documentation, since client initialization options (timeouts, base URL, retry settings) are the kind of detail that evolves between SDK versions.

Structuring an Async Submit Stage

Recall the submit_and_collect function from Lesson 2, which the pipeline calls to turn prepared records into results. Its asynchronous implementation needs an async function that processes one record, and a way to run many of them concurrently — the second half is covered fully in Lesson 4, but the shape of the per-record coroutine is worth establishing here:

from dataclasses import replace


async def process_one_record(client: AsyncOpenAI, record: PipelineRecord) -> PipelineRecord:
    try:
        response = await client.responses.create(
            model="gpt-5.6-terra",
            input=record.prompt,
        )
        record.model_response = response.output_text
        record.status = RecordStatus.SUCCEEDED
    except Exception as exc:
        record.error = str(exc)
        record.status = RecordStatus.FAILED
    record.attempts += 1
    return record

This function is deliberately narrow: it does one thing (call the model for one record) and always returns a record with its status set, whether the call succeeded or failed. Catching the exception here rather than letting it propagate is intentional — in a bulk workload, one failed item must not stop the other 9,999 from being processed, a theme Lesson 7 returns to in more depth for pipeline-level failure handling.

Mixing Sync and Async Code Safely

Real pipelines usually have synchronous pieces — a database call using a synchronous driver, a CSV read, file I/O — sitting next to the async request logic. Calling a blocking, synchronous function directly inside a coroutine has the same effect as CPU-bound work: it blocks the entire event loop for its duration, defeating the purpose of using async in the first place. When a blocking call cannot be avoided (an older database library with no async version, for instance), move it off the event loop using asyncio.to_thread:

def blocking_db_write(record_id: str, result: str) -> None:
    # Simulates a synchronous, blocking database call.
    import time
    time.sleep(0.05)


async def save_result_async(record: PipelineRecord) -> None:
    await asyncio.to_thread(blocking_db_write, record.record_id, record.model_response)

asyncio.to_thread runs the blocking function in a separate worker thread and gives you back an awaitable, so the event loop stays free to keep other coroutines moving while that thread works. This is the correct tool specifically for occasional blocking calls mixed into an otherwise async pipeline — if the majority of your work is blocking I/O, it's usually simpler to look for an async-native library instead of routing everything through threads.

A related and easy mistake is calling time.sleep() inside a coroutine when you meant to pause without blocking everything else:

import time

async def bad_pause():
    time.sleep(1)          # blocks the ENTIRE event loop for 1 second

async def good_pause():
    await asyncio.sleep(1)  # yields control; other coroutines keep running

asyncio.sleep is a coroutine itself — awaiting it tells the event loop "this task has nothing to do for one second, feel free to run something else," which is exactly the behavior you want when, for example, implementing backoff between retries in a concurrent pipeline (Lesson 5).

Common Mistakes

• Forgetting await on an async SDK call. Calling client.responses.create(...) without await returns a coroutine object instead of a response, and Python won't raise an error until you try to use that coroutine object as if it were a real response — this typically surfaces as a confusing AttributeError far from the actual mistake.
• Doing CPU-heavy work inline inside a coroutine. Parsing enormous JSON payloads, running regex over megabytes of text, or numeric computation between await points blocks every other in-flight request. Move genuinely CPU-bound work to asyncio.to_thread or a process pool.
• Calling blocking synchronous I/O (file reads, requests.get, time.sleep) directly inside async functions. Each blocking call silently serializes what should be concurrent work, and the resulting slowdown is easy to miss because the code still runs — just far slower than expected.

Best Practices

• Keep coroutines small and single-purpose, matching the process_one_record pattern: do the await-based work, catch and record errors locally, and return a consistent result shape rather than letting exceptions propagate out of individual item processing.
• Use one AsyncOpenAI client instance for the whole pipeline run rather than creating a new client per request — the client manages an underlying connection pool that is meant to be reused across many calls.
• Push blocking calls out with asyncio.to_thread rather than accepting silent serialization. If you're not sure whether a library call is async-safe, check whether it's async def or documented as async-compatible before assuming it's safe to await directly.