How to run concurrent LLM calls with asyncio.gather

Sending sequential requests to an LLM API wastes wall-clock time when you need multiple independent completions. Using python asyncio gather concurrent llm calls lets you fire dozens of prompts in parallel and await them as a batch, cutting total latency to the slowest single call instead of the sum of all. This pattern is the backbone of any fan-out LLM workload: eval harnesses, bulk classification, synthetic data generation.

Step 1: Install dependencies and point at an OpenAI-compatible endpoint

You need Python 3.8+ and the official OpenAI SDK. The async client ships with the package.

pip install openai

Configure the client once at process startup. If you want a single gateway that fronts many providers, n4n.ai offers one OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is degraded. Otherwise use the default OpenAI base URL.

import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    # base_url="https://api.n4n.ai/v1",  # optional OpenAI-compatible gateway
)

Keep the client object alive. It manages a connection pool and reuses TCP sockets across coroutines, which matters when you launch hundreds of concurrent calls.

Step 2: Write a single async LLM call

Wrap one completion in a coroutine. Keep it narrow: one prompt in, one string out. Set a low max_tokens for batch jobs to control cost and latency.

async def complete(prompt: str, model: str = "gpt-4o-mini") -> str:
    resp = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
        max_tokens=128,
    )
    return resp.choices[0].message.content.strip()

This function is the unit of work. When it hits await, the event loop is free to run other coroutines. That is the entire mechanism—no threads, no multiprocessing, just cooperative scheduling.

Step 3: Batch prompts with asyncio.gather

The core of python asyncio gather concurrent llm calls is asyncio.gather. It accepts coroutines, schedules them concurrently, and returns results in the same order as the inputs.

import asyncio

prompts = [f"Summarize topic {i}" for i in range(20)]

async def main():
    results = await asyncio.gather(*(complete(p) for p in prompts))
    for p, r in zip(prompts, results):
        print(p, "->", r[:50])

if __name__ == "__main__":
    asyncio.run(main())

All 20 requests are dispatched almost simultaneously. If a single call takes ~800 ms, the batch finishes in ~800 ms rather than 16 seconds. For contrast, a sequential loop:

async def sequential():
    out = []
    for p in prompts:
        out.append(await complete(p))  # blocks event loop each iteration
    return out

sequential is strictly slower and gains you nothing.

Step 4: Limit concurrency to avoid rate limits

Gateways enforce requests-per-minute (RPM) and tokens-per-minute (TPM) limits. Firing 1,000 coroutines at once will trigger 429s. Use asyncio.Semaphore to cap in-flight requests.

sem = asyncio.Semaphore(10)

async def bounded_complete(prompt: str) -> str:
    async with sem:
        return await complete(prompt)

async def main_bounded():
    results = await asyncio.gather(*(bounded_complete(p) for p in prompts))
    return results

When scaling python asyncio gather concurrent llm calls to thousands of prompts, tune the semaphore to your tier. Start at 10, watch for 429s, and raise gradually. The semaphore guarantees no more than N coroutines are inside complete at any moment.

Step 5: Survive partial failures

A single malformed prompt or transient 5xx should not kill the whole batch. Pass return_exceptions=True to gather, then partition the outcomes.

async def main_safe():
    outcomes = await asyncio.gather(
        *(bounded_complete(p) for p in prompts),
        return_exceptions=True,
    )
    ok = [o for o in outcomes if not isinstance(o, Exception)]
    errs = [o for o in outcomes if isinstance(o, Exception)]
    print(f"Got {len(ok)} ok, {len(errs)} failed")
    for e in errs:
        logging.warning("call failed: %s", e)
    return ok, errs

Alternatively, catch inside complete and return a sentinel string. return_exceptions is cleaner because it preserves the original traceback for logging.

Step 6: Enforce per-call timeouts

A hung TCP connection blocks a semaphore slot indefinitely. Wrap each call in asyncio.wait_for so a stalled provider drops that one task instead of starving the batch.

async def complete_with_timeout(prompt: str, timeout: float = 5.0) -> str:
    return await asyncio.wait_for(complete(prompt), timeout)

async def bounded_timeout_complete(prompt: str) -> str:
    async with sem:
        return await complete_with_timeout(prompt)

Use the same gather shape:

results = await asyncio.gather(
    *(bounded_timeout_complete(p) for p in prompts),
    return_exceptions=True,
)

Step 7: Verify success

Run the script and measure two things: wall-clock time and result integrity.

import time

async def main_timed():
    t0 = time.monotonic()
    ok, errs = await main_safe()
    dt = time.monotonic() - t0
    print(f"Completed {len(ok)} in {dt:.2f}s")
    assert all(isinstance(x, str) and x for x in ok)
    assert len(ok) + len(errs) == len(prompts)

if __name__ == "__main__":
    asyncio.run(main_timed())

Expected behavior: with 20 prompts and Semaphore(10), total time is roughly 2× a single call, not 20×. If you see 16s, you accidentally used the sync client or forgot asyncio.run. Check resp.usage per call if you want per-token metering—log total_tokens to confirm billing matches expectations.

Step 8: Streaming and advanced patterns

Non-streaming gather is simplest, but if you need tokens as they arrive, gather async generators. Each stream is itself async; you consume them concurrently with the same pattern.

async def stream_collect(prompt: str) -> str:
    buf = []
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices[0].delta.content:
            buf.append(chunk.choices[0].delta.content)
    return "".join(buf)

# wrap with semaphore + timeout exactly like before

One caveat: asyncio.gather does not cancel siblings on first error. For fail-fast semantics on Python 3.11+, use asyncio.TaskGroup:

async def main_taskgroup():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(bounded_timeout_complete(p)) for p in prompts]
    return [t.result() for t in tasks]

TaskGroup cancels remaining tasks if any raises, which is useful when the batch is all-or-nothing.

Step 9: Production notes

Reuse the AsyncOpenAI client; never create it per call.
Honor provider cache-control hints if your gateway forwards them. Prefix stable context so repeat tokens are served from cache, cutting cost on large fan-outs.
Add a retry decorator (e.g., tenacity.AsyncRetrying) inside complete for idempotent prompts, but keep the semaphore outside so retries don’t multiply concurrency.
If you need per-model routing, pass model per prompt and let the gateway honor client directives. A single OpenAI-compatible endpoint simplifies this—no separate clients per provider.

Get the semaphore and timeout right, and python asyncio gather concurrent llm calls will stay stable under heavy load. The pattern is boring in the best way: predictable latency, clean failure isolation, and linear scaling until the provider’s limit is the bottleneck.