asyncio.gather for Batch Processing — Async Requests & Advanced Patterns

Orchestrating Parallel Requests

asyncio.gather is the core function for running multiple coroutines in parallel and collecting all results. Combined with a semaphore, it lets you process batches of requests efficiently while respecting API limits.

Basic Syntax

import asyncio

requests = ["Question 1", "Question 2", "Question 3"]
tasks = [process_request(r) for r in requests]
responses = await asyncio.gather(*tasks)

asyncio.gather takes a variable number of coroutines (hence the *tasks to unpack the list) and returns a list of results in the same order as the submitted tasks. This is an important point: even if question 3 finishes before question 1, responses[0] will always contain the result for question 1.

Complete Pattern: Batch with Semaphore

Here’s the pattern you’ll use most often in production:

import os
from asyncio import Semaphore
from openai import AsyncOpenAI
import httpx

client = AsyncOpenAI(
    api_key=os.getenv("XAI_API_KEY"),
    base_url="https://api.x.ai/v1",
    timeout=httpx.Timeout(3600.0)
)

async def process_batch(prompts: list[str], max_concurrent: int = 5) -> list[str]:
    sem = Semaphore(max_concurrent)

    async def single_request(prompt: str) -> str:
        async with sem:
            response = await client.chat.completions.create(
                model="grok-4.5",
                messages=[{"role": "user", "content": prompt}]
            )
            return response.choices[0].message.content

    tasks = [single_request(p) for p in prompts]
    return await asyncio.gather(*tasks)

This pattern encapsulates the semaphore within the batch function. You just need to call process_batch(prompts) and retrieve the list of responses.

Handling Errors in a Batch

By default, if a single coroutine raises an exception, asyncio.gather cancels everything and propagates the error. The return_exceptions=True parameter changes this behavior:

async def safe_batch(prompts: list[str]) -> list[str | Exception]:
    sem = Semaphore(5)

    async def single_request(prompt: str) -> str:
        async with sem:
            response = await client.chat.completions.create(
                model="grok-4.5",
                messages=[{"role": "user", "content": prompt}]
            )
            return response.choices[0].message.content

    tasks = [single_request(p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# Usage
results = await safe_batch(mes_prompts)
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"Error on prompt {i}: {result}")
    else:
        print(f"Result {i}: {result[:100]}...")

With return_exceptions=True, errors are captured as values in the results list. You can then identify and retry failed requests.

Splitting into Sub-batches (Chunking)

For very large volumes, it’s better to split into sub-batches rather than launching thousands of tasks simultaneously:

async def chunked_batch(
    prompts: list[str],
    chunk_size: int = 50,
    max_concurrent: int = 5
) -> list[str]:
    all_results = []

    for i in range(0, len(prompts), chunk_size):
        chunk = prompts[i:i + chunk_size]
        print(f"Processing batch {i//chunk_size + 1} ({len(chunk)} prompts)")
        results = await process_batch(chunk, max_concurrent)
        all_results.extend(results)

    return all_results

Chunking offers several advantages: you can display progress, save intermediate results, and limit memory usage. If one sub-batch fails, you don’t lose the results from previous sub-batches.

Progress Tracking

For large batches, a progress indicator is useful:

import asyncio
from asyncio import Semaphore

async def batch_with_progress(prompts: list[str], max_concurrent: int = 5) -> list[str]:
    sem = Semaphore(max_concurrent)
    completed = 0
    total = len(prompts)

    async def single_request(prompt: str) -> str:
        nonlocal completed
        async with sem:
            response = await client.chat.completions.create(
                model="grok-4.5",
                messages=[{"role": "user", "content": prompt}]
            )
            completed += 1
            print(f"\r{completed}/{total} completed", end="", flush=True)
            return response.choices[0].message.content

    tasks = [single_request(p) for p in prompts]
    results = await asyncio.gather(*tasks)
    print()  # new line after progress
    return results

Key Takeaways

asyncio.gather(*tasks) runs coroutines in parallel and returns results in the original order
return_exceptions=True prevents a single error from failing the entire batch
Chunking lets you handle very large volumes without saturating memory
Always combine gather with a semaphore to respect rate limits
Add progress tracking for long-running processes