AI Load Testing: Your LLM Endpoint Will Buckle Under Real Traffic

TL;DR

LLM endpoints have unique failure modes: streaming interruptions, token exhaustion, cascading timeouts Load testing must measure: latency percentiles, token throughput, cache hit rates, cost under load Streaming responses fail ungracefully unless explicitly handled, clients see partial responses Rate limiting must be intelligent: per-token, per-user, with graceful degradation Test at 3-5x expected traffic; most AI endpoints fail before traditional systems Cost modeling during peak load is critical, a traffic spike can cost thousands per hour

Why Traditional Load Testing Doesn't Work

Traditional load testing assumes request-response pairs: fast requests, small payloads, clear pass/fail semantics. LLM endpoints violate all these assumptions.

The Unique Challenges

Long-tailed latency: A typical API request takes 50ms. An LLM request takes 2-8 seconds. This changes everything. Your timeout defaults are wrong. Your queue strategy is wrong. Your retry logic cascades failures differently.

Streaming responses: Clients don't wait for a complete response. They start receiving tokens immediately. If the connection drops midway, the client has a partial response. Your monitoring needs to understand this.

Variable token consumption: You can't predict response length. Some prompts return 10 tokens. Others return 1000. This makes capacity planning nearly impossible and cost modeling a moving target.

Cascading failures: When one LLM API goes down, your retry logic hits it again. And again. You end up amplifying the failure instead of gracefully degrading.

Context window exhaustion: If you're using conversation history, each request consumes tokens just to send context. Under load, context windows fill faster. Quality degrades. Cost spirals.

What to Measure During Load Testing

Latency Percentiles (Not Averages)

Average response time is useless. You need percentiles. The 50th percentile (median) tells you about typical users. The 95th percentile tells you about frustrated users. The 99th percentile tells you about your SLA.

import numpy as np
from concurrent.futures import ThreadPoolExecutor
import time

def load_test_llm_endpoint(endpoint, num_requests=1000, concurrency=50):
 latencies = []

 def make_request():
 start = time.time()
 try:
 response = endpoint.invoke(prompt="test")
 elapsed = time.time() - start
 latencies.append(elapsed)
 return elapsed
 except Exception as e:
 latencies.append(None) # Track failures
 return None

 with ThreadPoolExecutor(max_workers=concurrency) as executor:
 executor.map(make_request, range(num_requests))

 # Calculate percentiles
 valid_latencies = [l for l in latencies if l is not None]
 print(f"Median latency: {np.percentile(valid_latencies, 50):.2f}s")
 print(f"p95 latency: {np.percentile(valid_latencies, 95):.2f}s")
 print(f"p99 latency: {np.percentile(valid_latencies, 99):.2f}s")
 print(f"Max latency: {max(valid_latencies):.2f}s")
 print(f"Error rate: {(len(latencies) - len(valid_latencies)) / len(latencies) * 100:.1f}%")

Your p95 latency is what users experience. If it's 15 seconds, users will timeout. Your p99 latency is your worst-case SLA. If you promise 99% of requests within 20 seconds, you need p99 below 20s.

Token Throughput

Don't measure requests per second. Measure tokens per second. This is what matters for capacity and cost.

def measure_token_throughput(endpoint, duration_seconds=60, concurrency=50):
 total_tokens_consumed = 0
 total_tokens_generated = 0
 start_time = time.time()

 def consume_tokens():
 nonlocal total_tokens_consumed, total_tokens_generated
 while time.time() - start_time < duration_seconds:
 prompt = "Tell me about machine learning"
 result = endpoint.invoke(prompt=prompt)

 # Count tokens
 prompt_tokens = count_tokens(prompt)
 response_tokens = count_tokens(result.content)

 total_tokens_consumed += prompt_tokens
 total_tokens_generated += response_tokens

 with ThreadPoolExecutor(max_workers=concurrency) as executor:
 executor.map(consume_tokens, range(concurrency))

 elapsed = time.time() - start_time
 print(f"Input tokens/sec: {total_tokens_consumed / elapsed:.0f}")
 print(f"Output tokens/sec: {total_tokens_generated / elapsed:.0f}")
 print(f"Total tokens/sec: {(total_tokens_consumed + total_tokens_generated) / elapsed:.0f}")

This metric is critical. Your LLM provider bills you per token. Under load, token throughput is directly proportional to cost. If you're generating 10k tokens/second and that costs $0.10 per 1M tokens, you're burning $1/second under peak load.

Cache Hit Rates

If you're using prompt caching (smart move), measure hit rates under load. This is where you save money and latency.

