How to simulate traffic spikes against LLM APIs

When you ship a feature that calls an LLM, you need to know how the stack behaves when requests pile up. Simulating traffic spikes on LLM APIs is not the same as hammering a CRUD endpoint—token generation latency and provider rate limits change the failure modes. This tutorial builds a reproducible spike test with Locust and Python so you can see exactly where your timeouts and queues break.

Prerequisites

• Python 3.11 or newer
• pip install locust==2.24.0 (or recent)
• An OpenAI-compatible LLM endpoint and API key. For multi-provider setups, a gateway such as n4n.ai exposes one OpenAI-compatible endpoint for 240+ models with automatic fallback when a provider is degraded; pointing your test at it validates failover under load.
• A cheap, fast model (e.g., gpt-3.5-turbo or a small open-weight model) to keep token spend low.
• export LLM_API_KEY=... and optional LLM_API_BASE, LLM_MODEL.

Step 1: Write a minimal request task

Locust drives virtual users that execute @task methods. The code below posts a tiny chat completion. Keeping max_tokens low controls cost during the test.

import os
from locust import HttpUser, task, between

API_BASE = os.getenv("LLM_API_BASE", "https://api.openai.com/v1")
API_KEY = os.getenv("LLM_API_KEY", "")
MODEL = os.getenv("LLM_MODEL", "gpt-3.5-turbo")

class LLMUser(HttpUser):
    wait_time = between(0.1, 0.5)

    @task
    def chat(self):
        self.client.post(
            f"{API_BASE}/chat/completions",
            headers={
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json",
            },
            json={
                "model": MODEL,
                "messages": [{"role": "user", "content": "Say hi in three words."}],
                "max_tokens": 16,
            },
        )

Run a 30-second flat load to confirm connectivity:

locust -f locustfile.py --headless -u 10 -r 10 -t 30s

Expected output (truncated):

 Name                      # req  # fail |  Avg   Min   Max  Med  req/s
 chat                      310      0   420ms  180ms 1.2s  380ms  10.3

If you see 0 failures and sub-second medians, the baseline works. If you get 401/429, fix auth or rate limits before proceeding.

Step 2: Define a spike shape

A flat concurrent user count does not resemble a real spike—think a Hacker News burst or a cron that fans out to 500 workers. When simulating traffic spikes on LLM APIs, the shape matters more than peak RPS. Locust’s LoadTestShape lets you script user count over time. The shape below holds 5 users, jumps to 200 for 10 seconds, then decays.

from locust import LoadTestShape

class SpikeShape(LoadTestShape):
    def tick(self):
        rt = self.get_run_time()
        if rt < 20:
            return (5, 5)          # warm-up
        elif rt < 30:
            return (200, 200)      # sudden spike: spawn 200 users at once
        elif rt < 50:
            return (5, 50)         # rapid tear-down
        else:
            return None            # stop

Attach the shape by running with --shape SpikeShape. The tuple is (user_count, spawn_rate). Setting spawn rate equal to user count makes the spike instantaneous; a lower rate would ramp gradually.

Step 3: Instrument token usage and failures

Raw response time hides LLM-specific behavior: a request can return 200 but carry truncated output or high token latency. Extend the task to capture usage and flag non-200s explicitly.

@task
def chat_instrumented(self):
    with self.client.post(
        f"{API_BASE}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "model": MODEL,
            "messages": [{"role": "user", "content": "Say hi in three words."}],
            "max_tokens": 16,
        },
        catch_response=True,
    ) as resp:
        if resp.status_code != 200:
            resp.failure(f"HTTP {resp.status_code}")
            return
        try:
            data = resp.json()
        except ValueError:
            resp.failure("non-JSON body")
            return
        comp_tokens = data.get("usage", {}).get("completion_tokens", 0)
        if comp_tokens == 0:
            resp.failure("empty completion")

Locust records each request’s response time automatically. To aggregate tokens per second, hook the request event:

from locust import events

@events.request.add_listener
def log_request(request_type, name, response_time, response_length, exception, **kwargs):
    if exception is None and name == "chat_instrumented":
        # response_length is bytes; we approximate tokens elsewhere
        pass

