How to queue requests to stay under LLM rate limits

When you build against LLM APIs, you must queue requests under llm rate limits or you will eat 429s and drop user work. This guide gives a concrete, end-to-end pattern: a token-bucket throttle, an async worker pool, and clean retry handling that survives provider degradation.

Step 1: Map the limits you actually have

Rate limits come in two flavors: requests per minute (RPM) and tokens per minute (TPM). Some providers also cap concurrent requests. Pull the numbers from your provider’s dashboard, not from a blog post.

LIMITS = {
    "rpm": 60,          # max requests per minute
    "tpm": 90_000,      # max tokens per minute (prompt + completion)
    "max_concurrency": 10,
}

If you batch many small calls, RPM bites first. If you send long prompts, TPM is the wall. Design your queue around both. Estimate token counts before enqueueing using a tokenizer like tiktoken so the TPM bucket can make informed decisions:

import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
def est_tokens(messages):
    # rough: sum of message content lengths tokenized
    return sum(len(enc.encode(m["content"])) for m in messages) + 8

Underestimating tokens will cause late 429s; overestimating needlessly throttles your own throughput.

Step 2: Pick a queue topology

For a single process (CLI, cron, or one server), an in-memory asyncio.Queue is enough. For a distributed service, use Redis or a dedicated broker like BullMQ so multiple workers share one throttle.

In-memory is simpler and has no network hop. Distributed survives restarts and scales horizontally. Choose based on whether you already run more than one replica. If you go distributed, the token bucket must live in a shared store (Redis with Lua atomicity) rather than process memory.

Step 3: Build a token-bucket rate limiter

A token bucket refills at a fixed rate and allows bursts up to capacity. It handles both RPM and TPM if you treat tokens as either requests or estimated token counts. Prefer it over a sliding-window log because it absorbs legitimate bursts without complex bookkeeping.

import asyncio, time

class TokenBucket:
    def __init__(self, rate: float, capacity: float):
        self.rate = rate          # tokens per second
        self.capacity = capacity
        self.tokens = capacity
        self.last = time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self, needed: float = 1.0):
        while True:
            async with self.lock:
                now = time.monotonic()
                self.tokens += (now - self.last) * self.rate
                self.tokens = min(self.tokens, self.capacity)
                self.last = now
                if self.tokens >= needed:
                    self.tokens -= needed
                    return
                wait = (needed - self.tokens) / self.rate
            await asyncio.sleep(wait)

For RPM=60, set rate = 1.0 tokens/sec and capacity = 60. For TPM, estimate tokens per request and use rate = tpm/60. The lock prevents races within one event loop; across processes you need a centralized bucket.

Step 4: Couple the limiter to an async request queue

Create a queue, spawn N workers, and have each worker pull a job, acquire tokens, then call the model. Cap workers at max_concurrency. This is the core loop that lets you queue requests under llm rate limits without spawning unbounded concurrent calls.

from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="https://api.openai.com/v1", api_key="KEY")
queue = asyncio.Queue()
rpm_bucket = TokenBucket(rate=1.0, capacity=60)
tpm_bucket = TokenBucket(rate=1500.0, capacity=90_000)  # 90k/60

async def worker():
    while True:
        job = await queue.get()
        est = job["est_tokens"]
        await tpm_bucket.acquire(est)
        await rpm_bucket.acquire(1)
        try:
            resp = await call_with_retry(lambda: client.chat.completions.create(
                model=job["model"], messages=job["messages"]))
            job["future"].set_result(resp)
        except Exception as e:
            job["future"].set_exception(e)
        finally:
            queue.task_done()

Start workers once at boot:

for _ in range(LIMITS["max_concurrency"]):
    asyncio.create_task(worker())

Step 5: Call the LLM with proper headers and fallback

Pass timeout and capture 429 with Retry-After. If you front your calls with n4n.ai, an OpenAI-compatible endpoint that addresses 240+ models, its automatic fallback when a provider is rate-limited or degraded means your queue only sees hard errors, not transient throttles.

async def enqueue(messages, model="gpt-4o-mini"):
    fut = asyncio.get_event_loop().create_future()
    await queue.put({"messages": messages, "model": model,
                     "est_tokens": est_tokens(messages), "future": fut})
    return await fut

Keep est_tokens honest. The queue is only as accurate as your token math. If you routinely miss, log the delta and recalibrate.

Step 6: Retry with backoff and respect Retry-After

Wrap the client call in a retry loop. Use exponential backoff with jitter, but honor the Retry-After header when present. Never retry on 401 or 404; those are deterministic failures.

import random

async def call_with_retry(fn, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            return await fn()
        except Exception as e:
            status = getattr(e, "status_code", None)
            if status == 429:
                headers = getattr(e, "response", {}).headers or {}
                ra = float(headers.get("Retry-After", 2 ** attempt))
                await asyncio.sleep(ra + random.uniform(0, 0.5))
                continue
            if status in (500, 502, 503):
                await asyncio.sleep((2 ** attempt) + random.uniform(0, 0.5))
                continue
            raise
    raise RuntimeError("exhausted retries")

Integrate it inside the worker before job["future"].set_result. The jitter prevents thundering-herd retries when a provider recovers.

Step 7: Verify the queue holds under load

Write a test that enqueues 200 jobs and asserts zero unhandled 429s. Run it against a staging key with low limits.

async def test_queue():
    tasks = [enqueue([{"role": "user", "content": "hi"}])
             for _ in range(200)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    errors = [r for r in results if isinstance(r, Exception)]
    assert not errors, f"{len(errors)} failures"

Add logging to the worker so you can see bucket waits:

import logging
logging.basicConfig(level=logging.INFO)

# inside worker before acquire
logging.info("queue_depth=%d", queue.qsize())

Monitor queue depth and token bucket levels with a metrics hook (Prometheus or StatsD). If queue.qsize() grows unbounded, your limiter rate is below arrival rate—increase limits or shed load. Successful verification means the test passes and your metrics show zero 429s over a sustained burst.

Beyond the basics

Add a dead-letter queue for jobs that fail after retries. Use priority weights if some requests are user-facing and others are batch. For distributed setups, centralize the token bucket in Redis with a Lua script to avoid race conditions across workers.

Queueing requests under llm rate limits is not optional at scale. The pattern above gives you a deterministic throttle, clean retries, and a path to horizontal scaling without rewriting your call sites.