# LLM API Benchmarking & Stress Testing

Patterns for benchmarking OpenAI-compatible LLM APIs with concurrent requests, latency measurement, and throughput analysis. Developed from aiand.com stress testing.

## When to use

- User asks to benchmark/models/throughput of an LLM API
- User wants concurrent stress test of API keys
- User asks to compare model performance (latency, TPS, tokens)
- User has a pool of API keys and wants to test capacity

## Curl-based approach (bypasses Cloudflare)

Many LLM API providers use Cloudflare protection that blocks Python HTTP libraries (`requests`, `aiohttp`, `urllib`). The system `curl` binary passes through because it uses native TLS and different fingerprinting. Use `subprocess.run`:

```python
import subprocess, json, time

def call_curl(key, model, prompt, api_url, max_tokens=50):
    body = json.dumps({
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": max_tokens
    })
    t0 = time.perf_counter()
    r = subprocess.run([
        "curl", "-s", "-o", "-", "-w", "|%{http_code}|%{time_total}",
        "-X", "POST", api_url,
        "-H", f"Authorization: Bearer {key}",
        "-H", "Content-Type: application/json",
        "-d", body,
        "--max-time", "45"
    ], capture_output=True, text=True, timeout=50)
    t = time.perf_counter() - t0
    
    # Parse: body text before last two |-separated fields
    out = r.stdout.strip()
    parts = out.rsplit('|', 2)
    if len(parts) >= 3:
        body_text, http_code, curl_time = parts[0], parts[1], parts[2]
        if http_code == "200":
            data = json.loads(body_text)
            msg = data["choices"][0]["message"]
            # Check BOTH content and reasoning (reasoning models)
            txt = msg.get("content") or msg.get("reasoning", "")
            tok = data.get("usage", {}).get("completion_tokens", 0)
            return {"ok": True, "latency": round(t, 3), "tokens": tok, "output": txt[:50]}
    return {"ok": False, "error": "parse"}
```

## Concurrent stress test pattern

Use `ThreadPoolExecutor` with `subprocess` (not `asyncio` with `aiohttp` — blocked by Cloudflare):

```python
import concurrent.futures, statistics, time

def stress_test(keys, model, api_url, concurrency=3):
    """Run concurrency requests per key, measure throughput."""
    all_results = []
    t0 = time.perf_counter()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(keys)*concurrency, 50)) as ex:
        futures = []
        for key in keys:
            for _ in range(concurrency):
                futures.append(ex.submit(call_curl, key, model, "Say hi", api_url))
        for f in concurrent.futures.as_completed(futures):
            all_results.append(f.result())
    
    total_t = time.perf_counter() - t0
    oks = [r for r in all_results if r.get("ok")]
    
    return {
        "total": len(all_results),
        "ok": len(oks),
        "fail": len(all_results) - len(oks),
        "time": total_t,
        "throughput": len(oks)/total_t,
        "latency_avg": statistics.mean([r["latency"] for r in oks]) if oks else 0,
        "latency_p50": statistics.median([r["latency"] for r in oks]) if oks else 0,
        "latency_min": min([r["latency"] for r in oks]) if oks else 0,
        "latency_max": max([r["latency"] for r in oks]) if oks else 0,
        "total_tokens": sum([r["tokens"] for r in oks]) if oks else 0,
        "total_tps": sum([r["tokens"] for r in oks])/total_t if oks else 0,
    }
```

## Metrics to collect

| Metric | What it tells you |
|---|---|
| `throughput` (req/s) | API capacity under load |
| `latency_avg` | Typical response time |
| `latency_p50` | Median — better than avg for skewed distributions |
| `latency_min/max` | Range — high max = tail latency issues |
| `total_tps` (tok/s) | Real token generation speed |
| `fail rate` | Reliability under concurrency |

## Key parsing from bulk files

Common format: `email|sk-xxx|org_id`. Don't split once and treat `sk-xxx|org_id` as the token:

[REDACTED]
def load_keys(path):
    keys = []
    for line in Path(path).read_text().strip().split('\n'):
        parts = line.strip().split('|')
        for p in parts:
            if p.startswith('sk-') and len(p) > 40:
                keys.append(p)
                break
    return keys
```

## GitHub tools for benchmarking

Quick-start tools that can be cloned and adapted:
- `theodoreniu/llmperf` — Python/Streamlit, comprehensive LLM benchmarking
- `gouwsxander/stress-test` — Python/aiohttp, simple stress testing (needs curl adaptation for CF-protected APIs)
- `Yoosu-L/llmapibenchmark` — Go-based, measures API throughput across concurrency levels

## Pitfalls

1. **Python HTTP blocked by Cloudflare** — always use `subprocess` + `curl` for CF-protected APIs
2. **Reasoning models return null content** — check `reasoning` field, not just `content`
3. **Model list ≠ working models** — `GET /v1/models` may return models that don't exist on the host
4. **Terminal truncates long keys** — verify file contents, not terminal output
5. **ThreadPoolExecutor max_workers** — cap at 50-100 to avoid overwhelming the local system
6. **curl timeout** — set `--max-time` higher than expected latency to avoid false failures