Controlling Concurrency and Avoiding Rate Limits

Running requests concurrently (Lesson 4) creates a new problem that a single sequential request never has to deal with: the OpenAI API enforces rate limits, and a worker pool with too much concurrency will hit them constantly. Getting rate-limited isn't just an inconvenience — every rejected request wastes the time it took to build and send, forces a retry, and if handled badly, can create a feedback loop where increasingly aggressive retries make the rate-limiting worse rather than better. This lesson covers how rate limits actually work, and how to build a pipeline that adapts to them instead of colliding with them.

Two Kinds of Rate Limits

The OpenAI API enforces limits along at least two independent dimensions, and it's important to design for both because you can be well within one and still blocked by the other:

• RPM (requests per minute): a cap on how many API calls you can make in a rolling one-minute window, regardless of their size.
• TPM (tokens per minute): a cap on the total number of input and output tokens processed in a rolling one-minute window, regardless of how many separate requests that represents.

A pipeline that sends many small requests can hit the RPM limit long before it comes close to the TPM limit. Conversely, a pipeline sending fewer but much larger requests (long documents, large context windows) can exhaust the TPM limit while staying well under the RPM limit. Bounding concurrency by a single fixed number, as in the semaphore examples so far, controls neither of these directly — it only limits how many requests are in flight simultaneously, not how many complete within a rolling minute.

Note: Exact rate limit values, tiers, and which response headers report them are account- and model-specific, and change as OpenAI adjusts its limits. Confirm current header names and default limits against the official documentation rather than hardcoding assumed values.

Reading Rate Limit Information from Responses

The API typically returns rate-limit information in response headers, which lets a well-behaved client track how close it is to a limit before being rejected, rather than only reacting after a 429 error occurs. The general shape (exact header names should be verified against current docs) looks like this:

async def call_and_inspect_limits(client, prompt: str):
    response = await client.responses.with_raw_response.create(
        model="gpt-5.6-terra",
        input=prompt,
    )
    headers = response.headers
    remaining_requests = headers.get("x-ratelimit-remaining-requests")
    remaining_tokens = headers.get("x-ratelimit-remaining-tokens")
    print(f"remaining requests: {remaining_requests}, remaining tokens: {remaining_tokens}")
    return response.parse()

with_raw_response is the SDK's mechanism for accessing the underlying HTTP response (headers included) instead of only the parsed result object; calling .parse() afterward gives you back the normal typed response. Building a pipeline that watches these values and proactively slows itself down when they run low is far more efficient than a pipeline that only finds out it's over budget from a 429 error.

Handling 429 Errors with Exponential Backoff and Jitter

Even a well-throttled pipeline will occasionally hit a rate limit, especially if other traffic shares the same account. The standard, well-established response is exponential backoff with jitter: wait progressively longer between retries, and add a small random component so that many concurrent workers retrying at once don't all retry at exactly the same moment and immediately re-trigger the limit together.

import asyncio
import random


async def call_with_backoff(client, prompt: str, max_retries: int = 5):
    base_delay = 1.0

    for attempt in range(max_retries):
        try:
            response = await client.responses.create(
                model="gpt-5.6-terra",
                input=prompt,
            )
            return response
        except Exception as exc:
            is_rate_limit = "rate_limit" in str(exc).lower() or "429" in str(exc)
            if not is_rate_limit or attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"rate limited, retrying in {delay:.1f}s (attempt {attempt + 1})")
            await asyncio.sleep(delay)

    raise RuntimeError("unreachable")

Each retry doubles the base wait (1s, 2s, 4s, 8s, ...), which gives the API time to recover its available capacity rather than hammering it with retries at a fixed short interval that never lets the limit reset. The random.uniform(0, 1) jitter term prevents synchronized retries: if fifty concurrent workers all got rate-limited at the same instant, pure exponential backoff without jitter would have all fifty retry at exactly the same future moment, recreating the same spike. Checking attempt == max_retries - 1 ensures the function eventually gives up and re-raises rather than retrying forever, which matters because an item that keeps failing needs to become a recorded failure (Lesson 7), not an infinite loop.

In production code, prefer checking the actual exception type the SDK raises for rate limiting (typically a specific exception class) rather than string-matching on the error message, since message text is more likely to change between SDK versions than exception class names.

from openai import RateLimitError

async def call_with_backoff_typed(client, prompt: str, max_retries: int = 5):
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            return await client.responses.create(model="gpt-5.6-terra", input=prompt)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(delay)

Note: Confirm the exact exception class name (RateLimitError here) against the installed SDK version — exception hierarchies do occasionally change across major SDK releases.

Adaptive Concurrency: Shrinking and Growing the Worker Pool

A fixed concurrency limit chosen once at the start of a job is a compromise: set it too high and you spend most of your time backing off from rate limits; set it too low and you leave throughput on the table during periods when the API has spare capacity. An adaptive concurrency controller adjusts the effective concurrency limit at runtime based on recent success and failure signals — a simplified but practical version tracks a target concurrency value that a semaphore-like structure enforces, decreasing it on rate-limit errors and slowly increasing it again after a run of successes.

