#!/usr/bin/env python3 # cmpbench.py 1.0: the cmp.wiki Benchmark protocol 1.0 for OpenAI-compatible inference servers. # Protocol and download: https://cmp.wiki/wiki/Benchmark_protocol # # MIT License # # Copyright (c) 2026 cmp.wiki contributors # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """Measure an OpenAI-compatible LLM server the cmp.wiki way and print the numbers ready for a Recipe or Confirmation. python3 cmpbench.py --base-url http://localhost:8000/v1 --model MODEL [--api-key KEY] Stages, in this order, after a warmup: 1. decode with one user: three fixed prompts (code, structured JSON, prose), temperature 0, 1,024-token answers, streamed; tok/s = completion tokens / (last token time - first token time); time to first token (TTFT) apart 2. prefill: one prompt of about 8,000 tokens with a unique start (no prefix-cache hit), 1 output token; tok/s = prompt tokens / TTFT 3. decode with many users: 8 concurrent requests of the prose prompt (--users N), 512-token answers; total tok/s = all completion tokens / (last token of any request - first token of any request) Token counts come from the server's usage fields, else from llama.cpp's timings, else from an approximation of 4 characters per token (the output says which). Python 3.9 or newer, standard library only. The script talks only to --base-url (proxy variables are ignored) and reads nvidia-smi and lspci on this machine. It never reads GPU serial numbers or UUIDs. """ import argparse import http.client import json import os import re import shutil import socket import ssl import subprocess import sys import threading import time from datetime import datetime, timezone from urllib.parse import urlsplit VERSION = "1.0" PROTOCOL = "1.0" USER_AGENT = "cmpbench/" + VERSION + " (+https://cmp.wiki/wiki/Benchmark_protocol)" CHARS_PER_TOKEN = 4.0 # fallback approximation when the server reports no token counts FULL = {"decode_tokens": 1024, "prefill_target": 8000, "calib_target": 1000, "multi_tokens": 512, "warm_tokens": 32} QUICK = {"decode_tokens": 128, "prefill_target": 2000, "calib_target": 500, "multi_tokens": 64, "warm_tokens": 16} ORDER = ("code", "structured", "prose") PROMPTS = { "code": ( "Write a complete, self-contained Python 3 module named ratelimit.py. It must implement a thread-safe token " "bucket rate limiter class called TokenBucket with the methods acquire(tokens=1, timeout=None), " "try_acquire(tokens=1) and available(); a decorator called rate_limited(rate, burst) that shares one bucket " "across all calls of the decorated function; and an asyncio version called AsyncTokenBucket with the same " "methods as coroutines. Use type hints and docstrings, and raise ValueError for invalid arguments. Then write " "a pytest file named test_ratelimit.py with at least twelve tests that cover refill timing, bursts, timeouts, " "several threads acquiring at once, the decorator and the async version. Use a fake clock so the tests run " "instantly. Output only the two files, each in its own fenced code block, with no text before, between or " "after them." ), "structured": ( "Generate a JSON inventory of 30 fictional rack servers. Output one JSON object with a single key, " "\"servers\", holding an array of 30 objects. Each object has exactly these keys: \"id\" (a string such as " "\"srv-001\"), \"rack\" (such as \"R3\"), \"position_u\" (an integer from 1 to 42), \"model\" (an " "invented name), \"cpu_cores\", \"memory_gb\", \"gpus\" (an array of objects with \"slot\", \"model\" and " "\"memory_gb\"), \"power_supply_w\", \"os\", \"status\" (one of \"active\", \"standby\" or " "\"maintenance\") and \"notes\" (one sentence). Use realistic, varied values. Output only valid JSON with " "two-space indentation and no text before or after it." ), "prose": ( "Write a long, detailed essay of about 1,500 words for a general reader on the history of lighthouses and the " "people who kept them. Cover the fire towers of the ancient world and the Pharos of Alexandria, the move from " "wood and coal fires to oil lamps and mirrors, the invention of the Fresnel lens, the daily routine and " "isolation of keepers and their families, storms and rescues, the arrival of electricity and automation, and " "what lighthouses mean to coastal towns today. Write in clear paragraphs with no headings, lists or bullet " "points, in a calm and descriptive style, with concrete details and dates where they help, and end with a short " "reflective paragraph." ), } # Sentences for the long prefill prompt. The text is built from them by a fixed pseudo-random sequence, so every run # and every machine sends the same text (cut to the length the tokenizer needs), after a short random reference line # that keeps prefix caches from answering for the engine. CORPUS = ( "Pump {a} on the north line was inspected and its seals were replaced after {b} hours of service.", "The crew logged an inlet pressure of {a} kilopascals and an outlet pressure of {b} kilopascals.", "A spare bearing was taken from shelf {a} and the stock count was lowered to {b} units.", "Valve {a} stuck halfway open during the morning test and was freed with penetrating oil.", "The cooling loop held a steady {a} degrees for the whole shift, well under the alarm limit of {b}.", "Inspectors noted light corrosion on bracket {a} and scheduled a repaint within {b} days.", "Filter housing {a} was cleaned, and the differential reading dropped from {b} to a normal value.", "The night shift reported a faint knocking sound near motor {a} that stopped after a restart.", "Cable tray {a} was relabeled so that the drawings match the {b} circuits installed last spring.", "A vibration survey found motor {a} running at {b} micrometers, inside the acceptable band.", "The backup generator ran for {a} minutes under load and its fuel tank was refilled to {b} liters.", "Gasket kit {a} arrived damaged, so the supplier was asked to ship a replacement within {b} days.", "The lubrication schedule for conveyor {a} moved from every {b} hours to every second week.", "Two technicians recalibrated flow meter {a} against the reference standard and found a small offset.", "Drain line {a} was flushed after a slow leak left a stain about {b} centimeters wide on the floor.", "Thermal images of panel {a} showed one breaker running warmer than its neighbours by {b} degrees.", "The compressor on skid {a} reached operating pressure in {b} seconds, faster than last month.", "Safety guard {a} was refitted with new bolts after the old ones failed the torque check.", "A water sample from tank {a} was sent to the lab, and the result is expected in {b} days.", "The control room updated alarm setpoint {a} after operators agreed the old value of {b} was too tight.", "Belt tension on fan {a} was adjusted, and the motor current fell by about {b} percent.", "Heat exchanger {a} was opened for cleaning and about {b} grams of scale were removed.", "Work order {a} was closed once the replacement sensor passed {b} consecutive readings.", "The forklift used in bay {a} had its hydraulic hose replaced and passed the load test.", "Emergency lighting in corridor {a} was tested for {b} minutes on battery and stayed bright.", "Coupling {a} showed a misalignment of {b} hundredths of a millimeter and was shimmed back into line.", "Insulation on steam line {a} was patched where the jacket had come loose over {b} centimeters.", "The weekly walkdown of area {a} found no new leaks, loose fittings or missing labels.", "Gearbox {a} oil was changed and a sample was kept, since the old oil had run {b} hours.", "The spare parts list for line {a} now names {b} critical items that must always be in stock.", ) class BenchError(Exception): """An error with a message meant for the person running the script.""" class HTTPFail(BenchError): def __init__(self, status, body, url): self.status, self.body, self.url = status, body, url super().__init__(f"HTTP {status} from {url}: {body[:300]}") # ---------------------------------------------------------------------------------------------------------------- HTTP class Endpoint: def __init__(self, base_url, api_key=None, timeout=900.0, insecure=False): u = urlsplit(base_url.strip()) if u.scheme not in ("http", "https") or not u.hostname: raise BenchError(f"--base-url must look like http://host:port/v1, got {base_url!r}") self.https = u.scheme == "https" self.host = u.hostname self.port = u.port or (443 if self.https else 80) self.path = u.path.rstrip("/") self.root = self.path[:-3] if self.path.endswith("/v1") else self.path self.display = f"{u.scheme}://{u.hostname}:{self.port}{self.path}" self.api_key = api_key self.timeout = timeout self.insecure = insecure def is_local(self): return self.host in ("localhost", "127.0.0.1", "::1") or self.host.startswith("127.") def _conn(self, timeout): if self.https: ctx = ssl.create_default_context() if self.insecure: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE return http.client.HTTPSConnection(self.host, self.port, timeout=timeout, context=ctx) return http.client.HTTPConnection(self.host, self.port, timeout=timeout) def _headers(self, body=False): h = {"User-Agent": USER_AGENT, "Accept": "application/json, text/event-stream"} if body: h["Content-Type"] = "application/json" if self.api_key: h["Authorization"] = "Bearer " + self.api_key return h def _open(self, method, path, payload, timeout): conn = self._conn(timeout) try: conn.request(method, path, body=payload, headers=self._headers(payload is not None)) return conn, conn.getresponse() except ConnectionRefusedError: conn.close() raise BenchError(f"connection refused at {self.display}: is the server running, and are the host and port " f"in --base-url right?") except socket.timeout: conn.close() raise BenchError(f"no answer from {self.display} within {timeout:.0f} s (--timeout)") except ssl.SSLError as e: conn.close() raise BenchError(f"TLS error talking to {self.display}: {e} (--insecure skips certificate checks)") except (OSError, http.client.HTTPException) as e: conn.close() raise BenchError(f"cannot reach {self.display}: {e}") def get(self, path, timeout=10.0): """-> (status, text); (None, error text) when the server cannot be reached.""" try: conn, resp = self._open("GET", path, None, timeout) except BenchError as e: return None, str(e) try: return resp.status, resp.read(4 << 20).decode("utf-8", "replace") except (OSError, http.client.HTTPException) as e: return None, str(e) finally: conn.close() def get_json(self, path, timeout=10.0): st, text = self.get(path, timeout) if st != 200: return st, None try: return st, json.loads(text) except ValueError: return st, None def chat(self, body): """POST chat/completions; returns a Result with timings. Streams when the server streams.""" payload = json.dumps(body).encode("utf-8") url = self.path + "/chat/completions" t0 = time.perf_counter() conn, resp = self._open("POST", url, payload, self.timeout) try: if resp.status != 200: raise HTTPFail(resp.status, resp.read(8192).decode("utf-8", "replace").strip(), self.display + "/chat/completions") r = Result(t0) first = b"" while True: # first non-empty line decides between SSE and a plain JSON body line = resp.readline() if not line: break if line.strip(): first = line break s = first.strip() if s.startswith(b"data:") or s.startswith(b"event:") or s.startswith(b":"): r.streamed = True self._read_sse(resp, first, r) else: rest = resp.read() r.read_json_body(first + rest, time.perf_counter()) r.t_end = time.perf_counter() return r except socket.timeout: raise BenchError(f"the server stopped sending within {self.timeout:.0f} s (--timeout)") except (OSError, http.client.HTTPException) as e: raise BenchError(f"connection to {self.display} broke during a request: {e}") finally: conn.close() @staticmethod def _read_sse(resp, first, r): line = first while line: s = line.strip() if s.startswith(b"data:"): data = s[5:].strip() if data == b"[DONE]": break try: obj = json.loads(data) except ValueError: r.bad_chunks += 1 obj = None if isinstance(obj, dict): if obj.get("error"): err = obj["error"] msg = err.get("message", err) if isinstance(err, dict) else err raise BenchError(f"the server reported an error during streaming: {msg}") r.chunk(obj, time.perf_counter()) line = resp.readline() class Result: """One request: arrival times of the first and last generated text, usage and llama.cpp timings.""" def __init__(self, t0): self.t0 = t0 self.t_first = None # first chunk carrying text self.t_last = None # last chunk carrying text self.t_event = None # first chunk carrying text or a finish reason (prefill with max_tokens 1) self.t_end = None self.streamed = False self.text_chunks = 0 self.bad_chunks = 0 self.parts = [] self.reasoning = False self.usage = None self.timings = None self.finish = None def chunk(self, obj, now): got = False for ch in obj.get("choices") or []: if not isinstance(ch, dict): continue delta = ch.get("delta") or {} piece = "" if isinstance(delta, dict): c = delta.get("content") rs = delta.get("reasoning_content") or delta.get("reasoning") if isinstance(c, str): piece += c if isinstance(rs, str) and rs: piece += rs self.reasoning = True if not delta and isinstance(ch.get("text"), str): piece += ch["text"] if piece: got = True self.parts.append(piece) if ch.get("finish_reason"): self.finish = ch["finish_reason"] if self.t_event is None: self.t_event = now if got: self.text_chunks += 1 if self.t_first is None: self.t_first = now if self.t_event is None: self.t_event = now self.t_last = now if isinstance(obj.get("usage"), dict) and obj["usage"]: self.usage = obj["usage"] if isinstance(obj.get("timings"), dict): self.timings = obj["timings"] def read_json_body(self, raw, now): try: obj = json.loads(raw.decode("utf-8", "replace")) except ValueError: raise BenchError("the server answered with something that is neither an event stream nor JSON: " + raw[:200].decode("utf-8", "replace")) if isinstance(obj, dict) and obj.get("error"): err = obj["error"] raise BenchError(f"the server reported an error: {err.get('message', err) if isinstance(err, dict) else err}") for ch in obj.get("choices") or []: msg = ch.get("message") or {} for key in ("content", "reasoning_content"): if isinstance(msg.get(key), str) and msg[key]: self.parts.append(msg[key]) self.reasoning = self.reasoning or key != "content" if isinstance(ch.get("text"), str): self.parts.append(ch["text"]) self.finish = ch.get("finish_reason") or self.finish self.usage = obj.get("usage") if isinstance(obj.get("usage"), dict) else None self.timings = obj.get("timings") if isinstance(obj.get("timings"), dict) else None self.t_event = now @property def text(self): return "".join(self.parts) def completion_tokens(self): """-> (count, source)""" u = self.usage or {} if isinstance(u.get("completion_tokens"), (int, float)) and u["completion_tokens"] > 0: return int(u["completion_tokens"]), "usage" t = self.timings or {} if isinstance(t.get("predicted_n"), (int, float)) and t["predicted_n"] > 0: return int(t["predicted_n"]), "timings" return max(1, round(len(self.text) / CHARS_PER_TOKEN)), "approx" def prompt_tokens(self, prompt_chars): u = self.usage or {} if isinstance(u.get("prompt_tokens"), (int, float)) and u["prompt_tokens"] > 0: return int(u["prompt_tokens"]), "usage" t = self.timings or {} if isinstance(t.get("prompt_n"), (int, float)) and t["prompt_n"] > 0: return int(t["prompt_n"]), "timings" return max(1, round(prompt_chars / CHARS_PER_TOKEN)), "approx" def ttft(self): t = self.t_first if self.t_first is not None else self.t_event return None if t is None else t - self.t0 def decode_tps(self, tokens): """tokens / (last - first); None without streaming or with a single chunk.""" if not self.streamed or self.t_first is None or self.t_last is None or self.t_last <= self.t_first: return None return tokens / (self.t_last - self.t_first) # ------------------------------------------------------------------------------------------------------- requests def chat_body(model, prompt, max_tokens, compat): b = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": 0, "stream": True} if compat.get("stream_options"): b["stream_options"] = {"include_usage": True} if compat.get("thinking_hint"): b["chat_template_kwargs"] = {"enable_thinking": False, "thinking": False} return b def explain_http(e, model): b = e.body.lower() if e.status in (401, 403): return f"the server refused the request (HTTP {e.status}): pass the right key with --api-key. {e.body[:200]}" if e.status == 404: if "model" in b: return f"model {model!r} was not found on the server (HTTP 404): {e.body[:300]}" return (f"HTTP 404 at {e.url}: the path is wrong. --base-url normally ends in /v1, " f"for example http://localhost:8000/v1") if e.status in (400, 413, 422) and any(w in b for w in ("context", "too long", "maximum", "exceed", "length")): return f"the prompt does not fit the server's context window (HTTP {e.status}): {e.body[:300]}" if e.status in (502, 503, 504): return f"the server is up but has no model ready (HTTP {e.status}): {e.body[:300]}" return f"HTTP {e.status} from the server: {e.body[:300]}" def detect_compat(ep, model, thinking_off, warm_tokens): """Send the warmup request, dropping optional fields one by one if the server refuses them (HTTP 400/422).""" tries = [] for th in ((True, False) if thinking_off else (False,)): for so in (True, False): tries.append({"thinking_hint": th, "stream_options": so}) first_err = None for compat in tries: try: r = ep.chat(chat_body(model, PROMPTS["code"], warm_tokens, compat)) return compat, r except HTTPFail as e: if e.status not in (400, 422): raise BenchError(explain_http(e, model)) first_err = first_err or e raise BenchError(explain_http(first_err, model)) def long_text(target_chars, nonce): """Deterministic maintenance-log text of about target_chars characters after a run-unique reference line. The same sentence sequence on every run and machine; a longer target only extends it.""" seed = 20261104 def nxt(): nonlocal seed seed = (seed * 1103515245 + 12345) % 2147483648 return seed >> 8 out = [f"Reference {nonce}. The notes below come from a plant maintenance log. Read all of them."] size = len(out[0]) n = 0 while size < target_chars: if n % 6 == 0: out.append(f"\n\nSection {n // 6 + 1}.") s = CORPUS[nxt() % len(CORPUS)] a = 10 + nxt() % 90 b = 100 + nxt() % 900 out.append(" " + s.format(a=a, b=b)) size += len(out[-1]) + (len(out[-2]) if n % 6 == 0 else 0) n += 1 out.append("\n\nQuestion: in one word, what is the main subject of the notes above?") return "".join(out) def new_nonce(): return os.urandom(6).hex() # ---------------------------------------------------------------------------------------------------------- GPUs GPU_FIELDS = ("index", "name", "pci.bus_id", "pci.device_id", "memory.total", "pcie.link.gen.current", "pcie.link.width.current", "pcie.link.gen.max", "pcie.link.width.max", "power.limit", "driver_version") GPU_CORE_FIELDS = ("index", "name", "pci.bus_id", "memory.total", "pcie.link.gen.current", "pcie.link.width.current", "power.limit", "driver_version") SAMPLE_FIELDS = ("index", "pcie.link.gen.current", "pcie.link.width.current", "power.draw") CMP_IDS = {"20c2": "8 GB", "2082": "10 GB"} def _run(cmd, timeout=15): try: p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, universal_newlines=True, env=dict(os.environ, LC_ALL="C")) except (OSError, subprocess.SubprocessError): return None return p.stdout if p.returncode == 0 else None def _val(s): s = s.strip() if not s or s.startswith("[") or s.lower() in ("n/a", "not supported"): return None try: return int(s) except ValueError: try: return float(s) except ValueError: return s def parse_smi_csv(text, fields): rows = [] for line in (text or "").splitlines(): if not line.strip(): continue vals = [v.strip() for v in line.split(",")] if len(vals) != len(fields): continue rows.append({f: (_val(v) if f not in ("name", "pci.bus_id", "driver_version", "pci.device_id") else (v if v and not v.startswith("[") else None)) for f, v in zip(fields, vals)}) return rows def query_smi(fields): out = _run(["nvidia-smi", "--query-gpu=" + ",".join(fields), "--format=csv,noheader,nounits"]) return parse_smi_csv(out, fields) if out is not None else None def norm_bus(bus): """'00000000:41:00.0' or '0000:41:00.0' -> '0000:41:00.0'""" m = re.match(r"^(?:([0-9a-fA-F]+):)?([0-9a-fA-F]{2}):([0-9a-fA-F]{2})\.([0-7])$", (bus or "").strip()) if not m: return (bus or "").strip().lower() dom = int(m.group(1) or "0", 16) return f"{dom:04x}:{m.group(2).lower()}:{m.group(3).lower()}.{m.group(4)}" def parse_lspci(text): """lspci -D -nn -> {bus: (vendor, device, description)} for NVIDIA functions.""" out = {} for line in (text or "").splitlines(): m = re.match(r"^(\S+)\s+(.*)\[10de:([0-9a-fA-F]{4})\]", line) if m: out[norm_bus(m.group(1))] = ("10de", m.group(3).lower(), m.group(2).strip()) return out def collect_gpus(select=None): info = {"available": False, "cards": []} if not shutil.which("nvidia-smi"): info["reason"] = "nvidia-smi is not on PATH" return info rows = query_smi(GPU_FIELDS) if rows is None: rows = query_smi(GPU_CORE_FIELDS) if not rows: info["reason"] = "nvidia-smi gave no GPU list" return info pci = parse_lspci(_run(["lspci", "-D", "-nn"])) if shutil.which("lspci") else {} info["available"] = True info["lspci"] = bool(pci) for r in rows: if select is not None and r.get("index") not in select: continue bus = norm_bus(r.get("pci.bus_id")) dev, src = None, None if bus in pci: dev, src = pci[bus][1], "lspci" elif r.get("pci.device_id"): m = re.match(r"^0x([0-9a-fA-F]{4})([0-9a-fA-F]{4})$", str(r["pci.device_id"])) if m and m.group(2).lower() == "10de": dev, src = m.group(1).lower(), "nvidia-smi" info["cards"].append({ "index": r.get("index"), "name": r.get("name"), "bus_id": bus, "device_id": ("10de:" + dev) if dev else None, "device_id_from": src, "variant": CMP_IDS.get(dev), "memory_total_mib": r.get("memory.total"), "link_gen_idle": r.get("pcie.link.gen.current"), "link_width_idle": r.get("pcie.link.width.current"), "link_gen_max": r.get("pcie.link.gen.max"), "link_width_max": r.get("pcie.link.width.max"), "power_limit_w": r.get("power.limit"), "driver": r.get("driver_version"), }) return info class Sampler(threading.Thread): """Reads link generation, width and power draw every 2 s while the benchmark runs (links drop when idle).""" def __init__(self, interval=2.0): super().__init__(daemon=True) self.interval = interval self.stop_ev = threading.Event() self.seen = {} self.samples = 0 def run(self): while not self.stop_ev.is_set(): rows = query_smi(SAMPLE_FIELDS) if rows: self.samples += 1 for r in rows: d = self.seen.setdefault(r.get("index"), {"gen": None, "width": None, "power": None}) for key, f in (("gen", "pcie.link.gen.current"), ("width", "pcie.link.width.current"), ("power", "power.draw")): v = r.get(f) if isinstance(v, (int, float)) and (d[key] is None or v > d[key]): d[key] = v self.stop_ev.wait(self.interval) def stop(self): self.stop_ev.set() self.join(timeout=20) # ---------------------------------------------------------------------------------------------------- server info SPEC_COUNTERS = ("vllm:spec_decode_num_drafts_total", "vllm:spec_decode_num_draft_tokens_total", "vllm:spec_decode_num_accepted_tokens_total") METRIC_RE = re.compile(r"^([a-zA-Z_:][a-zA-Z0-9_:]*)(\{[^}]*\})?\s+([-+]?(?:[0-9.]+(?:[eE][-+]?[0-9]+)?|inf|nan))", re.I) def read_metrics(ep): st, text = ep.get(ep.root + "/metrics", timeout=5.0) if st != 200 or not text: return None vals = {} for line in text.splitlines(): if line.startswith("#"): continue m = METRIC_RE.match(line) if m: try: vals[m.group(1)] = vals.get(m.group(1), 0.0) + float(m.group(3)) except ValueError: pass return vals or None def spec_summary(before, after): if not after: return None if before and all(k in after for k in SPEC_COUNTERS): d = {k: after[k] - before.get(k, 0.0) for k in SPEC_COUNTERS} drafts, draft_tok, acc = (d[k] for k in SPEC_COUNTERS) if drafts > 0 and draft_tok > 0: return {"source": "vLLM /metrics counters", "drafts": int(drafts), "draft_tokens": int(draft_tok), "accepted_tokens": int(acc), "acceptance_rate": round(acc / draft_tok, 3), "mean_acceptance_length": round(1 + acc / drafts, 2)} return {"source": "vLLM /metrics counters", "drafts": 0, "note": "no drafts during the stage (spec decode off)"} gauges = {k: v for k, v in after.items() if re.search(r"spec.*accept|accept.*(length|rate)", k)} if gauges: return {"source": "/metrics gauges after the stage", "values": gauges} return None ENGINE_FAMILY = {"vllm": "vLLM", "sglang": "SGLang", "llamacpp": "llama.cpp", "llama.cpp": "llama.cpp"} def server_info(ep, model, check_model): info = {"models_listed": None, "owned_by": None, "context_tokens": None, "engine_family": None, "engine_version": None} st, text = ep.get(ep.path + "/models") if st is None: # first contact: the connection problem, in words raise BenchError(text) try: obj = json.loads(text) if st == 200 else None except ValueError: obj = None if st in (401, 403): raise BenchError(f"the server refused the request (HTTP {st}): pass the right key with --api-key") if st == 200 and isinstance(obj, dict) and isinstance(obj.get("data"), list): ids = [m.get("id") for m in obj["data"] if isinstance(m, dict)] info["models_listed"] = ids entry = next((m for m in obj["data"] if isinstance(m, dict) and m.get("id") == model), None) if entry is None and ids and check_model: raise BenchError(f"model {model!r} is not in the server's model list. The server lists: " + ", ".join(str(i) for i in ids[:25]) + (" (and more)" if len(ids) > 25 else "") + ". Pass one of these to --model, or --no-model-check to send it anyway.") if entry: info["owned_by"] = entry.get("owned_by") for k in ("max_model_len", "context_length", "max_context_length"): if isinstance(entry.get(k), int) and entry[k] > 0: info["context_tokens"] = entry[k] info["context_from"] = "/v1/models " + k break else: info["models_note"] = f"GET {ep.path}/models answered HTTP {st}; model not checked" fam = ENGINE_FAMILY.get(str(info["owned_by"] or "").lower()) info["engine_family"] = fam st, obj = ep.get_json(ep.root + "/version", timeout=5.0) if st == 200 and isinstance(obj, dict) and isinstance(obj.get("version"), str): info["engine_version"] = obj["version"] if fam == "llama.cpp": st, obj = ep.get_json(ep.root + "/props", timeout=5.0) if st == 200 and isinstance(obj, dict) and isinstance(obj.get("build_info"), str): info["engine_version"] = obj["build_info"] return info # ------------------------------------------------------------------------------------------------------- stages def say(msg): print(msg, file=sys.stderr, flush=True) def stage_single(ep, model, compat, prof, warnings): out = {} for name in ORDER: say(f" 1 user, {name} prompt, {prof['decode_tokens']} tokens ...") try: r = ep.chat(chat_body(model, PROMPTS[name], prof["decode_tokens"], compat)) except HTTPFail as e: out[name] = {"error": explain_http(e, model)} continue except BenchError as e: out[name] = {"error": str(e)} continue n, src = r.completion_tokens() d = {"completion_tokens": n, "token_source": src, "ttft_s": _r(r.ttft(), 3), "decode_tps": _r(r.decode_tps(n), 1), "streamed": r.streamed, "finish_reason": r.finish, "chunks": r.text_chunks, "reasoning_text": r.reasoning} if not r.streamed: d["decode_tps_including_prefill"] = _r(n / (r.t_end - r.t0), 1) if r.t_end > r.t0 else None if n < 0.9 * prof["decode_tokens"]: d["short"] = True warnings.append(f"the {name} answer stopped at {n} tokens (finish reason {r.finish}), short of " f"{prof['decode_tokens']}; the number is what was measured") if r.reasoning: warnings.append(f"the {name} answer contained reasoning text; it is counted as decode tokens") out[name] = d return out def stage_prefill(ep, model, compat, prof, warnings): # calibration (and prefill warmup): tokens per character for this model's tokenizer cal = long_text(prof["calib_target"] * CHARS_PER_TOKEN, new_nonce()) say(f" prefill warmup and tokenizer calibration (about {prof['calib_target']} tokens) ...") try: r = ep.chat(chat_body(model, cal, 1, compat)) ct, src = r.prompt_tokens(len(cal)) except HTTPFail as e: return {"error": explain_http(e, model)} except BenchError as e: return {"error": str(e)} ratio = ct / len(cal) if src != "approx" else 1 / CHARS_PER_TOKEN target_chars = int(prof["prefill_target"] / ratio) text = long_text(target_chars, new_nonce()) say(f" prefill, about {prof['prefill_target']} tokens, 1 output token ...") try: r = ep.chat(chat_body(model, text, 1, compat)) except HTTPFail as e: return {"error": explain_http(e, model)} except BenchError as e: return {"error": str(e)} pt, src = r.prompt_tokens(len(text)) t = r.ttft() if r.streamed else (r.t_end - r.t0) d = {"prompt_tokens": pt, "token_source": src, "ttft_s": _r(t, 3), "streamed": r.streamed, "prefill_tps": _r(pt / t, 0) if t and t > 0 else None, "prompt_chars": len(text), "cold": True} if src == "timings": d["note"] = "prompt tokens are llama.cpp's prompt_n (tokens actually processed)" if not r.streamed: warnings.append("prefill: the server did not stream, so TTFT is the whole request time") return d def stage_multi(ep, model, compat, prof, users, warnings): say(f" {users} users, prose prompt, {prof['multi_tokens']} tokens each ...") results, errors = [None] * users, [None] * users barrier = threading.Barrier(users) def worker(i, max_tokens, store): try: barrier.wait(timeout=60) store[i] = ep.chat(chat_body(model, PROMPTS["prose"], max_tokens, compat)) except HTTPFail as e: errors[i] = explain_http(e, model) except (BenchError, threading.BrokenBarrierError) as e: errors[i] = str(e) or e.__class__.__name__ # warmup at the same concurrency, short answers warm = [None] * users ths = [threading.Thread(target=worker, args=(i, prof["warm_tokens"], warm), daemon=True) for i in range(users)] for t in ths: t.start() for t in ths: t.join() if any(errors): return {"error": "warmup failed: " + next(e for e in errors if e)} barrier.reset() ths = [threading.Thread(target=worker, args=(i, prof["multi_tokens"], results), daemon=True) for i in range(users)] for t in ths: t.start() for t in ths: t.join() bad = [e for e in errors if e] if bad: return {"error": f"{len(bad)} of {users} requests failed: {bad[0]}"} toks, per, srcs = [], [], set() for r in results: n, src = r.completion_tokens() toks.append(n) srcs.add(src) per.append(r.decode_tps(n)) d = {"users": users, "max_tokens": prof["multi_tokens"], "completion_tokens_total": sum(toks), "token_source": "approx" if "approx" in srcs else ("timings" if "timings" in srcs else "usage"), "streamed": all(r.streamed for r in results)} firsts = [r.t_first for r in results if r.t_first is not None] lasts = [r.t_last for r in results if r.t_last is not None] if d["streamed"] and len(firsts) == users and len(lasts) == users and max(lasts) > min(firsts): d["aggregate_tps"] = _r(sum(toks) / (max(lasts) - min(firsts)), 1) good = [p for p in per if p] d["per_user_tps_mean"] = _r(sum(good) / len(good), 1) if good else None d["per_user_tps_min"] = _r(min(good), 1) if good else None d["ttft_s_max"] = _r(max(r.ttft() for r in results), 3) if min(lasts) < max(firsts): d["queued"] = True warnings.append(f"{users} users: some requests started only after others had finished, so the server ran " f"fewer than {users} at once (check its parallel slots or max sequences)") else: start = min(r.t0 for r in results) end = max(r.t_end for r in results) d["aggregate_tps_including_prefill"] = _r(sum(toks) / (end - start), 1) if end > start else None d["wall_s"] = _r(max(r.t_end for r in results) - min(r.t0 for r in results), 2) return d def _r(v, nd): if v is None: return None return int(round(v)) if nd == 0 else round(v, nd) # ------------------------------------------------------------------------------------------------------- report def summarize_gpus(gpus, sampler): """Fold idle values and samples under load into per-card link and power, plus fields for the wiki.""" cards = gpus.get("cards", []) for c in cards: s = (sampler.seen.get(c["index"]) if sampler else None) or {} c["link_gen_load"] = s.get("gen") if s.get("gen") is not None else c.get("link_gen_idle") c["link_width_load"] = s.get("width") if s.get("width") is not None else c.get("link_width_idle") c["power_draw_peak_w"] = s.get("power") cmp_cards = [c for c in cards if c.get("variant")] w = {} if not cmp_cards: return w variants = sorted({c["variant"] for c in cmp_cards}) w["cards"] = variants[0] if len(variants) == 1 else "both" w["card_count"] = len(cmp_cards) gens = {c.get("link_gen_load") for c in cmp_cards} if len(gens) == 1 and None not in gens: g = gens.pop() w["pcie_link"] = f"Gen{g}" if g in (1, 2, 3) else None w["link_text"] = f"Gen{g}" elif None not in gens: w["pcie_link"] = "mixed" w["link_text"] = "mixed (" + ", ".join(f"Gen{g}" for g in sorted(gens)) + ")" widths = {c.get("link_width_load") for c in cmp_cards} if len(widths) == 1 and None not in widths: x = widths.pop() w["pcie_width"] = f"x{x}" if x in (1, 4, 16) else None w["width_text"] = f"x{x}" elif None not in widths: w["pcie_width"] = "mixed" w["width_text"] = "mixed (" + ", ".join(f"x{x}" for x in sorted(widths)) + ")" caps = {c.get("power_limit_w") for c in cmp_cards} if len(caps) == 1 and None not in caps: p = int(round(caps.pop())) w["power_w"] = p w["power_cap"] = f"{p} W" elif None not in caps: lo, hi = int(round(min(caps))), int(round(max(caps))) w["power_cap"] = f"{lo}-{hi} W" drivers = {c.get("driver") for c in cmp_cards if c.get("driver")} if len(drivers) == 1: w["driver"] = drivers.pop() return w def clean(s): """A value safe inside a template parameter: one line, no pipes or braces.""" s = re.sub(r"\s+", " ", str(s)).strip() return s.replace("|", "/").replace("{{", "(").replace("}}", ")").replace("[[", "").replace("]]", "") def num(v): return f"{v:,}" if isinstance(v, int) else (f"{v:,.1f}" if isinstance(v, float) else str(v)) def bench_note(rep, limit=300): a, s = rep["settings"], rep["stages"] parts = [f"cmpbench {VERSION} (Benchmark protocol {PROTOCOL})"] head = "temperature 0, thinking " + ("off" if a["thinking"] == "off" else "server default") head += f", {a['decode_tokens']:,}-token answers, code prompt" single = s.get("single") or {} others = [f"{k} {single[k]['decode_tps']}" for k in ("structured", "prose") if isinstance(single.get(k), dict) and single[k].get("decode_tps") is not None] if others: head += " (" + ", ".join(others) + ")" parts.append(head) m = s.get("multi") or {} if m.get("aggregate_tps") is not None: parts.append(f"{m['users']} users at {m['max_tokens']} tokens") p = s.get("prefill") or {} if p.get("prefill_tps") is not None: parts.append(f"cold prefill at {p['prompt_tokens']:,} tokens") if rep.get("gpu_wiki", {}).get("width_text") and not rep["gpu_wiki"].get("pcie_width"): parts.append("PCIe " + rep["gpu_wiki"].get("link_text", "") + " " + rep["gpu_wiki"]["width_text"]) if rep["token_source"] == "approx": parts.append("token counts approximated at 4 characters per token") if a["quick"]: parts.append("quick run, not protocol numbers") shorts = [k for k in ORDER if isinstance(single.get(k), dict) and single[k].get("short")] if shorts: parts.append("short answers: " + ", ".join(f"{k} {single[k]['completion_tokens']}" for k in shorts)) note = "; ".join(parts[:1]) + ": " + "; ".join(parts[1:]) if a.get("note"): room = limit - len(note) - 2 user = clean(a["note"]) if room > 10: note += "; " + (user if len(user) <= room else user[:room - 3].rstrip() + "...") return clean(note)[:limit] def wikitext(rep): a, s, g, srv = rep["settings"], rep["stages"], rep.get("gpu_wiki", {}), rep["server"] single = s.get("single") or {} code = single.get("code") or {} p = s.get("prefill") or {} m = s.get("multi") or {} head = [] if a["quick"]: head.append("") head.append(f"") rec = list(head) rec.append("") if code.get("decode_tps") is not None: rec += [f"|decode_tps={code['decode_tps']}", f"|decode_single={code['decode_tps']}"] if m.get("aggregate_tps") is not None: rec += [f"|decode_agg={m['aggregate_tps']}", f"|agg_users={m['users']}"] if p.get("prefill_tps") is not None: rec += [f"|prefill_tps={p['prefill_tps']}", f"|prefill_prompt={p['prompt_tokens']:,} tokens"] for k in ("cards", "card_count", "pcie_link", "pcie_width", "power_w", "power_cap", "driver"): if g.get(k) is not None: rec.append(f"|{k}={clean(g[k])}") if g.get("width_text") and not g.get("pcie_width"): rec.append(f"") if g.get("link_text") and not g.get("pcie_link"): rec.append(f"") if srv.get("context_tokens"): rec.append(f"|context_tokens={srv['context_tokens']}") if srv.get("engine_family"): rec.append(f"|engine_family={srv['engine_family']}") if srv.get("engine_version"): rec.append(f"|engine_version={clean(srv['engine_version'])}") if a.get("spec_decode"): rec.append(f"|spec_decode={clean(a['spec_decode'])}") else: rec.append("") rec.append(f"|bench_note={rep['bench_note']}") rec.append(f"|tested_on={rep['date']}") rec.append("") conf = list(head) conf.append("") conf.append("|result=") if g.get("cards"): conf.append(f"|cards={g['cards']}") conf.append(f"|date={rep['date']}") conf.append(f"|numbers={rep['numbers']}") notes = f"Measured with [[Benchmark protocol]] (cmpbench {VERSION})." if a["quick"]: notes += " Quick run, not protocol numbers." if a.get("note"): notes += " " + clean(a["note"]) conf.append(f"|notes={notes}") return "\n".join(rec), "\n".join(conf) def numbers_line(rep): s, g, a = rep["stages"], rep.get("gpu_wiki", {}), rep["settings"] single = s.get("single") or {} bits = [] vals = [(k, single[k]["decode_tps"]) for k in ORDER if isinstance(single.get(k), dict) and single[k].get("decode_tps") is not None] if vals: bits.append("1 user " + " / ".join(str(v) for _, v in vals) + " tok/s (" + " / ".join(k for k, _ in vals) + f", {a['decode_tokens']:,}-token answers)") m = s.get("multi") or {} if m.get("aggregate_tps") is not None: bits.append(f"{m['users']} users {num(m['aggregate_tps'])} tok/s total") p = s.get("prefill") or {} if p.get("prefill_tps") is not None: bits.append(f"prefill {num(p['prefill_tps'])} tok/s at {p['prompt_tokens']:,} tokens") setup = [] if g.get("card_count"): setup.append(f"{g['card_count']}x {g['cards'] if g['cards'] != 'both' else '8 GB and 10 GB'}") if g.get("link_text") or g.get("width_text"): setup.append(("PCIe " + (g.get("link_text") or "") + " " + (g.get("width_text") or "")).strip()) if g.get("power_cap"): setup.append(g["power_cap"]) if g.get("driver"): setup.append("driver " + g["driver"]) if a.get("spec_decode"): sd = "spec decode " + clean(a["spec_decode"]) acc = (s.get("spec") or {}).get("mean_acceptance_length") if acc: sd += f" (mean acceptance length {acc})" setup.append(sd) line = "; ".join(bits) if setup: line += "; " + ", ".join(setup) line = f"cmpbench {VERSION}: " + line if a["quick"]: line += "; quick run" return clean(line) def human_table(rep): a, s, srv = rep["settings"], rep["stages"], rep["server"] L = [] L.append(f"cmpbench {VERSION} (cmp.wiki Benchmark protocol {PROTOCOL}) {rep['timestamp']}") L.append(f"Endpoint {rep['endpoint']} model {a['model']}" + (f" engine {srv['engine_family']}" if srv.get("engine_family") else "") + (f" {srv['engine_version']}" if srv.get("engine_version") else "")) src = {"usage": "the server's usage fields", "timings": "llama.cpp timings", "none": "nothing (no stage finished)", "approx": "an APPROXIMATION (4 characters per token): the server sent no token counts"} L.append("Tokens from " + src.get(rep["token_source"], rep["token_source"]) + (" QUICK RUN, not protocol numbers" if a["quick"] else "")) L.append("") L.append(f"{'Stage':<36}{'tokens':>8}{'TTFT s':>9}{'tok/s':>10}") single = s.get("single") or {} for k in ORDER: d = single.get(k) if d is None: continue label = f"1 user, {k}" + (" (decode_single)" if k == "code" else "") if d.get("error"): L.append(f"{label:<36} ERROR {d['error']}") continue tps = d.get("decode_tps") tps_s = f"{tps:.1f}" if tps is not None else (f"{d.get('decode_tps_including_prefill')} incl. prefill" if d.get("decode_tps_including_prefill") else "n/a") L.append(f"{label:<36}{d['completion_tokens']:>8}{_f(d.get('ttft_s'), 3):>9}{tps_s:>10}") p = s.get("prefill") if p: if p.get("error"): L.append(f"{'Prefill':<36} ERROR {p['error']}") else: L.append(f"{'Prefill, cold':<36}{p['prompt_tokens']:>8}{_f(p.get('ttft_s'), 3):>9}" f"{(str(p['prefill_tps']) if p.get('prefill_tps') is not None else 'n/a'):>10}") m = s.get("multi") if m: label = f"{m.get('users', a['users'])} users, prose" if m.get("error"): L.append(f"{label:<36} ERROR {m['error']}") elif m.get("aggregate_tps") is not None: L.append(f"{label + ' (decode_agg)':<36}{m['completion_tokens_total']:>8}{_f(m.get('ttft_s_max'), 3):>9}" f"{m['aggregate_tps']:>10.1f}") L.append(f"{' per user, mean / min':<36}{'':>8}{'':>9}" f"{str(m.get('per_user_tps_mean')) + ' / ' + str(m.get('per_user_tps_min')):>10}") else: L.append(f"{label:<36}{m['completion_tokens_total']:>8}{'':>9}" f"{str(m.get('aggregate_tps_including_prefill')) + ' incl. prefill':>10}") sp = s.get("spec") if sp and sp.get("mean_acceptance_length"): L.append(f"Speculative decoding (server counters, 1-user stage): acceptance rate {sp['acceptance_rate']}, " f"mean acceptance length {sp['mean_acceptance_length']}") gp = rep.get("gpus") or {} L.append("") if not gp.get("available"): L.append("GPUs not read: " + gp.get("reason", "skipped")) else: L.append("GPUs on this machine" + ("" if rep.get("endpoint_local") else " (the endpoint is remote: these may not be the server's cards)")) L.append(f" {'idx':<4}{'name':<34}{'variant':<9}{'device':<11}{'link (load)':<13}{'cap W':>7}{'peak W':>8}" f"{'MiB':>8}") for c in gp.get("cards", []): link = "" if c.get("link_gen_load") is not None or c.get("link_width_load") is not None: link = f"Gen{c.get('link_gen_load')} x{c.get('link_width_load')}" L.append(f" {str(c.get('index')):<4}{str(c.get('name'))[:33]:<34}{str(c.get('variant') or '-'):<9}" f"{(c.get('device_id') or '-').replace('10de:', ''):<11}{link:<13}{_f(c.get('power_limit_w'), 0):>7}" f"{_f(c.get('power_draw_peak_w'), 0):>8}{str(c.get('memory_total_mib') or ''):>8}") if gp.get("cards"): L.append(f" driver {gp['cards'][0].get('driver')}" + ("" if gp.get("lspci") else " (lspci not available: device ids from nvidia-smi)")) if rep["warnings"]: L.append("") L.append("Notes") for w in rep["warnings"]: L.append(" * " + w) return "\n".join(L) def _f(v, nd): if v is None: return "-" return f"{v:.{nd}f}" if isinstance(v, (int, float)) else str(v) # --------------------------------------------------------------------------------------------------------- main def parse_args(argv): ap = argparse.ArgumentParser( prog="cmpbench.py", description="cmp.wiki Benchmark protocol " + PROTOCOL + " for OpenAI-compatible servers " "(vLLM, SGLang, llama.cpp server, TabbyAPI). https://cmp.wiki/wiki/Benchmark_protocol", epilog="Example: python3 cmpbench.py --base-url http://localhost:8000/v1 --model glm-5.3-flash " "--spec-decode \"MTP 3\" --json run1.json") ap.add_argument("--base-url", required=True, help="server URL up to /v1, e.g. http://localhost:8000/v1") ap.add_argument("--model", required=True, help="model id as the server lists it (GET /v1/models)") ap.add_argument("--api-key", default=os.environ.get("CMPBENCH_API_KEY"), help="bearer token, if the server needs one (or set CMPBENCH_API_KEY)") ap.add_argument("--quick", action="store_true", help="short smoke test (128-token answers, 2k prefill); " "its numbers are not protocol numbers") ap.add_argument("--users", type=int, default=8, help="concurrent requests in the many-user stage (default 8)") ap.add_argument("--no-concurrent", action="store_true", help="skip the many-user stage") ap.add_argument("--json", metavar="FILE", help="write the JSON report here ('-' prints it); default " "cmpbench-.json in the current directory") ap.add_argument("--note", default="", help="free text added to bench_note and the confirmation notes") ap.add_argument("--spec-decode", default="", metavar="TEXT", help="speculative decoding as run, e.g. \"MTP 3\", \"DFlash2, 7 draft tokens\" or none") ap.add_argument("--thinking", choices=("off", "default"), default="off", help="off (default) asks the chat template for no thinking; default sends nothing") ap.add_argument("--gpus", default=None, help="comma-separated nvidia-smi indices of the cards the server uses") ap.add_argument("--no-gpu-info", action="store_true", help="do not run nvidia-smi or lspci") ap.add_argument("--no-metrics", action="store_true", help="do not read the server's /metrics") ap.add_argument("--no-model-check", action="store_true", help="send --model even if /v1/models does not list it") ap.add_argument("--timeout", type=float, default=900.0, help="seconds per request (default 900)") ap.add_argument("--insecure", action="store_true", help="skip TLS certificate checks (https endpoints)") ap.add_argument("--version", action="version", version=f"cmpbench {VERSION} (protocol {PROTOCOL})") a = ap.parse_args(argv) if a.users < 1 or a.users > 256: ap.error("--users must be between 1 and 256") if a.gpus is not None: try: a.gpus = {int(x) for x in a.gpus.split(",") if x.strip()} except ValueError: ap.error("--gpus takes indices such as 0,1,2,3") return a def main(argv=None): a = parse_args(sys.argv[1:] if argv is None else argv) prof = dict(QUICK if a.quick else FULL) now = datetime.now(timezone.utc) warnings = [] try: ep = Endpoint(a.base_url, a.api_key, a.timeout, a.insecure) except BenchError as e: say("cmpbench: " + str(e)) return 2 rep = {"tool": "cmpbench", "version": VERSION, "protocol": PROTOCOL, "timestamp": now.strftime("%Y-%m-%d %H:%M UTC"), "date": now.strftime("%Y-%m-%d"), "endpoint": ep.display, "endpoint_local": ep.is_local(), "settings": {"model": a.model, "quick": a.quick, "users": a.users, "concurrent": not a.no_concurrent, "thinking": a.thinking, "note": a.note, "spec_decode": a.spec_decode, "temperature": 0, **prof}, "stages": {}, "warnings": warnings} say(f"cmpbench {VERSION}: {ep.display}, model {a.model}" + (" (quick)" if a.quick else "")) try: rep["server"] = server_info(ep, a.model, not a.no_model_check) if rep["server"].get("models_note"): warnings.append(rep["server"]["models_note"]) gpus = {"available": False, "reason": "--no-gpu-info"} if a.no_gpu_info else collect_gpus(a.gpus) rep["gpus"] = gpus if gpus.get("available") and not ep.is_local(): warnings.append("the endpoint is not on this machine: the GPU details describe this machine's cards; " "run the script on the inference server, or pass --no-gpu-info") if gpus.get("available") and not any(c.get("variant") for c in gpus["cards"]): warnings.append("no CMP 170HX (10de:20c2 or 10de:2082) among the GPUs read, so cards, link and power are " "left out of the wiki block") if gpus.get("available") and a.gpus is None and sum(1 for c in gpus["cards"] if c.get("variant")) > 1: warnings.append("card_count counts every CMP 170HX on this machine; pass --gpus to count only the cards " "the server uses") say("warmup ...") compat, _ = detect_compat(ep, a.model, a.thinking == "off", prof["warm_tokens"]) rep["compat"] = compat if a.thinking == "off" and not compat["thinking_hint"]: warnings.append("the server refused chat_template_kwargs, so thinking was left at the server default") if not compat["stream_options"]: warnings.append("the server refused stream_options, so usage counts come only if it sends them anyway") except BenchError as e: say("cmpbench: " + str(e)) return 2 except KeyboardInterrupt: say("cmpbench: interrupted") return 130 sampler = None if rep["gpus"].get("available"): sampler = Sampler() sampler.start() metrics = None if a.no_metrics else read_metrics(ep) rc = 0 try: say("stage 1: decode with one user") rep["stages"]["single"] = stage_single(ep, a.model, compat, prof, warnings) if metrics is not None: sp = spec_summary(metrics, read_metrics(ep)) if sp: rep["stages"]["spec"] = sp warnings.append("speculative-decoding counters are server-wide: other traffic during the run " "would mix in") say("stage 2: prefill") rep["stages"]["prefill"] = stage_prefill(ep, a.model, compat, prof, warnings) if not a.no_concurrent: say("stage 3: decode with many users") rep["stages"]["multi"] = stage_multi(ep, a.model, compat, prof, a.users, warnings) except KeyboardInterrupt: say("cmpbench: interrupted; the report covers the stages that finished") rc = 130 finally: if sampler: sampler.stop() srcs = set() for d in (rep["stages"].get("single") or {}).values(): if isinstance(d, dict) and d.get("token_source"): srcs.add(d["token_source"]) for k in ("prefill", "multi"): d = rep["stages"].get(k) or {} if d.get("token_source"): srcs.add(d["token_source"]) rep["token_source"] = ("none" if not srcs else "approx" if "approx" in srcs else "timings" if "timings" in srcs else "usage") if rep["token_source"] == "approx": warnings.append("token counts are APPROXIMATE (4 characters per token): the server sent no usage fields") if any(isinstance(d, dict) and d.get("streamed") is False for d in (rep["stages"].get("single") or {}).values()): warnings.append("the server did not stream: decode tok/s and TTFT cannot be separated from prefill") errors = [k for k, d in rep["stages"].items() if isinstance(d, dict) and d.get("error")] errors += [f"single/{k}" for k, d in (rep["stages"].get("single") or {}).items() if d.get("error")] if errors and rc == 0: rc = 1 rep["gpu_wiki"] = summarize_gpus(rep["gpus"], sampler) if rep["gpus"].get("available") else {} if sampler: rep["gpus"]["samples_under_load"] = sampler.samples rep["bench_note"] = bench_note(rep) rep["numbers"] = numbers_line(rep) rec, conf = wikitext(rep) rep["wikitext"] = {"recipe": rec, "confirmation": conf} print(human_table(rep)) print() print("---- wikitext: Recipe infobox fields " + "-" * 40) print(rec) print() print("---- wikitext: Confirmation fields " + "-" * 42) print(conf) print() doc = json.dumps(rep, indent=2, sort_keys=False) if a.json == "-": print(doc) else: path = a.json or f"cmpbench-{now.strftime('%Y%m%d-%H%M%S')}.json" try: with open(path, "w", encoding="utf-8") as f: f.write(doc + "\n") print(f"JSON report: {path}") except OSError as e: say(f"cmpbench: cannot write {path}: {e}") rc = rc or 1 return rc if __name__ == "__main__": sys.exit(main())