For a real test, push these to Prometheus or just print sums at the end via the quit event.

Step 4: Execute the spike and read results

Run the shaped test headless for 60 seconds:

locust -f locustfile.py --shape SpikeShape -t 60s --headless

Sample aggregated output during the spike window:

 Name                      # req  # fail |  Avg   Min   Max  Med  req/s
 chat_instrumented         1450    22   1.8s   200ms 14.5s 900ms  24.1

Key signals:

• Failure count jumps during 20–30s: provider 429s or gateway timeouts surface.
• Max latency spikes to 10x median: queueing at the provider, not your app.
• Post-spike recovery: after 30s, med should return near baseline within 5s. If it stays elevated, you have a connection pool leak.

If you front requests with a gateway that provides automatic fallback when a provider is rate-limited, you should see fewer 429s but possibly higher p95 due to retry overhead. That trade-off is exactly what simulating traffic spikes on LLM APIs is meant to expose.

Export for trend analysis

Add --csv=stats to write stats_history.csv. Plot current_rps and fail_percentage to see the spike shape mirrored in backend health. A healthy system shows fail percentage returning to zero within one spawn-rate interval after the spike ends.

Step 5: Validate routing and cache hints

Many teams pin a model or send cache directives to cut cost. If your endpoint honors client routing directives, include them in the headers. For example, a gateway may forward Cache-Control: max-age=3600 to the upstream provider; verify that under spike load the cache hit ratio doesn’t collapse because requests land on different nodes.

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
    "Cache-Control": "max-age=3600",
}

Run the same shape with and without the header. Compare # fail and Avg. A well-behaved gateway (n4n.ai honors client routing directives and forwards provider cache-control hints) should keep hit rate stable because it preserves the directive across the fan-out.

Also vary prompt size. Production traffic rarely sends 3-word prompts; inject a random max_tokens between 16 and 256 to mimic real variance:

import random
"max_tokens": random.choice([16, 64, 128, 256]),

This widens the latency distribution and exposes provider queueing that a fixed tiny request hides.

Step 6: Bound the cost

LLM load tests can get expensive if a bug loops. Compute a worst-case spend before running:

requests = integral of user_count over time ≈ (5*20)+(200*10)+(5*20) = 2300
max_tokens_per_req = 16
total_completion_tokens = 2300 * 16 = 36,800

At typical prices (e.g., $0.002/1K tokens for a small model) that’s under $0.10. Still, set a hard cap with environment variable and a guard in the task:

import os
MAX_REQ = int(os.getenv("MAX_REQ", "5000"))
if self.environment.runner.stats.total.num_requests >= MAX_REQ:
    self.environment.runner.quit()

Per-token usage metering on your gateway makes this accounting exact—read the usage block from each response and sum it rather than estimating.

Step 7: Streaming spikes (optional)

If your app uses streaming, the server sends tokens over time. Locust can handle this by reading the response body incrementally, but you must define “success” as receiving the data: [DONE] marker. Use stream=True in the client post and iterate resp.iter_lines(). The latency metric then measures time-to-first-token plus download time, which is the right SLO for chat UIs.

@task
def chat_stream(self):
    with self.client.post(
        f"{API_BASE}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={"model": MODEL, "messages": [{"role": "user", "content": "Say hi."}], "stream": True},
        stream=True,
        catch_response=True,
    ) as resp:
        done = False
        for line in resp.iter_lines():
            if line and b"[DONE]" in line:
                done = True
                break
        if not done:
            resp.failure("stream closed before DONE")

Streaming changes the concurrency profile: connections stay open longer, so your spike will exhaust file descriptors before CPU. Simulate with lower user counts (e.g., 50 instead of 200) to find that limit.

What to do with the data

After simulating traffic spikes on LLM APIs a few times, you will have a clear picture of three thresholds: the concurrency at which p95 doubles, the point where providers start 429ing, and the recovery slope. Feed those numbers into your autoscaler and your circuit breaker. If you only test flat load, you will miss the transient failure that hits at 3 a.m. when a batch job fires.

Run this suite in CI against a stub model before every deploy. It takes five minutes and catches the regression where someone lowered the client timeout below the provider’s p99.