Most production incidents with LLM integrations trace back to untested failure modes at the network boundary. If you skip LLM API timeout and retry testing, you will discover your retry storm only when a provider degrades during peak traffic. This how-to gives you a repeatable procedure to validate client resilience under concurrent load and injected faults.
Step 1: Define explicit timeout and retry budgets
Set client-side limits before writing any test. An LLM call is an HTTP request; treat it like one. Use connect, read, and total timeouts, and cap retries to avoid amplification. Default SDK timeouts are far too lenient for production—the OpenAI Python client historically defaulted to 10 minutes, which will queue your event loop into oblivion.
Document your LLM API timeout and retry testing goals up front: maximum acceptable latency per call, total attempts, and expected backend request multiplier. Two retries means at most three attempts. Under load, that triples your request volume if the backend is slow. If you serve 500 requests per second, a 30-second provider outage with two retries generates 1,500 outbound requests per second against a struggling dependency.
With the OpenAI Python client pointed at any OpenAI-compatible endpoint:
from openai import OpenAI
client = OpenAI(
base_url="https://api.example.com/v1", # swap for your gateway
api_key="sk-...",
timeout=10.0, # total request timeout in seconds
max_retries=2, # client-side retries on connection/5xx
)For finer control, configure transport retries with httpx:
import httpx
from openai import OpenAI
transport = httpx.HTTPTransport(retries=2)
client = OpenAI(
base_url="https://api.example.com/v1",
api_key="sk-...",
timeout=httpx.Timeout(connect=2.0, read=8.0, total=10.0),
http_client=httpx.Client(transport=transport),
)Step 2: Build a fault-injecting mock endpoint
You cannot reliably test against a live provider’s degradation. Stand up a local FastAPI service that mimics an inference endpoint with configurable latency and error rates. The mock from Step 2 is the backbone of LLM API timeout and retry testing because it removes provider flakiness and gives you deterministic chaos.
Use asyncio.sleep instead of time.sleep so the event loop stays free for concurrent requests:
from fastapi import FastAPI, Response, status
import asyncio, random, os
app = FastAPI()
SLOW_RATE = float(os.getenv("SLOW_RATE", "0.2"))
FAIL_429 = float(os.getenv("FAIL_429", "0.1"))
FAIL_503 = float(os.getenv("FAIL_503", "0.05"))
@app.post("/v1/chat/completions")
async def fake_completions(response: Response):
if random.random() < SLOW_RATE:
await asyncio.sleep(12) # breach a 10s client timeout
if random.random() < FAIL_429:
response.status_code = status.HTTP_429_TOO_MANY_REQUESTS
return {"error": "rate limited"}
if random.random() < FAIL_503:
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return {"error": "unavailable"}
await asyncio.sleep(0.2)
return {
"choices": [{"message": {"role": "assistant", "content": "ok"}}],
"usage": {"total_tokens": 10},
}Run it with uvicorn mock:app --port 8000. Point your test client at http://localhost:8000/v1.
Step 3: Generate concurrent load
A single threaded loop hides connection pool exhaustion. Use an async load generator that fires many requests and records outcomes. The default AsyncOpenAI client uses an httpx async pool limited to 100 connections; push concurrency past that to surface pool-wait deadlocks.
import asyncio, time
from openai import AsyncOpenAI
async def hit(client, i):
try:
await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"ping {i}"}],
)
return "ok"
except Exception as e:
return f"fail:{type(e).__name__}"
async def load(n, concurrency):
client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="x")
sem = asyncio.Semaphore(concurrency)
async def bounded(i):
async with sem:
return await hit(client, i)
start = time.monotonic()
results = await asyncio.gather(*[bounded(i) for i in range(n)])
elapsed = time.monotonic() - start
ok = sum(1 for r in results if r == "ok")
print(f"{ok}/{n} succeeded in {elapsed:.1f}s")
asyncio.run(load(200, 50))Run with python load.py. Tune n and concurrency to exceed your expected production peak. If you see TimeoutError on the client before the mock responds, your pool or timeout is mis-sized.
Step 4: Validate retry and fallback interactions
Client retries handle transient 5xx, but a gateway-level fallback changes the equation. An inference gateway such as n4n.ai performs automatic fallback when a provider is rate-limited or degraded, which means your client may see fewer errors than expected if you route through it. To test your own retry logic in isolation, bypass fallback by sending a routing directive that pins a single backend, or test directly against the mock.
If your gateway honors client routing headers, force a specific provider:
# Example header forwarded to provider; exact name depends on gateway
client.chat.completions.create(
model="provider/model",
messages=[{"role": "user", "content": "test"}],
extra_headers={"x-n4n-route": "provider-a"},
)Then kill provider-a in your mock (return 503 always) to confirm the client exhausts retries and surfaces a clear error rather than hanging. Without the pin, the gateway might silently reroute to provider-b, masking your broken retry code.
Step 5: Measure what matters
Raw success rate is not enough. Capture per-attempt latency, retry counts, and token metering if your endpoint supports it. Extend the load script to log attempts:
import logging
logging.basicConfig(level=logging.INFO)
async def hit_logged(client, i):
for attempt in range(3):
try:
resp = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"ping {i}"}],
)
logging.info(f"req {i} ok tokens={resp.usage.total_tokens}")
return "ok"
except Exception as e:
logging.warning(f"req {i} attempt {attempt} failed: {e}")
return "exhausted"If you use a gateway with per-token usage metering, compare the sum of usage.total_tokens against your billing export after the run to confirm metering survives retries (retried failed attempts should not bill). In LLM API timeout and retry testing, token accounting under failure is a common gap—teams assume zero tokens on failed calls, but some providers bill for partial completions.
Step 6: Run targeted chaos scenarios
Execute these in order, each against a fresh mock configuration:
- Timeout breach: Set
SLOW_RATE=1.0with 15s sleep, client timeout 5s. Verify all calls fail fast and retries do not exceed budget. - 429 storm: Set
FAIL_429=1.0. Confirm client backs off (theopenaiclient applies exponential backoff) and fails aftermax_retries. - Connection reset: Kill the uvicorn process mid-run. Ensure client raises
ConnectionErrornot deadlock. - Partial degradation:
SLOW_RATE=0.5,FAIL_503=0.5. Measure p95 latency and ensure event loop stays responsive (other coroutines should not stall).
Each scenario should be repeated at concurrency 10, 100, and 250 to expose pool limits.
Verify success
Your LLM API timeout and retry testing is complete when the following hold:
- No request hangs beyond
timeout * (max_retries + 1)plus backoff ceiling. - Under 50 concurrent calls with 20% timeout rate, the process does not OOM or block on pool acquisition.
- Retry exhaustion produces a typed exception logged with request ID and attempt count.
- Token counts from successful responses match expectations; failed retries show zero tokens on metered gateways.
- A pinned-route degradation test surfaces the error to your application layer instead of silently succeeding via fallback.
- Connection drop mid-flight triggers immediate client error, not a hung await.
Run this suite in CI against the mock on every client SDK bump. The cost is one afternoon of setup; the alternative is a retry amplification incident at 2 a.m. when your primary model provider has a bad day.