When you fan out hundreds of calls to an LLM API from a single Python process, you need to python asyncio rate limit concurrent requests or you’ll trip provider thresholds and trash your own latency. This article walks through a concrete pattern: cap in-flight calls with a semaphore, decouple submission with a queue, and apply backoff on throttling responses.
Step 1: Cap in-flight calls with asyncio.Semaphore
The simplest way to python asyncio rate limit concurrent requests is an asyncio.Semaphore. It acts as an async-aware counter. Every await sem.acquire() decrements; the context manager releases on exit. If the counter is zero, the caller suspends until another task exits.
import asyncio
import httpx
MAX_CONCURRENT = 20
sem = asyncio.Semaphore(MAX_CONCURRENT)
async def call_llm(client: httpx.AsyncClient, payload: dict) -> dict:
async with sem:
resp = await client.post("https://api.example.com/v1/chat", json=payload)
resp.raise_for_status()
return resp.json()Spawn 100 tasks and only 20 hit the wire at once. The rest wait at the async with sem line. This protects remote rate limits and local socket exhaustion.
Do not skip the semaphore by writing asyncio.gather(*[call_llm(c, p) for p in payloads]) where call_llm has no cap. Gather schedules all coroutines immediately; they will all attempt client.post concurrently. The semaphore must wrap the actual I/O.
A common mistake is creating the semaphore inside the task function. It must be shared across tasks, typically passed as an argument or module global.
Step 2: Decouple submission from execution with a queue
A raw semaphore blocks the caller at the call site. In a server, you want to accept requests fast and process them at a steady rate. Use asyncio.Queue to buffer work and a worker pool to consume it.
import asyncio
import logging
logger = logging.getLogger(__name__)
async def worker(queue: asyncio.Queue, client: httpx.AsyncClient):
while True:
payload = await queue.get()
try:
async with sem:
resp = await client.post("/v1/chat", json=payload)
resp.raise_for_status()
# process resp.json()
except Exception as e:
logger.error("request failed: %s", e)
finally:
queue.task_done()
async def run_pool(queue: asyncio.Queue, client: httpx.AsyncClient, n_workers: int = 10):
workers = [asyncio.create_task(worker(queue, client)) for _ in range(n_workers)]
await queue.join()
for w in workers:
w.cancel()Set maxsize on the queue to apply backpressure:
queue = asyncio.Queue(maxsize=100)If producers exceed consumers, await queue.put(payload) suspends instead of growing memory without bound. This is the second pillar of python asyncio rate limit concurrent requests: not just limiting parallel execution but limiting pending intent.
Step 3: Enforce timeouts and cancel cleanly
A hung TCP connection holds a semaphore slot and a worker. Wrap the network call in asyncio.wait_for so it is cancelled after a bound.
async def call_with_timeout(client, payload, timeout=30.0):
async with sem:
try:
resp = await asyncio.wait_for(
client.post("/v1/chat", json=payload), timeout
)
return resp.json()
except asyncio.TimeoutError:
logger.warning("llm call timed out")
raiseWhen the timeout triggers, the wait_for cancels the inner coroutine. The async with sem block exits, releasing the slot. Without this, a slow stream can wedge a worker indefinitely and silently erode your concurrency cap.
If you cancel the outer task, the semaphore release still happens because the context manager __aexit__ runs during cancellation unwinding. Verify this in tests by asserting the semaphore value returns to initial after a cancelled batch.
Step 4: Back off on 429 and 5xx
Even with a local cap, the provider may throttle due to other tenants or account limits. Inspect status codes and retry with exponential backoff plus jitter.
import random
async def call_with_backoff(client, payload, max_retries=5):
delay = 0.5
for attempt in range(max_retries):
async with sem:
resp = await client.post("/v1/chat", json=payload)
if resp.status_code == 429 or resp.status_code >= 500:
retry_after = resp.headers.get("retry-after")
wait = float(retry_after) if retry_after else delay
await asyncio.sleep(wait + random.uniform(0, 0.2))
delay *= 2
continue
resp.raise_for_status()
return resp.json()
raise RuntimeError("exhausted retries")If your endpoint is a gateway such as n4n.ai that provides automatic fallback when a provider is rate-limited, client-side backoff still prevents a retry storm from your own event loop. The gateway shifts traffic to a healthy provider; you avoid hammering it with 50 immediate retries.
Step 5: Stream without blocking the event loop
LLM completions often stream tokens. Use httpx.AsyncClient.stream and async for. Decide whether the semaphore should cover the whole stream or just the connect.
async def stream_llm(client, payload):
async with sem:
async with client.stream("POST", "/v1/chat", json=payload) as resp:
async for line in resp.aiter_lines():
yield lineIf streams last many seconds, holding sem for the full duration limits concurrent streams to MAX_CONCURRENT. That may be too strict for connection pooling. Split the concerns:
connect_sem = asyncio.Semaphore(10)
stream_sem = asyncio.Semaphore(50)
async def stream_split(client, payload):
async with connect_sem:
async with client.stream("POST", "/v1/chat", json=payload) as resp:
async with stream_sem:
async for line in resp.aiter_lines():
yield lineThis nuance is critical when you python asyncio rate limit concurrent requests that are long-lived: connection count and active stream count are separate resources.
Step 6: Verify the limit holds
Instrument the live count and write a test that proves the cap.
import prometheus_client as prom
ACTIVE = prom.Gauge("llm_active", "in-flight")
async def call_instrumented(client, payload):
async with sem:
ACTIVE.inc()
try:
return await client.post("/v1/chat", json=payload)
finally:
ACTIVE.dec()A concurrency test using an event to track peak:
def test_peak_concurrency():
sem = asyncio.Semaphore(3)
current = 0
peak = 0
lock = asyncio.Lock()
async def tracked():
nonlocal current, peak
async with sem:
async with lock:
current += 1
peak = max(peak, current)
await asyncio.sleep(0.01)
async with lock:
current -= 1
async def main():
await asyncio.gather(*[tracked() for _ in range(12)])
asyncio.run(main())
assert peak <= 3, f"peak {peak} exceeded cap"Run it:
pytest -q test_concurrency.pySuccess criteria: the test passes, your metrics show the active gauge never above MAX_CONCURRENT during a load run, and you receive no local 429 from self-inflicted burst. If you use a gateway that honors client routing directives, confirm those headers survive the queued client.
Step 7: Tune for production
Derive MAX_CONCURRENT from provider quota and observed latency. If the provider allows 600 requests per minute and each call takes 2 seconds, steady-state concurrency is 600/60*2 = 20. Set the cap at 15–18 to leave headroom for retries and jitter.
For multi-process services, each process needs its own semaphore; a local in-process counter cannot see siblings. Use a distributed lease (Redis) or rely on the gateway’s per-token usage metering to detect global overuse. n4n.ai exposes per-token usage metering that helps you align client caps with account quotas.
Keep queue.maxsize proportional to RAM. A queue holding 50k JSON payloads can OOM before backpressure matters. Start with maxsize = MAX_CONCURRENT * 5 and adjust from metrics.
Final checklist
- Semaphore wraps the actual I/O, not the task creation.
- Queue decouples intake from execution and bounds memory.
- Timeout cancels stuck calls and frees slots.
- Backoff respects
retry-afterand jitters. - Streaming uses a separate connect/stream cap if needed.
- Tests assert peak concurrency; metrics expose it live.
Follow these steps and you will python asyncio rate limit concurrent requests predictably against any LLM endpoint. Replace the mock URL with your real OpenAI-compatible route and ship.