class AdaptiveLimiter:
    """A semaphore-like limiter whose capacity shrinks on rate limits
    and grows slowly during sustained success."""

    def __init__(self, initial: int, minimum: int = 1, maximum: int = 50):
        self._value = initial
        self._minimum = minimum
        self._maximum = maximum
        self._lock = asyncio.Lock()
        self._semaphore = asyncio.Semaphore(initial)
        self._consecutive_successes = 0

    async def acquire(self):
        await self._semaphore.acquire()

    def release(self):
        self._semaphore.release()

    async def report_success(self):
        async with self._lock:
            self._consecutive_successes += 1
            if self._consecutive_successes >= 20 and self._value < self._maximum:
                self._value += 1
                self._semaphore.release()  # grow capacity by one permit
                self._consecutive_successes = 0

    async def report_rate_limit(self):
        async with self._lock:
            self._consecutive_successes = 0
            if self._value > self._minimum:
                self._value -= 1
                # Note: shrinking a Semaphore's permits isn't directly
                # supported; a production version tracks a target and
                # has acquire() honor it, e.g. via a custom gate class.

This sketch demonstrates the idea of adaptive concurrency — grow slowly on sustained success, shrink immediately on a rate-limit signal — which is a standard pattern borrowed from network congestion control (the same additive-increase, multiplicative-or-stepwise-decrease shape used in TCP congestion avoidance). The comment in report_rate_limit is intentionally honest about a real limitation: asyncio.Semaphore supports adding permits at runtime (via release()) but has no built-in way to remove permits already issued, so a fully correct implementation typically wraps a target value and has new acquire() calls check against it, or uses a small custom gate class instead of asyncio.Semaphore directly. The teaching point is the policy — react fast to overload, recover capacity slowly — not the exact class used to enforce it.

Combining Backoff and Adaptive Concurrency in the Worker Pool

Putting this together with the worker pool pattern from Lesson 4, each worker calls the model through call_with_backoff_typed, and reports outcomes to a shared AdaptiveLimiter so that the whole pool's effective concurrency responds to real conditions rather than staying fixed for the entire run:

async def robust_worker(name, queue, client, results, limiter: AdaptiveLimiter):
    while True:
        record = await queue.get()
        if record is None:
            queue.task_done()
            break
        await limiter.acquire()
        try:
            response = await call_with_backoff_typed(client, record.prompt)
            record.model_response = response.output_text
            record.status = RecordStatus.SUCCEEDED
            await limiter.report_success()
        except RateLimitError as exc:
            record.error = str(exc)
            record.status = RecordStatus.FAILED
            await limiter.report_rate_limit()
        except Exception as exc:
            record.error = str(exc)
            record.status = RecordStatus.FAILED
        finally:
            limiter.release()
        results.append(record)
        queue.task_done()

Testing Backoff Logic Without Waiting or Calling the API

Backoff logic is straightforward to test by injecting a fake client that fails a fixed number of times before succeeding, and by keeping the sleep durations short:

class FlakyFakeClient:
    def __init__(self, fail_times: int):
        self.fail_times = fail_times
        self.calls = 0

    class responses:
        pass

    async def _create(self, model, input):
        self.calls += 1
        if self.calls <= self.fail_times:
            raise RateLimitError("simulated rate limit", response=None, body=None)

        class FakeResponse:
            output_text = "ok"
        return FakeResponse()


async def test_backoff_eventually_succeeds():
    client = FlakyFakeClient(fail_times=2)
    client.responses.create = client._create  # wire up the fake method

    async def fast_backoff(client, prompt, max_retries=5):
        for attempt in range(max_retries):
            try:
                return await client.responses.create(model="gpt-5.6-terra", input=prompt)
            except RateLimitError:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(0)  # no real delay in the test

    result = await fast_backoff(client, "test")
    assert result.output_text == "ok"
    assert client.calls == 3
    print("PASS: backoff retries past transient rate limits then succeeds")


asyncio.run(test_backoff_eventually_succeeds())

Replacing asyncio.sleep(delay) with asyncio.sleep(0) inside the test keeps it fast while still exercising the real retry-and-give-up control flow.

Common Mistakes

• Retrying immediately with no delay, or with a fixed short delay. This doesn't give the API's rate limit window time to reset and often makes the situation worse by adding more requests into an already-throttled period.
• Retrying without jitter under high concurrency. Many workers backing off by the exact same schedule collide again at the next retry, producing a visible "thundering herd" pattern of repeated synchronized failures.
• Setting concurrency once at pipeline start and never revisiting it. A limit tuned for typical conditions can be far too aggressive during periods of shared account load, and far too conservative during quiet periods — leaving throughput unused for the entire run.

Best Practices

• Always cap the number of retries and record a final failure rather than retrying indefinitely — an item that cannot succeed after several backoff attempts belongs in the failure-handling flow from Lesson 7, not in an infinite retry loop.
• Catch the SDK's specific rate-limit exception type rather than string-matching error text, since typed exceptions are far more stable across SDK versions than error message wording.
• Treat concurrency as a dynamic runtime parameter, not a constant, especially for long-running jobs where API load conditions can change significantly over the course of hours.