Parallelism and Concurrency — Production, Evals, and Optimization

Waiting Together Isn’t Computing Together

An OpenAI API call is an I/O-bound operation: your program isn’t crunching numbers, it’s waiting for a network response. This nature dictates the right tool. Parallelism, in the sense of multiprocessing, distributes computation across multiple cores; it costs a full process per unit of work, and running a hundred processes that are all waiting on the network is pure waste. Concurrency, as provided by asyncio, allows a single process to handle hundreds of simultaneous waits at negligible cost. This is what maximizes your call throughput.

The distinction is more than academic: a team that parallelizes its LLM calls with a process pool will consume several gigabytes of memory for zero gain compared to a few lines of asyncio.

Launching Concurrent Calls

The SDK exposes an asynchronous client, AsyncOpenAI, whose calls await with await. Gathering multiple coroutines in an asyncio.gather lets them all progress simultaneously, with the total duration approaching that of the slowest call rather than their sum.

import asyncio
import openai

client = openai.AsyncOpenAI()

async def single_call(prompt: str, model: str = "gpt-5.6-terra") -> str:
    """A single asynchronous API call."""
    response = await client.responses.create(
        model=model,
        input=prompt,
    )
    return response.output_text

async def concurrent_calls(prompts: list[str]) -> list[str]:
    """Launches all calls concurrently."""
    tasks = [single_call(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Error on prompt {i}: {result}")

    return [r for r in results if not isinstance(r, Exception)]

The return_exceptions=True flag is what distinguishes a prototype from production-grade processing. Without it, the first exception interrupts the gather and you lose the ninety-nine results already obtained. With it, errors become values that you inspect, log, and retry at your own pace.

However, this raw version has an immediate flaw: launching a thousand calls at once will blow past your rate limits in seconds. A semaphore solves this by allowing only a fixed number of simultaneous calls, with the rest waiting patiently for their turn. Twenty is a reasonable starting point, to be adjusted based on your organization’s actual quotas.

async def calls_with_limit(
    prompts: list[str],
    max_concurrency: int = 20,
) -> list[str]:
    """Limits the number of simultaneous calls."""
    semaphore = asyncio.Semaphore(max_concurrency)

    async def limited_call(prompt: str) -> str:
        async with semaphore:
            return await single_call(prompt)

    tasks = [limited_call(prompt) for prompt in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

The Map-Reduce Pattern Applied to Text

A two-hundred-page document cannot be summarized in a single call, and slicing it sequentially would take minutes. The map-reduce pattern applies naturally here: split into chunks, summarize each chunk concurrently, then ask the model for a synthesis of the summaries. The map phase benefits fully from concurrency, while the reduce phase is a single call on text that has become short.

async def map_reduce_summary(
    document: str,
    chunk_size: int = 3000,
) -> str:
    """Summarizes a long document via parallel map-reduce."""

    # Phase MAP: split and summarize in parallel
    chunks = [
        document[i:i + chunk_size]
        for i in range(0, len(document), chunk_size)
    ]

    map_prompts = [
        f"Summarize this passage in 3 key points:\n\n{chunk}"
        for chunk in chunks
    ]

    partial_summaries = await calls_with_limit(map_prompts, max_concurrency=10)

    # Phase REDUCE: combine summaries
    summary_text = "\n\n---\n\n".join(
        r for r in partial_summaries if isinstance(r, str)
    )

    final_summary = await single_call(
        f"Synthesize these partial summaries into a coherent summary "
        f"of 5 key points:\n\n{summary_text}"
    )

    return final_summary

The filtering if isinstance(r, str) is as much a product decision as a technical precaution: it means the final synthesis will be produced even if three chunks failed, silently. Depending on the context, this is exactly what you want, or it is unacceptable. Decide explicitly, and log the number of lost chunks in any case.

When Multiprocessing Becomes Relevant Again

There remain rare situations where computation takes back control: heavy post-processing on responses, entity extraction, scoring, structured parsing. This work is CPU-bound and gains nothing from asyncio. The elegant combination is to chain the two regimes in a single pipeline: network calls in concurrency, then computation distributed across a process pool via run_in_executor.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def heavy_post_processing(text: str) -> dict:
    """CPU-bound processing on the result (NLP, parsing, etc.)."""
    # Example: entity extraction, scoring, etc.
    words = text.split()
    return {
        "word_count": len(words),
        "unique_words": len(set(words)),
    }

async def complete_pipeline(prompts: list[str]) -> list[dict]:
    """Pipeline: async LLM calls + parallel CPU post-processing."""
    # Step 1: concurrent LLM calls
    llm_results = await calls_with_limit(prompts)

    # Step 2: parallel post-processing (CPU)
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor(max_workers=4) as pool:
        tasks = [
            loop.run_in_executor(pool, heavy_post_processing, r)
            for r in llm_results
            if isinstance(r, str)
        ]
        final_results = await asyncio.gather(*tasks)

    return final_results

Verify that the computation truly justifies the pool before introducing it: serializing data to another process has a cost, and for treatments lasting a few milliseconds this cost exceeds the gain.

Verifying Real Throughput

Tuning max_concurrency isn’t guesswork. Too low, and you leave throughput on the table; too high, and you trigger rate limits that cost you more time than they save. Measure on your actual load, varying the value, and look at both throughput and success count—a very fast pipeline that fails one in five times isn’t fast.

import time

async def benchmark_throughput(num_requests: int = 100):
    """Measures the real throughput of your pipeline."""
    prompts = [
        f"Give an interesting fact number {i}."
        for i in range(num_requests)
    ]

    start = time.perf_counter()
    results = await calls_with_limit(prompts, max_concurrency=20)
    duration = time.perf_counter() - start

    successes = sum(1 for r in results if isinstance(r, str))
    print(f"Requests: {num_requests}")
    print(f"Successes: {successes}")
    print(f"Duration: {duration:.1f}s")
    print(f"Throughput: {successes / duration:.1f} req/s")

Key Takeaways

Use AsyncOpenAI and asyncio.gather for I/O concurrency
Limit concurrency with a semaphore to respect quotas
The map-reduce pattern parallelizes long document processing
Reserve ProcessPoolExecutor for CPU-intensive post-processing