Most teams discover LLM gateway limits only after a real outage. Synthetic load tests for chat completions let you rehearse those failure modes on your terms, with controlled concurrency and a realistic prompt mix. This tutorial builds a runnable harness in Python and then wraps it in Locust for distributed runs.
Prerequisites
- Python 3.11 or newer
pip install aiohttp numpy locust- An OpenAI-compatible chat completions endpoint and API key. If you point the harness at n4n.ai, the same OpenAI-compatible endpoint fronts 240+ models with automatic fallback when a provider is degraded, but the request shape below is identical for any compliant server.
- Basic comfort with async Python and a terminal.
Designing a synthetic prompt mix
Random token soup does not exercise the same code paths as production traffic. Real chat workloads have skewed distributions: short coding questions, long document summaries, occasional heavy SQL generation. Build a weighted template set so your test reflects that.
import random
PROMPT_TEMPLATES = [
{"system": "You are a terse helper.", "user": "Summarize: {text}", "weight": 0.5},
{"system": "You are a coding assistant.", "user": "Fix this Python: {code}", "weight": 0.3},
{"system": "You are a SQL expert.", "user": "Write a query for {schema}", "weight": 0.2},
]
CORPUS = {
"text": ["Long article about distributed systems. " * 30, "Quarterly report excerpt. " * 25],
"code": ["def foo():\n return 1", "import os\nprint(os.getcwd())"],
"schema": ["users(id, name, created_at)", "orders(id, total, user_id)"],
}
def build_messages(rng: random.Random):
tmpl = rng.choices(PROMPT_TEMPLATES, weights=[t["weight"] for t in PROMPT_TEMPLATES])[0]
if "{text}" in tmpl["user"]:
fill = rng.choice(CORPUS["text"])
elif "{code}" in tmpl["user"]:
fill = rng.choice(CORPUS["code"])
else:
fill = rng.choice(CORPUS["schema"])
return [
{"role": "system", "content": tmpl["system"]},
{"role": "user", "content": tmpl["user"].format(text=fill, code=fill, schema=fill)},
]A summary prompt pushes ~600 input tokens; a code fix maybe 50. If you ignore this variance, your synthetic load tests for chat completions will over-represent cheap requests and underestimate prefill compute. Keep the weights honest.
Async client for non-streaming requests
We use aiohttp to fire concurrent POSTs at /v1/chat/completions. Capture status, latency, and token usage.
import aiohttp, time, os
ENDPOINT = os.environ["LLM_ENDPOINT"]
API_KEY = os.environ["LLM_API_KEY"]
async def send_completion(session, messages, model="gpt-4o-mini"):
payload = {"model": model, "messages": messages, "max_tokens": 128, "temperature": 0.7}
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
start = time.monotonic()
async with session.post(f"{ENDPOINT}/chat/completions", json=payload, headers=headers) as resp:
status = resp.status
if status != 200:
body = await resp.text()
return {"status": status, "error": body[:200], "latency": time.monotonic()-start}
data = await resp.json()
return {
"status": 200,
"latency": time.monotonic() - start,
"completion_tokens": data.get("usage", {}).get("completion_tokens", 0),
}Running bounded concurrent load
A semaphore caps in-flight requests so you don’t accidentally DOS your own laptop or blow through provider RPM in one second. Collect results in a list.
import asyncio
async def worker(session, sem, rng, results, model):
async with sem:
msgs = build_messages(rng)
res = await send_completion(session, msgs, model)
results.append(res)
async def run_load(concurrency=50, total=500, model="gpt-4o-mini"):
rng = random.Random(42)
sem = asyncio.Semaphore(concurrency)
results = []
async with aiohttp.ClientSession() as session:
tasks = [worker(session, sem, rng, results, model) for _ in range(total)]
await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
results = asyncio.run(run_load(concurrency=20, total=100))
ok = [r for r in results if r["status"] == 200]
print(f"OK: {len(ok)}/{len(results)}")
if ok:
avg_lat = sum(r["latency"] for r in ok) / len(ok)
avg_tok = sum(r["completion_tokens"] for r in ok) / len(ok)
print(f"Avg latency: {avg_lat:.2f}s, avg completion tokens: {avg_tok:.1f}")Expected output on a healthy endpoint:
OK: 98/100
Avg latency: 1.34s, avg completion tokens: 112.3The two failures are likely 429s if you exceed provider RPM. That is exactly the signal you want—it proves your client handles rejection instead of hanging.
Why not just curl in a loop?
Shell loops can’t hold 200 concurrent connections with accurate timers, and they sure can’t parse streaming SSE without awkward awk. Python’s async model gives you precise monotonic clocks and structured result aggregation with less than 60 lines.
Measuring time-to-first-token with streaming
Production chat UIs stream. Your synthetic load tests for chat completions should too, because TTFT behaves differently than full-response latency under congestion.
async def send_stream(session, messages, model="gpt-4o-mini"):
payload = {"model": model, "messages": messages, "stream": True, "max_tokens": 128}
headers = {"Authorization": f"Bearer {API_KEY}"}
start = time.monotonic()
ttft = None
tokens = 0
async with session.post(f"{ENDPOINT}/chat/completions", json=payload, headers=headers) as resp:
async for line in resp.content:
if not line.startswith(b"data:"):
continue
chunk = line[5:].strip()
if chunk == b"[DONE]":
break
if ttft is None:
ttft = time.monotonic() - start
tokens += 1
return {"ttft": ttft, "tokens": tokens, "total": time.monotonic() - start}Swap send_completion for send_stream in the worker to capture TTFT percentiles. Under load, TTFT often climbs before full latency does, because the gateway queues requests waiting for a free slot on the model replica.
Scaling with Locust
For distributed load, Locust drives many workers from one master. Reuse the prompt builder.
from locust import HttpUser, task, between
import random, os
ENDPOINT = os.environ["LLM_ENDPOINT"]
API_KEY = os.environ["LLM_API_KEY"]
class ChatUser(HttpUser):
wait_time = between(0.1, 1.0)
def on_start(self):
self.model = "gpt-4o-mini"
@task
def complete(self):
msgs = build_messages(random.Random())
self.client.post(
f"{ENDPOINT}/chat/completions",
json={"model": self.model, "messages": msgs, "max_tokens": 128},
headers={"Authorization": f"Bearer {API_KEY}"},
name="chat_completion"
)Launch a headless run with ramp-up:
locust -f locustfile.py --headless -u 200 -r 20 -t 2m --csv=reportSample aggregated output (truncated):
Name # reqs # fails Avg p95
chat_completion 2400 42 1.42s 3.10sLocust makes synthetic load tests for chat completions reproducible across a cluster and gives you CSVs for plotting. Add --processes 4 to use multiple cores if a single Python worker becomes the bottleneck.
Footguns and interpretation
Per-token metering means a 10k-request test at 128 output tokens costs real money. Run the local harness first to get average completion tokens, multiply by total requests, and check your provider’s published price before scaling.
If your gateway honors cache-control hints (n4n.ai forwards them), tag static system prompts with "cache_control": {"type": "ephemeral"} to exercise cached prefill paths. That separates cache-hit latency from cold prefill latency—two very different curves.
Watch the 429 rate. A gateway with automatic fallback will reroute to a secondary provider; your assertion should be that error rate stays under 1% even when you inject artificial latency via a proxy. Use the status field in results to compute that.
Finally, synthetic load tests for chat completions must include both streaming and non-streaming shapes. Streaming hides queueing behind first-token latency; non-streaming exposes full worker saturation. Run both before trusting capacity numbers.
Checkpoint: full harness
You now have a weighted prompt sampler, an async client for two modes, a bounded local runner, and a Locust wrapper. Point LLM_ENDPOINT at your gateway, set a low total first, and watch the error bucket. Then raise concurrency until p95 breaks your SLO. That number is your real capacity—not the marketing slide.