LLM applications have non-linear failure modes: a request that takes 800ms under 10 concurrent users can balloon to 15s under 200. Load testing finds your breaking point in staging — not in production at 2am.
Select a metric to understand what it measures, what a good target looks like, and how to fix problems:
# pip install locust
# locustfile.py
from locust import HttpUser, task, between
import random
SAMPLE_QUERIES = [
"What is machine learning?",
"Explain gradient descent",
"How does attention work in transformers?",
"What is RAG?",
"Compare BERT and GPT",
]
class LLMUser(HttpUser):
wait_time = between(1, 3) # Simulate realistic user pacing
@task(3)
def simple_query(self):
"""High-frequency: simple factual queries."""
self.client.post(
"/chat",
json={"message": random.choice(SAMPLE_QUERIES[:2])},
timeout=10,
)
@task(1)
def complex_query(self):
"""Low-frequency: complex reasoning tasks."""
self.client.post(
"/chat",
json={"message": "Analyze the pros and cons of using RAG vs fine-tuning for a customer support bot"},
timeout=30,
)
# Run:
# locust -f locustfile.py --host http://localhost:8000
# Then open http://localhost:8089 and set:
# - Users: 50 (ramp up)
# - Spawn rate: 5 users/second
# - Run for: 5 minutesimport asyncio
import httpx
import time
import statistics
async def single_request(client: httpx.AsyncClient, prompt: str) -> dict:
start = time.monotonic()
try:
response = await client.post(
"http://localhost:8000/chat",
json={"message": prompt},
timeout=15.0,
)
latency = (time.monotonic() - start) * 1000
return {"success": response.status_code == 200, "latency_ms": latency}
except Exception as e:
return {"success": False, "latency_ms": (time.monotonic() - start) * 1000, "error": str(e)}
async def load_test(
prompts: list[str],
concurrent_users: int = 20,
total_requests: int = 100,
) -> dict:
results = []
sem = asyncio.Semaphore(concurrent_users)
async def bounded_request(prompt: str):
async with sem:
return await single_request(client, prompt)
async with httpx.AsyncClient() as client:
tasks = [bounded_request(prompts[i % len(prompts)]) for i in range(total_requests)]
results = await asyncio.gather(*tasks)
latencies = [r["latency_ms"] for r in results if r["success"]]
errors = sum(1 for r in results if not r["success"])
return {
"p50_ms": statistics.median(latencies),
"p95_ms": sorted(latencies)[int(len(latencies) * 0.95)],
"p99_ms": sorted(latencies)[int(len(latencies) * 0.99)],
"error_rate": errors / len(results),
"rps_throughput": len(results) / (max(latencies) / 1000),
}
results = asyncio.run(load_test(SAMPLE_QUERIES, concurrent_users=20, total_requests=100))
print(f"p50: {results['p50_ms']:.0f}ms p95: {results['p95_ms']:.0f}ms errors: {results['error_rate']:.1%}")Run load tests at 2× your expected peak. If you expect 100 concurrent users at launch, load test at 200. LLM latency degrades non-linearly — the system that handles 100 users fine may collapse at 150 if your connection pool is exhausted.
You now have the tools to measure your system under pressure. In Lesson 7, we look at how to iterate safely using A/B testing for AI features.