class CachingLLMEndpoint:
 def __init__(self):
 self.cache = {}
 self.hits = 0
 self.misses = 0

 def invoke(self, prompt: str) -> str:
 cache_key = hash(prompt)

 if cache_key in self.cache:
 self.hits += 1
 return self.cache[cache_key]
 else:
 self.misses += 1
 result = self.llm.invoke(prompt)
 self.cache[cache_key] = result
 return result

 @property
 def cache_hit_rate(self) -> float:
 total = self.hits + self.misses
 return self.hits / total if total > 0 else 0.0

# Under load, monitor hit rate
endpoint = CachingLLMEndpoint()
# ... run load test ...
print(f"Cache hit rate: {endpoint.cache_hit_rate * 100:.1f}%")

# Healthy systems see 40-60% hit rates under normal load
# Below 20% suggests prompts are too varied
# Above 80% suggests queries are repetitive

Cost Projection

Track cost per request under load. This is the metric your CFO cares about.

def project_cost_under_load(tokens_per_second, cost_per_million_tokens=0.10):
 """Project hourly cost based on token throughput"""
 tokens_per_hour = tokens_per_second * 3600
 cost_per_hour = (tokens_per_hour / 1_000_000) * cost_per_million_tokens

 print(f"Tokens/second: {tokens_per_second:.0f}")
 print(f"Tokens/hour: {tokens_per_hour:.0f}")
 print(f"Cost/hour: ${cost_per_hour:.2f}")
 print(f"Cost/day (24h): ${cost_per_hour * 24:.2f}")
 print(f"Cost/month (30d): ${cost_per_hour * 24 * 30:.2f}")

 # Example: 1000 tokens/second at $0.10 per 1M tokens
 project_cost_under_load(1000)

At 1000 tokens/second, you're spending $25.92/hour. At 5000 tokens/second, you're spending $129.60/hour. This is why load testing matters. A 5x traffic spike becomes a $100/hour problem.

How AI Endpoints Fail

Failure Mode 1: Streaming Interruptions

The client receives tokens one by one. If the connection drops at token 500 of 1000, the client gets a partial response. Your application needs to handle this gracefully.

async def stream_with_resilience(endpoint, prompt: str, max_retries: int = 3):
 tokens_received = []

 for attempt in range(max_retries):
 try:
 async for token in endpoint.stream(prompt):
 tokens_received.append(token)
 except ConnectionError:
 # Connection dropped, retry with partial completion
 if attempt < max_retries - 1:
 await asyncio.sleep(2 ** attempt) # Exponential backoff
 continue
 else:
 # Final attempt failed
 break

 # Return best-effort response
 return "".join(tokens_received)

Streaming without retry logic is dangerous. Under load, network conditions degrade. Connections drop. You need explicit handling.

Failure Mode 2: Token Budget Exhaustion

Your LLM has a token limit (context window). If you're storing conversation history, each new request uses tokens just for context. Under load with long conversations, you hit the limit.

class ConversationWithTokenBudget:
 def __init__(self, max_context_tokens: int = 4000):
 self.history = []
 self.max_context_tokens = max_context_tokens

 def add_exchange(self, user_message: str, assistant_response: str):
 """Add conversation turn, evict old messages if needed"""
 new_tokens = count_tokens(user_message) + count_tokens(assistant_response)

 # Check if we're within budget
 current_tokens = sum(count_tokens(m) for m in self.history)
 while current_tokens + new_tokens > self.max_context_tokens:
 # Remove oldest message
 self.history.pop(0)
 current_tokens = sum(count_tokens(m) for m in self.history)

 self.history.append(user_message)
 self.history.append(assistant_response)

Without explicit token budget management, your system fails when context fills. Messages get dropped silently, or requests fail with mysterious errors.

Failure Mode 3: Cascading Timeouts

Your timeout is set to 30 seconds. Under load, latency hits 25 seconds. Your retry logic kicks in, making a second request. Now you're at 50 seconds. The client timeout fires. You've wasted capacity and frustrated the user.

def invoke_with_smart_retries(endpoint, prompt: str, base_timeout: float = 30.0):
 """Retry logic that doesn't cascade failures"""
 max_attempts = 2
 attempt = 1

 while attempt <= max_attempts:
 # Scale timeout by attempt number
 # First attempt: 30s, second attempt: 45s
 timeout = base_timeout * (1 + (attempt - 1) * 0.5)

 try:
 return endpoint.invoke(prompt, timeout=timeout)
 except TimeoutError:
 if attempt < max_attempts:
 attempt += 1
 else:
 raise

 raise TimeoutError("Max retries exceeded")

Smart retry logic waits longer on subsequent attempts. It doesn't assume the first timeout was a transient blip. It backs off and gives the system time to recover.

Failure Mode 4: Cost Explosion

