Measuring inter-token latency in server-sent event streams

Inter-token latency SSE streams is the only metric that tells you if a model response will feel typed-out by a human or dumped in chunks. Time-to-first-token hides the tail; measuring the gap between successive tokens exposes proxy buffering, provider throttling, and client-side stalls. This guide gives you a working client, parsing rules, and a verification loop you can run today.

Step 1: Stand up a raw streaming client

Do not use a high-level SDK that buffers the stream into a single string. You need byte-level visibility. requests with stream=True is enough for Python; in Node, use fetch and read the ReadableStream.

import requests, os, json

URL = "https://your-endpoint/v1/chat/completions"
HEADERS = {
    "Authorization": f"Bearer {os.environ['API_KEY']}",
    "Content-Type": "application/json",
}
BODY = {
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "Explain SSE in one sentence."}],
    "stream": True,
}

with requests.post(URL, headers=HEADERS, json=BODY, stream=True) as r:
    for line in r.iter_lines(decode_unicode=True):
        if line:
            print(line)

Run this once to confirm you see data: {...} lines and a final data: [DONE]. If you see nothing until the response ends, a proxy is buffering—fix that before measuring.

Step 2: Parse SSE frames without guessing

Server-sent events are line-oriented. A frame starts with data: and ends with a blank line. Providers also send comment lines (starting with :) as keep-alives. Ignore anything that is not a data: line.

def parse_sse(line: str):
    if not line.startswith("data:"):
        return None
    payload = line[len("data:"):].strip()
    if payload == "[DONE]":
        return "DONE"
    try:
        return json.loads(payload)
    except json.JSONDecodeError:
        return None

The JSON object contains choices[0].delta.content. That string may be empty on the first frame (role announcement) or contain multiple characters. Treat each non-empty content arrival as one token-event; if you need true sub-token precision, run the string through the model tokenizer client-side.

Step 3: Record monotonic timestamps per delta

Wall-clock time drifts and jumps on laptop suspend. Use time.monotonic(). Capture the gap between successive content deliveries, not just the absolute time.

import time

class TokenLatencyRecorder:
    def __init__(self):
        self.gaps = []
        self.last_ts = None

    def on_token(self, content: str):
        if not content:
            return
        now = time.monotonic()
        if self.last_ts is not None:
            self.gaps.append(now - self.last_ts)
        self.last_ts = now

Wire it into the loop:

recorder = TokenLatencyRecorder()
with requests.post(URL, headers=HEADERS, json=BODY, stream=True) as r:
    for line in r.iter_lines(decode_unicode=True):
        obj = parse_sse(line)
        if isinstance(obj, dict):
            content = obj["choices"][0]["delta"].get("content", "")
            recorder.on_token(content)

You now have a list of inter-arrival times in seconds. That list is your raw signal.

Step 4: Compute the latency distribution

A single average is useless. Compute percentiles. If you already pull in NumPy, use it; otherwise statistics is fine for moderate samples.

import statistics

def report(recorder):
    gaps = recorder.gaps
    if not gaps:
        return {"count": 0}
    sorted_g = sorted(gaps)
    return {
        "count": len(sorted_g),
        "p50_ms": statistics.median(sorted_g) * 1000,
        "p90_ms": sorted_g[int(len(sorted_g) * 0.9)] * 1000,
        "p99_ms": sorted_g[int(len(sorted_g) * 0.99)] * 1000,
    }

In a healthy local mock you will see p50 under 10 ms. Across a real provider, p50 of 20–40 ms is typical; p99 above 200 ms usually means a downstream throttle or a proxy flushing in batches.

Step 5: Control for transport artifacts

Most “latency spikes” are not the model. They are:

• Reverse proxies with proxy_buffering on (nginx default). Set proxy_buffering off or send X-Accel-Buffering: no.
• GZip on a stream. Disable compression for the streaming route.
• TCP Nagle combined with delayed ACKs. Not much you can do client-side, but measuring from a nearby region reduces it.

If you route through n4n.ai, the OpenAI-compatible endpoint forwards provider cache-control hints and handles fallback without altering the SSE frame shape, so the recorder above works unchanged across 240+ models. You still must own your own edge proxy config.

Step 6: Verify against a controlled source

Never trust a measurement you cannot reproduce. Spin up a mock SSE server that emits tokens at a known fixed gap, then confirm your client reports that gap.

from http.server import BaseHTTPRequestHandler, HTTPServer
import time, json

class MockSSE(BaseHTTPRequestHandler):
    def do_POST(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.end_headers()
        for i in range(20):
            data = json.dumps({"choices":[{"delta":{"content":f"tok{i} "}}]})
            self.wfile.write(f"data: {data}\n\n".encode())
            self.wfile.flush()
            time.sleep(0.05)  # enforced 50 ms gap
        self.wfile.write(b"data: [DONE]\n\n")

HTTPServer(("127.0.0.1", 8080), MockSSE).serve_forever()

Point URL at http://127.0.0.1:8080/v1/chat/completions, run the client, and check report(recorder)["p50_ms"] is within 5 ms of 50. If it reads 0 or 500, your client is batching lines—go back to Step 2.

Step 7: Sample in production without overhead

You do not need to record every session. Wrap the recorder in a sampling guard and emit to your metrics pipeline.

import random

def stream_with_metrics(body, sample_rate=0.01):
    if random.random() < sample_rate:
        rec = TokenLatencyRecorder()
    else:
        rec = None
    with requests.post(URL, headers=HEADERS, json=body, stream=True) as r:
        for line in r.iter_lines(decode_unicode=True):
            obj = parse_sse(line)
            if isinstance(obj, dict) and rec:
                rec.on_token(obj["choices"][0]["delta"].get("content", ""))
    if rec:
        metrics.emit("inter_token_latency", report(rec))

Keep the sampling decision per-request, not per-token, to avoid biased distributions. Store p50/p90/p99 as histograms, not gauges, so you can spot regressions when a provider rotates a backend.

Verify success

You have a working measurement when:

1. The mock server test yields p50 within 5 ms of the injected gap.
2. A real provider run shows non-zero gaps with p99 below your UI’s animation budget (typically 100 ms).
3. Disabling your nginx buffer changes p99 by more than 2×—confirming you were measuring the proxy, not the model.

If those hold, you can alert on p99 drift and stop guessing about “why the chat feels laggy.”