Building an async worker pool for batch LLM requests

When you need to fire hundreds of prompts at an inference endpoint, a naive sequential loop will bottleneck on latency or trip rate limits. This tutorial builds a python async worker pool batch llm requests system using asyncio that caps concurrency, isolates failures, and streams tokens without blocking the event loop.

Prerequisites

• Python 3.10 or newer (uses asyncio from stdlib).
• openai Python package >= 1.0 (pip install openai).
• An API key for an OpenAI-compatible endpoint. If you want automatic fallback across providers and per-token metering, n4n.ai exposes one OpenAI-compatible endpoint covering 240+ models.
• Comfort with async/await and basic coroutine scheduling.

Why a worker pool beats a gather bomb

asyncio.gather(*[call(p) for p in prompts]) looks clean but sends everything at once. Providers respond with 429s, and a single slow request stalls the whole batch if you await in order. A python async worker pool batch llm requests design decouples production of tasks from consumption, letting you tune exactly how many in-flight requests exist.

The pattern has three parts:

1. A bounded queue of (task_id, prompt) tuples.
2. N long-lived worker coroutines pulling from that queue.
3. A results dict keyed by task_id so order is reconstructed later.

Setting up the async client

Use the official async client. Point base_url at your gateway; the rest of the API is unchanged.

import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["API_KEY"],
    base_url="https://api.n4n.ai/v1",  # OpenAI-compatible, 240+ models
)

If you run your own proxy, swap the URL. Nothing else in this tutorial changes.

Building the queue and worker

The worker coroutine

A worker loops forever, pulls a task, calls the model, stores the result, and marks the queue item done. Errors are caught per-task so one bad prompt doesn’t kill the pool.

async def worker(name: str, queue: asyncio.Queue, results: dict):
    while True:
        task_id, prompt = await queue.get()
        try:
            resp = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0.0,
            )
            results[task_id] = resp.choices[0].message.content
        except Exception as e:
            results[task_id] = f"ERROR: {type(e).__name__}: {e}"
        finally:
            queue.task_done()

Feeding the queue

Production is just put for each prompt. Because asyncio.Queue is unbounded by default, the producer will not block, but you control concurrency via worker count, not queue size.

async def produce(queue: asyncio.Queue, prompts: list[str]):
    for i, p in enumerate(prompts):
        await queue.put((i, p))

Controlling concurrency

Spawn a fixed number of workers. This is the entire concurrency knob—if you want 10 parallel requests, start 10 workers.

async def run_pool(prompts: list[str], concurrency: int = 5):
    queue: asyncio.Queue = asyncio.Queue()
    results: dict = {}
    workers = [
        asyncio.create_task(worker(f"w{i}", queue, results))
        for i in range(concurrency)
    ]
    await produce(queue, prompts)
    await queue.join()          # block until all tasks processed
    for w in workers:
        w.cancel()              # shut down cleanly
    return results

Call it from asyncio.run:

prompts = [f"Give a one-word category for: {i}" for i in range(12)]
res = asyncio.run(run_pool(prompts, concurrency=4))
print([res[i] for i in range(12)])

Expected output (truncated, order preserved):

['numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric']

Adding retries with backoff

Providers degrade. Wrap the call in a retry loop with exponential sleep. Keep retries inside the worker so other tasks proceed.

import asyncio

async def worker_retry(name, queue, results, max_retries=3):
    while True:
        task_id, prompt = await queue.get()
        for attempt in range(max_retries):
            try:
                resp = await client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                )
                results[task_id] = resp.choices[0].message.content
                break
            except Exception as e:
                if attempt == max_retries - 1:
                    results[task_id] = f"ERROR: {e}"
                await asyncio.sleep(0.5 * (2 ** attempt))
        queue.task_done()

The python async worker pool batch llm requests approach isolates these sleeps to the failing worker; the other N-1 workers keep pulling.

Streaming without blocking

If you want tokens as they arrive (for UX or pipelining), set stream=True and accumulate deltas. The event loop stays free because async for yields control.

async def worker_stream(name, queue, results):
    while True:
        task_id, prompt = await queue.get()
        try:
            stream = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                stream=True,
            )
            chunks = []
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    chunks.append(delta)
            results[task_id] = "".join(chunks)
        except Exception as e:
            results[task_id] = f"ERROR: {e}"
        finally:
            queue.task_done()

Streaming does not change pool mechanics—each worker still handles one prompt at a time.

Full runnable example

Combining the pieces into one file you can execute:

import asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["API_KEY"],
    base_url="https://api.n4n.ai/v1",
)

async def worker(name, queue, results, max_retries=3):
    while True:
        task_id, prompt = await queue.get()
        for attempt in range(max_retries):
            try:
                resp = await client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.0,
                )
                results[task_id] = resp.choices[0].message.content
                break
            except Exception as e:
                if attempt == max_retries - 1:
                    results[task_id] = f"ERROR: {e}"
                await asyncio.sleep(0.5 * (2 ** attempt))
        queue.task_done()

async def main():
    prompts = [f"Summarize the integer {i} in one word." for i in range(10)]
    queue = asyncio.Queue()
    results = {}
    workers = [asyncio.create_task(worker(f"w{i}", queue, results)) for i in range(4)]
    for i, p in enumerate(prompts):
        await queue.put((i, p))
    await queue.join()
    for w in workers:
        w.cancel()
    for i in range(len(prompts)):
        print(i, "->", results[i])

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

Run with API_KEY=sk-... python pool.py. Checkpoint output:

0 -> zero
1 -> one
2 -> two
3 -> three
4 -> four
5 -> five
6 -> six
7 -> seven
8 -> eight
9 -> nine

Operational notes

• Concurrency tuning: Start at 5–10 workers. Increase until you see 429s, then back off. The pool makes the limit explicit.
• Timeouts: Wrap client.chat.completions.create in asyncio.wait_for(..., timeout=30) to avoid a hung worker.
• Routing hints: If your gateway honors client routing directives or provider cache-control headers, pass them via extra_headers on the create call. The worker pattern forwards those unchanged.
• Result handling: For large batches, write results to a file or database inside the worker instead of a dict to avoid memory growth.

The python async worker pool batch llm requests structure is boring on purpose: it gives you a single number (worker count) to reason about throughput and a clean place to add retries, logging, and metrics.