Most engineers who try to measure LLM API latency get it wrong. They measure wall-clock time from request send to response complete and call it a day, missing the distinction between time-to-first-token, inter-token latency, and the overhead that sits between their client and the model. If you want to measure LLM API latency correctly, you need to instrument each stage separately and understand what each number actually tells you about user experience and system capacity.
This guide walks through the correct methodology end to end. You’ll build a measurement harness that captures the metrics that matter, handles streaming responses properly, and produces numbers you can trust for capacity planning and SLO definition.
Step 1: Define the metrics you actually need
Before writing code, agree on what you’re measuring. Three metrics cover the vast majority of production decisions:
Time to first token (TTFT) — elapsed time from request initiation to the first byte of the first token arriving. This drives perceived responsiveness in chat and streaming UIs.
Inter-token latency (ITL) — the time between successive tokens in a streaming response. The median and p99 of this distribution determine whether the stream feels smooth or stutters.
End-to-end latency (E2E) — total wall-clock time from request send to the final token received (or full response parsed for non-streaming). This is what you put in SLOs and capacity models.
Do not conflate these. A system with 200 ms TTFT and 50 ms median ITL feels fast. A system with 50 ms TTFT and 500 ms median ITL feels broken, even if E2E is identical.
Step 2: Build a minimal measurement harness
Use Python’s asyncio and aiohttp for precision. The standard library’s time.perf_counter() gives nanosecond-resolution monotonic timestamps, which you need for sub-millisecond accuracy. Avoid time.time() — it’s subject to NTP adjustments and wall-clock skew.
import asyncio
import time
import statistics
from dataclasses import dataclass
from typing import AsyncIterator, Optional
import aiohttp
@dataclass
class LatencySample:
ttft_ms: Optional[float] = None
itl_ms: list[float] = None
e2e_ms: Optional[float] = None
token_count: int = 0
error: Optional[str] = None
def __post_init__(self):
if self.itl_ms is None:
self.itl_ms = []
async def measure_streaming_latency(
session: aiohttp.ClientSession,
url: str,
headers: dict,
payload: dict,
) -> LatencySample:
sample = LatencySample()
request_start = time.perf_counter()
first_token_received = False
last_token_time = None
try:
async with session.post(url, headers=headers, json=payload) as resp:
resp.raise_for_status()
async for line in resp.content:
line = line.decode("utf-8").strip()
if not line or not line.startswith("data: "):
continue
data = line[6:]
if data == "[DONE]":
break
now = time.perf_counter()
if not first_token_received:
sample.ttft_ms = (now - request_start) * 1000
first_token_received = True
last_token_time = now
else:
sample.itl_ms.append((now - last_token_time) * 1000)
last_token_time = now
sample.token_count += 1
sample.e2e_ms = (time.perf_counter() - request_start) * 1000
except Exception as e:
sample.error = str(e)
sample.e2e_ms = (time.perf_counter() - request_start) * 1000
return sampleThis captures TTFT at the first data: line, ITL between each subsequent token, and E2E at stream termination. It ignores SSE keep-alive comments and the [DONE] sentinel.
Step 3: Handle non-streaming responses correctly
Non-streaming endpoints return a single JSON payload. The latency profile is different — there’s no TTFT or ITL, only E2E. But you still need to measure from request send to full body received, not to when your JSON parser finishes.
async def measure_nonstreaming_latency(
session: aiohttp.ClientSession,
url: str,
headers: dict,
payload: dict,
) -> LatencySample:
sample = LatencySample()
request_start = time.perf_counter()
try:
async with session.post(url, headers=headers, json=payload) as resp:
resp.raise_for_status()
# Read the full body before stopping the clock
await resp.read()
sample.e2e_ms = (time.perf_counter() - request_start) * 1000
# Estimate token count from usage if present
data = await resp.json()
usage = data.get("usage", {})
sample.token_count = usage.get("completion_tokens", 0)
except Exception as e:
sample.error = str(e)
sample.e2e_ms = (time.perf_counter() - request_start) * 1000
return sampleCall resp.read() before stopping the timer. Awaiting resp.json() includes JSON parsing time, which is client overhead, not API latency.
Step 4: Run enough iterations for statistical validity
A single request tells you nothing. Run at least 100 iterations per configuration, spaced to avoid rate limits and cache effects. Warm up with 10–20 discarded requests first — cold starts on the provider side can add 200–500 ms on the first request after idle periods.
async def run_benchmark(
url: str,
headers: dict,
payload: dict,
iterations: int = 100,
warmup: int = 20,
stream: bool = True,
concurrency: int = 1,
) -> list[LatencySample]:
connector = aiohttp.TCPConnector(limit=concurrency)
timeout = aiohttp.ClientTimeout(total=120)
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
# Warmup
for _ in range(warmup):
if stream:
await measure_streaming_latency(session, url, headers, payload)
else:
await measure_nonstreaming_latency(session, url, headers, payload)
# Measured runs
semaphore = asyncio.Semaphore(concurrency)
async def measured_run():
async with semaphore:
if stream:
return await measure_streaming_latency(session, url, headers, payload)
return await measure_nonstreaming_latency(session, url, headers, payload)
tasks = [measured_run() for _ in range(iterations)]
return await asyncio.gather(*tasks)The semaphore controls concurrency. Start with concurrency=1 for baseline latency. Increase it to measure latency under load — but that’s a separate benchmark (see Step 8).
Step 5: Aggregate and report percentiles
Raw samples are noisy. Report p50, p90, p99, and max for each metric. Use the statistics module for percentiles — it’s in the standard library and correct.
def summarize(samples: list[LatencySample], label: str = "") -> dict:
ttfts = [s.ttft_ms for s in samples if s.ttft_ms is not None]
itls = [itl for s in samples for itl in s.itl_ms]
e2es = [s.e2e_ms for s in samples if s.e2e_ms is not None]
errors = [s.error for s in samples if s.error is not None]
def pct(values: list[float], p: float) -> float:
if not values:
return float("nan")
return statistics.quantiles(values, n=100)[int(p) - 1]
return {
"label": label,
"sample_count": len(samples),
"error_count": len(errors),
"ttft_ms": {
"p50": pct(ttfts, 50),
"p90": pct(ttfts, 90),
"p99": pct(ttfts, 99),
"max": max(ttfts) if ttfts else None,
},
"itl_ms": {
"p50": pct(itls, 50),
"p90": pct(itls, 90),
"p99": pct(itls, 99),
"max": max(itls) if itls else None,
},
"e2e_ms": {
"p50": pct(e2es, 50),
"p90": pct(e2es, 90),
"p99": pct(e2es, 99),
"max": max(e2es) if e2es else None,
},
"tokens_per_second": (
sum(s.token_count for s in samples) / (sum(e2es) / 1000) if e2es else 0
),
}Print or log this as JSON. It’s machine-readable and diffable across runs.
Step 6: Isolate network overhead from provider latency
Your measured latency includes network round-trip time (RTT) between your client and the API endpoint. To isolate provider-side latency, measure RTT separately and subtract it — or run the benchmark from the same region as the provider.
Measure baseline RTT with a lightweight endpoint (like /models or a health check) using the same connection pool:
async def measure_rtt(session: aiohttp.ClientSession, url: str, iterations: int = 50) -> float:
rtts = []
for _ in range(iterations):
start = time.perf_counter()
async with session.get(url) as resp:
await resp.read()
rtts.append((time.perf_counter() - start) * 1000)
return statistics.median(rtts)Subtract the median RTT from your TTFT and E2E numbers to approximate provider-internal latency. This matters when comparing providers across regions or evaluating edge routing.
Step 7: Account for tokenization differences
Token counts vary by tokenizer. A 1,000-token completion in GPT-4o is not the same character count as 1,000 tokens in Llama 3. If you’re comparing throughput across models, normalize by characters or words, not tokens. Or report both.
def estimate_chars_per_token(model: str) -> float:
# Rough heuristics; replace with actual tokenizer for precision
heuristics = {
"gpt-4o": 3.8,
"gpt-4o-mini": 3.8,
"llama-3-70b": 3.2,
"llama-3-8b": 3.2,
"claude-3-5-sonnet": 3.5,
}
return heuristics.get(model.lower(), 3.5)Multiply token_count by this factor for approximate character throughput. For rigorous work, pull the tokenizer (via tiktoken or transformers) and count actual tokens in the response body.
Step 8: Measure latency under concurrency (optional but recommended)
Baseline latency at concurrency=1 is the best-case number. Production systems run at concurrency > 1. Extend the harness to sweep concurrency levels and plot latency vs. throughput.
async def concurrency_sweep(
url: str,
headers: dict,
payload: dict,
concurrency_levels: list[int] = [1, 2, 4, 8, 16, 32],
iterations_per_level: int = 50,
) -> list[dict]:
results = []
for c in concurrency_levels:
samples = await run_benchmark(
url, headers, payload,
iterations=iterations_per_level,
concurrency=c,
)
summary = summarize(samples, f"concurrency={c}")
summary["concurrency"] = c
results.append(summary)
print(f"Concurrency {c}: E2E p50={summary['e2e_ms']['p50']:.1f}ms, "
f"throughput={summary['tokens_per_second']:.1f} tok/s")
return resultsWatch for the knee of the curve — where p99 latency starts climbing exponentially. That’s your effective capacity limit.
Step 9: Verify your measurements are sane
Before trusting any number, run these sanity checks:
-
TTFT < E2E always — If TTFT exceeds E2E, your stream parsing is broken (likely counting
[DONE]or a keep-alive as a token). -
ITL median is stable — Inter-token latency should be roughly constant for a given model and provider. Wild variance (e.g., p50=40ms, p99=2000ms) indicates queueing or provider-side batching artifacts.
-
Tokens per second matches provider specs — If a provider advertises 100 tok/s and you measure 10 tok/s at concurrency=1, something is wrong (wrong endpoint, streaming disabled, or you’re measuring through a slow proxy).
-
Error rate near zero — Any non-zero error rate in a clean benchmark invalidates the latency numbers. Retries, timeouts, and 429s all distort percentiles.
-
Warmup actually warms — Compare the first 5 measured samples after warmup to the last 5. They should be statistically indistinguishable. If not, increase warmup iterations.
Step 10: Automate and version your benchmarks
Check the harness into version control. Parameterize the endpoint, model, prompt, and expected token count. Run it in CI on a schedule (nightly) to detect regressions — provider performance drifts over time.
# benchmark_config.json
{
"endpoint": "https://api.example.com/v1/chat/completions",
"model": "gpt-4o-mini",
"prompt": "Write a 200-word explanation of TCP congestion control.",
"max_tokens": 300,
"temperature": 0.7,
"stream": true,
"iterations": 100,
"warmup": 20,
"concurrency": 1
}import json
def load_config(path: str) -> dict:
with open(path) as f:
return json.load(f)
async def main():
config = load_config("benchmark_config.json")
headers = {
"Authorization": f"Bearer {os.environ['API_KEY']}",
"Content-Type": "application/json",
}
payload = {
"model": config["model"],
"messages": [{"role": "user", "content": config["prompt"]}],
"max_tokens": config["max_tokens"],
"temperature": config["temperature"],
"stream": config["stream"],
}
samples = await run_benchmark(
config["endpoint"],
headers,
payload,
iterations=config["iterations"],
warmup=config["warmup"],
stream=config["stream"],
concurrency=config["concurrency"],
)
summary = summarize(samples, config["model"])
print(json.dumps(summary, indent=2))Store the JSON output alongside the commit. A simple jq query on-call runbook: “If p99 E2E increases > 20% vs. last week’s baseline, page the platform team.”
Common pitfalls to avoid
Measuring from the wrong clock — time.time() jumps with NTP. Always use time.perf_counter() or time.monotonic().
Including client-side parsing in E2E — Stop the clock after resp.read(), not after resp.json() or your Pydantic model validation.
Ignoring SSL handshake time — The first request on a new connection pays TLS handshake cost. Reuse connections (the ClientSession does this) and warm up adequately.
Treating 429/5xx as latency samples — Errors are not latency. Track them separately. If you retry, measure each attempt independently.
Comparing streaming vs. non-streaming E2E directly — Streaming E2E includes token generation time. Non-streaming E2E includes server-side buffering. They measure different things.
Running from a laptop on WiFi — Run benchmarks from a stable network (CI runner, cloud VM in the same region as the API). Document the measurement environment.
What good numbers look like
As a rough reference for modern APIs in the same region:
| Metric | Good | Acceptable | Investigate |
|---|---|---|---|
| TTFT p50 | < 300 ms | 300–600 ms | > 600 ms |
| ITL p50 | < 50 ms | 50–100 ms | > 100 ms |
| ITL p99 | < 200 ms | 200–500 ms | > 500 ms |
| E2E p99 (1k tokens) | < 8 s | 8–15 s | > 15 s |
These are rules of thumb, not SLAs. Your product requirements define the real thresholds.
Closing note
Measuring LLM API latency correctly is not hard, but it is easy to do wrong. The difference between a useful benchmark and a misleading one comes down to: separating TTFT from ITL from E2E, using a monotonic clock, reading the full response body before stopping the timer, running enough iterations for stable percentiles, and isolating network overhead. Get those right and you have numbers you can build capacity plans, SLOs, and routing logic on top of.