LLM Batching and Throughput Optimization

Processing one LLM request at a time works for prototypes. Production systems that handle hundreds or thousands of requests need batching and concurrency to keep costs down and throughput up. This article covers the practical patterns for processing LLM requests at scale, whether you are using hosted APIs or self-hosted models.

API Batch Endpoints

Major providers offer batch APIs that process requests at a discount in exchange for higher latency. OpenAI’s Batch API, for example, offers 50% cost reduction with results delivered within 24 hours.

import json
from openai import OpenAI

client = OpenAI()

def create_batch_file(requests: list[dict], filename: str = "batch.jsonl"):
    """Create a JSONL file for the OpenAI Batch API."""
    with open(filename, "w") as f:
        for i, req in enumerate(requests):
            line = {
                "custom_id": f"request-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": req.get("model", "gpt-4o-mini"),
                    "messages": req["messages"],
                    "max_tokens": req.get("max_tokens", 1024),
                },
            }
            f.write(json.dumps(line) + "\n")
    return filename

def submit_batch(filename: str) -> str:
    """Upload file and submit batch job."""
    with open(filename, "rb") as f:
        file_obj = client.files.create(file=f, purpose="batch")

    batch = client.batches.create(
        input_file_id=file_obj.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
    )
    return batch.id

def check_batch(batch_id: str) -> dict:
    """Check batch status and retrieve results if complete."""
    batch = client.batches.retrieve(batch_id)
    if batch.status == "completed":
        content = client.files.content(batch.output_file_id)
        results = [json.loads(line) for line in content.text.strip().split("\n")]
        return {"status": "completed", "results": results}
    return {"status": batch.status, "progress": f"{batch.request_counts}"}

Use batch APIs when results are not time-sensitive: nightly data processing, bulk classification, dataset generation, or evaluation runs.

Concurrent API Calls

When you need results faster than batch APIs allow but still want parallelism, use concurrent async calls with rate limiting.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

class RateLimitedProcessor:
    def __init__(self, max_concurrent: int = 10, rpm_limit: int = 500):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rpm_limit = rpm_limit
        self._request_times: list[float] = []

    async def _wait_for_rate_limit(self):
        """Simple sliding window rate limiter."""
        import time
        now = time.time()
        # Remove requests older than 60 seconds
        self._request_times = [
            t for t in self._request_times if now - t < 60
        ]
        if len(self._request_times) >= self.rpm_limit:
            wait_time = 60 - (now - self._request_times[0])
            if wait_time > 0:
                await asyncio.sleep(wait_time)
        self._request_times.append(time.time())

    async def process_one(self, messages: list[dict], model: str = "gpt-4o-mini") -> str:
        async with self.semaphore:
            await self._wait_for_rate_limit()
            response = await async_client.chat.completions.create(
                model=model,
                messages=messages,
                max_tokens=1024,
            )
            return response.choices[0].message.content

    async def process_batch(self, requests: list[list[dict]]) -> list[str]:
        tasks = [self.process_one(msgs) for msgs in requests]
        return await asyncio.gather(*tasks, return_exceptions=True)

# Usage
async def main():
    processor = RateLimitedProcessor(max_concurrent=10)
    requests = [
        [{"role": "user", "content": f"Summarize topic {i}"}]
        for i in range(100)
    ]
    results = await processor.process_batch(requests)
    successful = [r for r in results if isinstance(r, str)]
    failed = [r for r in results if isinstance(r, Exception)]
    print(f"Success: {len(successful)}, Failed: {len(failed)}")

The semaphore limits concurrent connections while asyncio.gather runs all requests as parallel coroutines. Set max_concurrent based on your API tier’s rate limits.

Exponential Backoff for Retries

Rate limit errors (HTTP 429) and server errors (HTTP 5xx) are inevitable at scale. Implement exponential backoff with jitter.

import asyncio
import random

async def retry_with_backoff(
    func,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
):
    """Retry with exponential backoff and jitter."""
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except Exception as e:
            error_str = str(e).lower()
            is_retryable = any(
                keyword in error_str
                for keyword in ["rate_limit", "429", "500", "502", "503"]
            )

            if not is_retryable or attempt == max_retries:
                raise

            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.5)
            await asyncio.sleep(delay + jitter)

Jitter prevents the “thundering herd” problem where all retries fire simultaneously after a rate limit window resets.

Queue-Based Processing

For sustained high-volume workloads, a queue decouples request ingestion from processing. This handles traffic spikes gracefully and makes it easy to scale workers.

import asyncio
from dataclasses import dataclass
from collections import deque

@dataclass
class QueuedRequest:
    id: str
    messages: list[dict]
    future: asyncio.Future

class LLMQueue:
    def __init__(self, num_workers: int = 5):
        self._queue: deque[QueuedRequest] = deque()
        self._num_workers = num_workers
        self._running = False

    async def enqueue(self, request_id: str, messages: list[dict]) -> str:
        future = asyncio.get_event_loop().create_future()
        self._queue.append(QueuedRequest(request_id, messages, future))
        return await future

    async def _worker(self, worker_id: int):
        while self._running:
            if not self._queue:
                await asyncio.sleep(0.05)
                continue

            request = self._queue.popleft()
            try:
                result = await async_client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=request.messages,
                )
                request.future.set_result(
                    result.choices[0].message.content
                )
            except Exception as e:
                request.future.set_exception(e)

    async def start(self):
        self._running = True
        workers = [
            asyncio.create_task(self._worker(i))
            for i in range(self._num_workers)
        ]
        return workers

    async def stop(self):
        self._running = False

In production, replace the in-memory deque with Redis, RabbitMQ, or AWS SQS for durability and multi-process coordination.

Choosing the Right Approach

Approach	Latency	Cost	Complexity	Best For
Sequential	Highest	Baseline	Lowest	Prototyping
Concurrent async	Low	Baseline	Medium	Real-time applications
Batch API	1-24 hours	50% savings	Low	Offline processing
Queue-based	Medium	Baseline	Highest	High-volume production

Most production systems combine approaches. Use concurrent async for user-facing requests and batch APIs for background jobs. Add a queue when you need to handle traffic spikes or distribute work across multiple services.

Monitoring Throughput

Track these metrics to understand your system’s performance:

from dataclasses import dataclass, field
from time import time

@dataclass
class ThroughputMetrics:
    requests_completed: int = 0
    total_tokens_used: int = 0
    errors: int = 0
    _start_time: float = field(default_factory=time)

    @property
    def requests_per_minute(self) -> float:
        elapsed = (time() - self._start_time) / 60
        return self.requests_completed / max(elapsed, 0.001)

    @property
    def error_rate(self) -> float:
        total = self.requests_completed + self.errors
        return self.errors / max(total, 1)

An increasing error rate usually means you are hitting rate limits and need to reduce concurrency or add more API keys. Decreasing throughput with stable error rates suggests your requests are getting larger (more tokens per request).

Summary

Efficient LLM throughput requires matching the processing pattern to the use case. Use batch APIs for offline work at half the cost. Use concurrent async calls with rate limiting for real-time applications. Add queue-based processing when you need durability and scalability. Always implement exponential backoff with jitter, and monitor throughput metrics to catch performance regressions early.