Under load, context windows grow, retry logic kicks in, and you're generating tokens faster than ever. Your bill becomes 10x larger than expected.

class CostMonitor:
 def __init__(self, cost_limit_per_hour: float = 100.0):
 self.cost_limit = cost_limit_per_hour
 self.current_hour_cost = 0.0
 self.start_time = time.time()

 def track_request(self, input_tokens: int, output_tokens: int, cost_per_mtok: float = 0.001):
 """Track cost and alert if budget exceeded"""
 total_tokens = input_tokens + output_tokens
 cost = (total_tokens / 1_000_000) * (cost_per_mtok * 1000)
 self.current_hour_cost += cost

 if self.current_hour_cost > self.cost_limit:
 alert(f"Cost limit exceeded: ${self.current_hour_cost:.2f}/hour")

 elapsed_hours = (time.time() - self.start_time) / 3600
 if elapsed_hours >= 1.0:
 # Reset for next hour
 self.current_hour_cost = 0.0
 self.start_time = time.time()

Without cost monitoring, a traffic spike silently becomes a thousand-dollar incident. Monitor it explicitly.

Load Testing Strategies

Baseline Testing (1x expected load)

Test at your expected normal traffic. This establishes baselines for latency, throughput, and cost. You'll find 90% of issues here.

# Expected: 100 RPS, average response 3 seconds
load_test(
 endpoint=my_endpoint,
 requests_per_second=100,
 concurrency=300,
 duration_seconds=300 # 5 minute test
)

Stress Testing (3-5x expected load)

Push beyond expected limits. This is where you find failure modes. Most AI endpoints degrade gracefully up to 2-3x load, then fail hard at 4-5x.

# Expected: 100 RPS, stressing at 300-500 RPS
load_test(
 endpoint=my_endpoint,
 requests_per_second=400,
 concurrency=1200,
 duration_seconds=120 # Shorter test at higher load
)

Sustained Load Testing (8+ hours)

Long-running tests reveal memory leaks, connection pool exhaustion, and cache degradation. AI endpoints need this.

load_test(
 endpoint=my_endpoint,
 requests_per_second=150,
 concurrency=450,
 duration_seconds=28800 # 8 hours
)

# Monitor for:
# - Memory growth over time
# - Connection pool saturation
# - Cache hit rate degradation
# - Increasing latency percentiles

Designing Resilient AI Systems

1. Implement Circuit Breakers

When downstream services (LLM APIs) fail, stop sending requests. Return a degraded response instead.

class CircuitBreaker:
 def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
 self.failures = 0
 self.threshold = failure_threshold
 self.timeout = timeout_seconds
 self.last_failure_time = None
 self.state = "closed" # closed, open, half-open

 def call(self, func, *args, **kwargs):
 if self.state == "open":
 if time.time() - self.last_failure_time > self.timeout:
 self.state = "half-open"
 else:
 raise Exception("Circuit breaker open")

 try:
 result = func(*args, **kwargs)
 self.failures = 0
 self.state = "closed"
 return result
 except Exception as e:
 self.failures += 1
 self.last_failure_time = time.time()
 if self.failures >= self.threshold:
 self.state = "open"
 raise

2. Rate Limiting with Token Awareness

Limit by tokens, not requests. This prevents cost explosion.

class TokenRateLimiter:
 def __init__(self, tokens_per_second: int = 10000):
 self.limit = tokens_per_second
 self.current_tokens = tokens_per_second
 self.last_refill = time.time()

 def allow_request(self, estimated_tokens: int) -> bool:
 now = time.time()
 elapsed = now - self.last_refill

 # Refill at rate
 refill = elapsed * self.limit
 self.current_tokens = min(self.limit, self.current_tokens + refill)
 self.last_refill = now

 if self.current_tokens >= estimated_tokens:
 self.current_tokens -= estimated_tokens
 return True
 return False

3. Graceful Degradation

When your LLM endpoint is overloaded, return a cached response or a simpler model.

def invoke_with_fallback(primary_endpoint, fallback_endpoint, prompt: str):
 try:
 # Try primary (expensive, slow, high-quality)
 return primary_endpoint.invoke(prompt, timeout=10.0)
 except (TimeoutError, RateLimitError):
 # Fall back to cache or simpler model
 if cached_response := check_cache(prompt):
 return cached_response
 return fallback_endpoint.invoke(prompt) # Faster, cheaper, lower-quality

The Bottom Line

Before you launch an AI-powered feature, load test it. Not "when you have time." Before. This is table stakes. Test at 3x expected load. Measure latency percentiles, token throughput, cache hit rates, and cost. Find the failure modes. Fix them.

Then deploy to production knowing you won't wake up at 3am to a bill for $50,000.