From 75b2c8958c29d400c63aa90c46ca227c1bbaf75e Mon Sep 17 00:00:00 2001 From: Christian Meyer Date: Mon, 31 Aug 2026 16:38:59 +0200 Subject: [PATCH] feat: add native MLX inference benchmarks --- benchmarks/.gitignore | 2 + benchmarks/inference/README.md | 76 +++ benchmarks/inference/benchmark.py | 500 +++++++++++++++++++ benchmarks/inference/coding_benchmark.py | 364 ++++++++++++++ benchmarks/inference/direct_mlx_benchmark.py | 155 ++++++ 5 files changed, 1097 insertions(+) create mode 100644 benchmarks/.gitignore create mode 100644 benchmarks/inference/README.md create mode 100644 benchmarks/inference/benchmark.py create mode 100644 benchmarks/inference/coding_benchmark.py create mode 100644 benchmarks/inference/direct_mlx_benchmark.py diff --git a/benchmarks/.gitignore b/benchmarks/.gitignore new file mode 100644 index 0000000..43ae0e2 --- /dev/null +++ b/benchmarks/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.py[cod] diff --git a/benchmarks/inference/README.md b/benchmarks/inference/README.md new file mode 100644 index 0000000..e4ca47b --- /dev/null +++ b/benchmarks/inference/README.md @@ -0,0 +1,76 @@ +# Ollama versus vLLM-Metal benchmark + +This benchmark compares native Apple-Silicon execution through Ollama and vLLM-Metal. It does not change the Laravel application, the automation worker, or their model configuration. + +The completed 2026-08-31 comparison and recommendation are documented in `REPORT.md`. + +## Model classification + +- Ollama: `qwen3.8:27b-mlx` +- vLLM-Metal: `mlx-community/Qwen3.8-27B-nvfp4`, revision `5ff8ef173ad0d7c3aae92f0be43031a6ab8067c6` +- Both derive from Qwen3.8-27B and use nominal NVFP4 group-size-16 quantization. +- They are **not byte-identical artifacts**. A complete corresponding 44,564,480-byte packed weight tensor produced different SHA-256 hashes. Results therefore compare deployable runtime/model packages, not runtime overhead with identical bytes. + +The MXFP8 comparison uses Ollama `qwen3.8:27b-mxfp8` and +`mlx-community/Qwen3.8-27B-mxfp8`, revision +`d48d163bcdf24acaf656474854ab88ea17d65bd1`. Both declare group-size-32 MXFP8 +but are also not byte-identical. + +## Controlled protocol + +1. Run both services natively on the same otherwise-idle Apple Silicon host. +2. Use a context limit of 8,192, temperature 0, seed 42, and 128 maximum output tokens. + Use raw text completions and disable Ollama speculative drafting so neither runtime has a draft model. +3. Run two warmup rounds before measured rounds. +4. Run five measured rounds at concurrency 1 and five groups at concurrency 4. +5. Alternate three fixed prompts to reduce prefix-cache bias. +6. Compare client-observed TTFT and total latency, warmed decode speed, and aggregate output throughput. +7. Preserve metadata, prompts, per-request measurements, errors, and summaries as JSON and CSV. +8. Treat startup/model loading as a separate cold-load observation; do not mix it into warm inference percentiles. + +`benchmark.py` uses only the Python standard library. It expects Ollama at `127.0.0.1:11434` and the vLLM OpenAI-compatible API at `127.0.0.1:8001`. + +## Complex coding suite + +`coding_benchmark.py` sends the same three complex implementation tasks through each runtime's +native chat API: an asynchronous Python cache, an idempotent Laravel money transfer, and a +concurrent TypeScript DAG scheduler. Thinking and speculative drafting are disabled, temperature is +zero, and full generated responses are retained for inspection. The summary records TTFT, total +latency, output tokens, decode rate, and finish reason. + +When every response reaches the configured output limit, the suite measures time for a fixed amount +of long-form coding output rather than time to a naturally completed solution. This distinction must +be preserved in reports. + +## Direct MLX control + +`direct_mlx_benchmark.py` loads a Hugging Face checkpoint directly with MLX-LM and measures the +same raw prompts at concurrency one. Running it against the exact NVFP4 or MXFP8 checkpoint used by +vLLM isolates short-generation MLX performance from HTTP, scheduling, paged-cache, and request-state +overhead. It is not a concurrent-serving benchmark and cannot isolate Ollama because Ollama's packed +artifact is different. + +## Metrics + +- **TTFT:** request start to the first non-empty generated text chunk at the client. +- **Total latency:** request start to completion of the response stream. +- **Decode tokens/s:** Ollama's server-reported evaluation rate; for vLLM, output tokens divided by post-TTFT client time. +- **Request tokens/s:** output tokens divided by total client latency, including prefill and network overhead. +- **Aggregate output tokens/s:** all successful generated tokens in a concurrent group divided by group wall time. +- **p50/p95:** linearly interpolated percentiles over measured requests or groups; warmups are excluded. + +Prompt throughput is reported when the server exposes both prompt token count and prompt duration. A missing value is not estimated. + +## Result files + +Each run creates a timestamped directory below `results/` containing: + +- `metadata.json`: environment, arguments, model metadata, prompts, and summary +- `raw.json`: complete per-request evidence +- `raw.csv`: tabular per-request evidence +- `summary.json`: grouped statistics + +The coding suite stores equivalent evidence below `coding-results/`. Direct MLX controls store +`metadata.json`, `raw.json`, and `summary.json` below `direct-results/`. + +Do not compare runs performed under materially different memory pressure, thermal conditions, model revisions, or runtime versions. diff --git a/benchmarks/inference/benchmark.py b/benchmarks/inference/benchmark.py new file mode 100644 index 0000000..f239ad1 --- /dev/null +++ b/benchmarks/inference/benchmark.py @@ -0,0 +1,500 @@ +#!/usr/bin/env python3 +"""Reproducible HTTP benchmark for Ollama and OpenAI-compatible runtimes.""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import csv +import json +import math +import platform +import statistics +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.request +import uuid +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +PROMPTS = { + "short": ( + "In exactly four concise bullet points, explain why database transactions are useful. " + "Do not use a heading." + ), + "medium": ( + "You are reviewing a Laravel service that transfers money between two accounts. " + "Explain the failure modes involving concurrency, partial writes, retries, and audit logs. " + "Then provide a robust implementation plan. Keep the response technical and concise." + ), + "code": ( + "Write a Python 3.12 function named merge_intervals that accepts a list of integer pairs, " + "validates each pair, merges overlapping or adjacent intervals, and returns sorted tuples. " + "Include type hints, a short docstring, and five assert-based examples. Output only code." + ), +} + +CSV_FIELDS = ( + "run_id", + "runtime", + "model", + "phase", + "concurrency", + "round", + "slot", + "request_id", + "prompt_name", + "prompt_chars", + "max_tokens", + "started_at", + "ttft_s", + "total_s", + "prompt_tokens", + "output_tokens", + "prompt_tps", + "decode_tps", + "request_tps", + "group_elapsed_s", + "group_output_tps", + "error", +) + + +@dataclass(frozen=True) +class RuntimeConfig: + name: str + kind: str + base_url: str + model: str + + +@dataclass +class Measurement: + run_id: str + runtime: str + model: str + phase: str + concurrency: int + round: int + slot: int + request_id: str + prompt_name: str + prompt_chars: int + max_tokens: int + started_at: str + ttft_s: float | None = None + total_s: float | None = None + prompt_tokens: int | None = None + output_tokens: int | None = None + prompt_tps: float | None = None + decode_tps: float | None = None + request_tps: float | None = None + group_elapsed_s: float | None = None + group_output_tps: float | None = None + error: str | None = None + + +def utc_now() -> str: + return datetime.now(UTC).isoformat() + + +def percentile(values: list[float], fraction: float) -> float | None: + if not values: + return None + ordered = sorted(values) + position = (len(ordered) - 1) * fraction + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) + + +def safe_version(command: list[str]) -> str | None: + try: + result = subprocess.run(command, capture_output=True, text=True, timeout=15, check=False) + except (OSError, subprocess.SubprocessError): + return None + value = (result.stdout or result.stderr).strip() + return value or None + + +def http_json(url: str, timeout: float = 15) -> dict[str, Any]: + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.load(response) + + +def read_ndjson(response: Any, started: float) -> tuple[float | None, dict[str, Any], int]: + first_token_at: float | None = None + output_chars = 0 + terminal: dict[str, Any] = {} + for raw_line in response: + if not raw_line.strip(): + continue + event = json.loads(raw_line) + content = event.get("response", "") + if content: + output_chars += len(content) + if first_token_at is None: + first_token_at = time.perf_counter() + if event.get("done"): + terminal = event + return (first_token_at - started if first_token_at else None, terminal, output_chars) + + +def read_sse(response: Any, started: float) -> tuple[float | None, dict[str, Any], int]: + first_token_at: float | None = None + output_chars = 0 + terminal: dict[str, Any] = {} + for raw_line in response: + line = raw_line.decode("utf-8").strip() + if not line.startswith("data:"): + continue + payload = line[5:].strip() + if payload == "[DONE]": + break + event = json.loads(payload) + terminal = event + choices = event.get("choices") or [] + if choices: + text = choices[0].get("text") or choices[0].get("delta", {}).get("content") or "" + if text: + output_chars += len(text) + if first_token_at is None: + first_token_at = time.perf_counter() + return (first_token_at - started if first_token_at else None, terminal, output_chars) + + +def benchmark_request( + runtime: RuntimeConfig, + prompt_name: str, + prompt: str, + max_tokens: int, + seed: int, + timeout: float, + run_id: str, + phase: str, + concurrency: int, + round_number: int, + slot: int, +) -> Measurement: + measurement = Measurement( + run_id=run_id, + runtime=runtime.name, + model=runtime.model, + phase=phase, + concurrency=concurrency, + round=round_number, + slot=slot, + request_id=str(uuid.uuid4()), + prompt_name=prompt_name, + prompt_chars=len(prompt), + max_tokens=max_tokens, + started_at=utc_now(), + ) + headers = {"Content-Type": "application/json", "Accept": "text/event-stream"} + if runtime.kind == "ollama": + url = f"{runtime.base_url.rstrip('/')}/api/generate" + payload = { + "model": runtime.model, + "prompt": prompt, + "stream": True, + "raw": True, + "keep_alive": "30m", + "options": { + "temperature": 0, + "seed": seed, + "num_predict": max_tokens, + "num_ctx": 8192, + "draft_num_predict": 0, + }, + } + else: + url = f"{runtime.base_url.rstrip('/')}/v1/completions" + payload = { + "model": runtime.model, + "prompt": prompt, + "stream": True, + "stream_options": {"include_usage": True}, + "temperature": 0, + "seed": seed, + "max_tokens": max_tokens, + } + + started = time.perf_counter() + try: + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers=headers, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + if runtime.kind == "ollama": + ttft, terminal, output_chars = read_ndjson(response, started) + measurement.prompt_tokens = terminal.get("prompt_eval_count") + measurement.output_tokens = terminal.get("eval_count") + prompt_ns = terminal.get("prompt_eval_duration") + decode_ns = terminal.get("eval_duration") + if prompt_ns and measurement.prompt_tokens: + measurement.prompt_tps = measurement.prompt_tokens / (prompt_ns / 1e9) + if decode_ns and measurement.output_tokens: + measurement.decode_tps = measurement.output_tokens / (decode_ns / 1e9) + else: + ttft, terminal, output_chars = read_sse(response, started) + usage = terminal.get("usage") or {} + measurement.prompt_tokens = usage.get("prompt_tokens") + measurement.output_tokens = usage.get("completion_tokens") + measurement.ttft_s = ttft + measurement.total_s = time.perf_counter() - started + if measurement.output_tokens and measurement.total_s: + measurement.request_tps = measurement.output_tokens / measurement.total_s + if measurement.decode_tps is None and measurement.ttft_s is not None: + decode_time = measurement.total_s - measurement.ttft_s + if decode_time > 0: + measurement.decode_tps = measurement.output_tokens / decode_time + if output_chars == 0: + measurement.error = "stream contained no generated text" + except urllib.error.HTTPError as error: + detail = error.read().decode("utf-8", errors="replace")[:1000] + measurement.error = f"HTTP {error.code}: {detail}" + measurement.total_s = time.perf_counter() - started + except (OSError, ValueError, json.JSONDecodeError) as error: + measurement.error = f"{type(error).__name__}: {error}" + measurement.total_s = time.perf_counter() - started + return measurement + + +def run_round( + runtime: RuntimeConfig, + prompt_items: list[tuple[str, str]], + max_tokens: int, + seed: int, + timeout: float, + run_id: str, + phase: str, + concurrency: int, + round_number: int, +) -> tuple[list[Measurement], float]: + barrier = threading.Barrier(concurrency) + + def execute(slot: int) -> Measurement: + prompt_name, prompt = prompt_items[(round_number * concurrency + slot) % len(prompt_items)] + barrier.wait() + return benchmark_request( + runtime, + prompt_name, + prompt, + max_tokens, + seed, + timeout, + run_id, + phase, + concurrency, + round_number, + slot, + ) + + group_started = time.perf_counter() + with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor: + measurements = list(executor.map(execute, range(concurrency))) + group_elapsed = time.perf_counter() - group_started + successful_tokens = sum(item.output_tokens or 0 for item in measurements if not item.error) + group_tps = successful_tokens / group_elapsed if group_elapsed else 0.0 + for item in measurements: + item.group_elapsed_s = group_elapsed + item.group_output_tps = group_tps + return measurements, group_elapsed + + +def summarize(measurements: list[Measurement]) -> list[dict[str, Any]]: + groups: dict[tuple[str, int], list[Measurement]] = {} + for measurement in measurements: + if measurement.phase != "measured": + continue + groups.setdefault((measurement.runtime, measurement.concurrency), []).append(measurement) + + summaries: list[dict[str, Any]] = [] + for (runtime, concurrency), items in sorted(groups.items()): + successful = [item for item in items if not item.error] + summary: dict[str, Any] = { + "runtime": runtime, + "model": items[0].model, + "concurrency": concurrency, + "requests": len(items), + "successes": len(successful), + "failures": len(items) - len(successful), + } + for field in ("ttft_s", "total_s", "prompt_tps", "decode_tps", "request_tps"): + values = [ + float(value) + for item in successful + if (value := getattr(item, field)) is not None + ] + summary[f"{field}_mean"] = statistics.fmean(values) if values else None + summary[f"{field}_p50"] = percentile(values, 0.50) + summary[f"{field}_p95"] = percentile(values, 0.95) + group_tps = [ + float(item.group_output_tps) + for item in successful + if item.slot == 0 and item.group_output_tps is not None + ] + summary["aggregate_output_tps_mean"] = statistics.fmean(group_tps) if group_tps else None + summary["aggregate_output_tps_p50"] = percentile(group_tps, 0.50) + summary["aggregate_output_tps_p95"] = percentile(group_tps, 0.95) + summaries.append(summary) + return summaries + + +def write_csv(path: Path, measurements: list[Measurement]) -> None: + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS, lineterminator="\n") + writer.writeheader() + for measurement in measurements: + writer.writerow(asdict(measurement)) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output-dir", type=Path, default=Path(__file__).parent / "results") + parser.add_argument("--runtime", choices=("all", "ollama", "vllm"), default="all") + parser.add_argument("--ollama-url", default="http://127.0.0.1:11434") + parser.add_argument("--ollama-model", default="qwen3.8:27b-mlx") + parser.add_argument("--vllm-url", default="http://127.0.0.1:8001") + parser.add_argument("--vllm-model", default="qwen3.8-27b-nvfp4") + parser.add_argument("--warmups", type=int, default=2) + parser.add_argument("--rounds", type=int, default=5) + parser.add_argument("--concurrency", type=int, nargs="+", default=(1, 4)) + parser.add_argument("--max-tokens", type=int, default=128) + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--timeout", type=float, default=600) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.warmups < 0 or args.rounds < 1 or any(value < 1 for value in args.concurrency): + raise SystemExit("warmups must be >= 0; rounds and concurrency must be >= 1") + + runtimes = { + "ollama": RuntimeConfig("ollama", "ollama", args.ollama_url, args.ollama_model), + "vllm": RuntimeConfig("vllm-metal", "openai", args.vllm_url, args.vllm_model), + } + selected = list(runtimes.values()) if args.runtime == "all" else [runtimes[args.runtime]] + for runtime in selected: + health_url = ( + f"{runtime.base_url.rstrip('/')}/api/tags" + if runtime.kind == "ollama" + else f"{runtime.base_url.rstrip('/')}/v1/models" + ) + try: + http_json(health_url) + except Exception as error: + raise SystemExit(f"{runtime.name} is not ready at {health_url}: {error}") from error + + run_id = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + output_dir = args.output_dir / run_id + output_dir.mkdir(parents=True, exist_ok=False) + metadata = { + "run_id": run_id, + "created_at": utc_now(), + "host": { + "platform": platform.platform(), + "machine": platform.machine(), + "python": platform.python_version(), + }, + "arguments": vars(args) | {"output_dir": str(args.output_dir)}, + "runtimes": [asdict(runtime) for runtime in selected], + "versions": { + "ollama": safe_version(["ollama", "--version"]), + "vllm": safe_version([str(Path.home() / ".venv-vllm-metal/bin/vllm"), "--version"]), + }, + "model_metadata": {}, + "prompts": PROMPTS, + "measurement_notes": { + "ttft": ( + "Client wall time from request start until first non-empty generated text chunk." + ), + "total": "Client wall time from request start until stream completion.", + "decode_tps": "Ollama server eval metric; otherwise output tokens / (total - TTFT).", + "aggregate_output_tps": "Sum of successful output tokens / concurrent group wall time.", + }, + } + for runtime in selected: + url = ( + f"{runtime.base_url.rstrip('/')}/api/show" + if runtime.kind == "ollama" + else f"{runtime.base_url.rstrip('/')}/v1/models" + ) + try: + if runtime.kind == "ollama": + request = urllib.request.Request( + url, + data=json.dumps({"model": runtime.model}).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=30) as response: + metadata["model_metadata"][runtime.name] = json.load(response) + else: + metadata["model_metadata"][runtime.name] = http_json(url) + except Exception as error: + metadata["model_metadata"][runtime.name] = {"metadata_error": str(error)} + + prompt_items = list(PROMPTS.items()) + measurements: list[Measurement] = [] + total_groups = len(selected) * sum(args.warmups + args.rounds for _ in args.concurrency) + completed_groups = 0 + for runtime in selected: + for concurrency in args.concurrency: + for phase, rounds in (("warmup", args.warmups), ("measured", args.rounds)): + for round_number in range(rounds): + items, elapsed = run_round( + runtime, + prompt_items, + args.max_tokens, + args.seed, + args.timeout, + run_id, + phase, + concurrency, + round_number, + ) + measurements.extend(items) + completed_groups += 1 + errors = sum(bool(item.error) for item in items) + print( + f"[{completed_groups}/{total_groups}] {runtime.name} phase={phase} " + f"concurrency={concurrency} round={round_number + 1}/{rounds} " + f"elapsed={elapsed:.3f}s errors={errors}", + flush=True, + ) + + summaries = summarize(measurements) + metadata["completed_at"] = utc_now() + metadata["summary"] = summaries + (output_dir / "raw.json").write_text( + json.dumps([asdict(item) | { + "group_elapsed_s": getattr(item, "group_elapsed_s", None), + "group_output_tps": getattr(item, "group_output_tps", None), + } for item in measurements], indent=2), + encoding="utf-8", + ) + (output_dir / "metadata.json").write_text( + json.dumps(metadata, indent=2, default=str), encoding="utf-8" + ) + (output_dir / "summary.json").write_text(json.dumps(summaries, indent=2), encoding="utf-8") + write_csv(output_dir / "raw.csv", measurements) + print(json.dumps(summaries, indent=2)) + print(f"Results: {output_dir}") + return 1 if any(item.error for item in measurements if item.phase == "measured") else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/inference/coding_benchmark.py b/benchmarks/inference/coding_benchmark.py new file mode 100644 index 0000000..8b6e215 --- /dev/null +++ b/benchmarks/inference/coding_benchmark.py @@ -0,0 +1,364 @@ +#!/usr/bin/env python3 +"""Measure end-to-end latency for deterministic, complex programming tasks.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import platform +import subprocess +import sys +import time +import urllib.error +import urllib.request +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +TASKS = { + "python_async_cache": """ +Act as a senior Python engineer. Implement a production-ready, single-file Python 3.12 module +containing a generic AsyncTTLCache[K, V]. Requirements: weighted LRU capacity; per-entry TTL based +on time.monotonic; single-flight get_or_create so concurrent misses for one key invoke the async +factory exactly once; cancellation of one waiter must not cancel the shared factory; exceptions are +never cached; invalidating a key while its factory is running must prevent stale publication; never +hold the cache lock while awaiting user code; clear(), invalidate(), and immutable hit/miss/eviction +statistics. Include concise pytest-asyncio tests covering races, cancellation, expiry, weighted +capacity, factory failure, and invalidation during creation. Use only the standard library plus +pytest. Output only the complete module, without Markdown fences or prose. +""".strip(), + "php_idempotent_transfer": """ +Act as a senior Laravel 12 engineer. Produce one self-contained PHP file that demonstrates an +idempotent money-transfer service and its PHPUnit tests. Store money as integer cents. Requirements: +validate positive amount and distinct accounts; use a database transaction; lock both account rows +in deterministic ID order; reject insufficient funds; use a unique client idempotency key so retries +return the original transfer; retry deadlocks without duplicating effects; persist an outbox event +in the same transaction; and prove with tests that duplicate and competing requests preserve +balances, create one transfer, and create one outbox event. Include minimal model/table assumptions +as comments. +Output only valid PHP, without Markdown fences or prose. +""".strip(), + "typescript_dag_scheduler": """ +Act as a senior TypeScript engineer. Implement a strict-mode, dependency-free DAG task scheduler in +one TypeScript file. Each task has an ID, dependency IDs, and an async run(signal) function. +Requirements: validate unknown dependencies and duplicate IDs; report an explicit cycle path; stable +input-order scheduling; configurable maximum concurrency; AbortSignal support; fail fast after the +first error while awaiting already-running tasks; never start dependents of failed tasks; return a +Map of typed results on success; and avoid unhandled promise rejections. Include a compact test +harness covering a diamond graph, concurrency enforcement, stable order, cycle detection, external +cancellation, and rejection. Output only TypeScript, without Markdown fences or prose. +""".strip(), +} + + +@dataclass(frozen=True) +class Runtime: + name: str + kind: str + base_url: str + model: str + + +@dataclass +class Result: + task: str + runtime: str + model: str + started_at: str + prompt_chars: int + max_tokens: int + ttft_s: float | None = None + total_s: float | None = None + prompt_tokens: int | None = None + output_tokens: int | None = None + decode_tps: float | None = None + finish_reason: str | None = None + output_chars: int = 0 + output_sha256: str | None = None + output: str = "" + error: str | None = None + + +def utc_now() -> str: + return datetime.now(UTC).isoformat() + + +def post_json(url: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]: + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.load(response) + + +def health(runtime: Runtime) -> None: + path = "/api/tags" if runtime.kind == "ollama" else "/v1/models" + with urllib.request.urlopen(f"{runtime.base_url.rstrip('/')}{path}", timeout=15): + pass + + +def consume_ollama(response: Any, started: float) -> tuple[float | None, str, dict[str, Any]]: + first_at: float | None = None + chunks: list[str] = [] + terminal: dict[str, Any] = {} + for line in response: + if not line.strip(): + continue + event = json.loads(line) + text = event.get("message", {}).get("content") or event.get("response") or "" + if text: + if first_at is None: + first_at = time.perf_counter() + chunks.append(text) + if event.get("done"): + terminal = event + return (first_at - started if first_at else None, "".join(chunks), terminal) + + +def consume_sse( + response: Any, started: float +) -> tuple[float | None, str, dict[str, Any], str | None]: + first_at: float | None = None + chunks: list[str] = [] + terminal: dict[str, Any] = {} + finish_reason: str | None = None + for raw_line in response: + line = raw_line.decode("utf-8").strip() + if not line.startswith("data:"): + continue + payload = line[5:].strip() + if payload == "[DONE]": + break + event = json.loads(payload) + terminal = event + for choice in event.get("choices") or []: + text = choice.get("text") or choice.get("delta", {}).get("content") or "" + if text: + if first_at is None: + first_at = time.perf_counter() + chunks.append(text) + finish_reason = choice.get("finish_reason") or finish_reason + return (first_at - started if first_at else None, "".join(chunks), terminal, finish_reason) + + +def request_payload( + runtime: Runtime, prompt: str, max_tokens: int, seed: int +) -> tuple[str, dict[str, Any]]: + if runtime.kind == "ollama": + return f"{runtime.base_url.rstrip('/')}/api/chat", { + "model": runtime.model, + "messages": [ + { + "role": "system", + "content": ( + "You are a senior software engineer. Follow every output constraint " + "exactly and return a complete solution." + ), + }, + {"role": "user", "content": prompt}, + ], + "think": False, + "stream": True, + "keep_alive": "30m", + "options": { + "temperature": 0, + "seed": seed, + "num_predict": max_tokens, + "num_ctx": 8192, + "draft_num_predict": 0, + }, + } + return f"{runtime.base_url.rstrip('/')}/v1/chat/completions", { + "model": runtime.model, + "messages": [ + { + "role": "system", + "content": ( + "You are a senior software engineer. Follow every output constraint exactly " + "and return a complete solution." + ), + }, + {"role": "user", "content": prompt}, + ], + "stream": True, + "stream_options": {"include_usage": True}, + "chat_template_kwargs": {"enable_thinking": False}, + "temperature": 0, + "seed": seed, + "max_tokens": max_tokens, + } + + +def run_task( + runtime: Runtime, + task_name: str, + prompt: str, + max_tokens: int, + seed: int, + timeout: float, +) -> Result: + result = Result( + task=task_name, + runtime=runtime.name, + model=runtime.model, + started_at=utc_now(), + prompt_chars=len(prompt), + max_tokens=max_tokens, + ) + url, payload = request_payload(runtime, prompt, max_tokens, seed) + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json", "Accept": "text/event-stream"}, + method="POST", + ) + started = time.perf_counter() + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + if runtime.kind == "ollama": + result.ttft_s, result.output, terminal = consume_ollama(response, started) + result.prompt_tokens = terminal.get("prompt_eval_count") + result.output_tokens = terminal.get("eval_count") + result.finish_reason = terminal.get("done_reason") + decode_ns = terminal.get("eval_duration") + if decode_ns and result.output_tokens: + result.decode_tps = result.output_tokens / (decode_ns / 1e9) + else: + result.ttft_s, result.output, terminal, result.finish_reason = consume_sse( + response, started + ) + usage = terminal.get("usage") or {} + result.prompt_tokens = usage.get("prompt_tokens") + result.output_tokens = usage.get("completion_tokens") + result.total_s = time.perf_counter() - started + if result.decode_tps is None and result.output_tokens and result.ttft_s is not None: + decode_s = result.total_s - result.ttft_s + if decode_s > 0: + result.decode_tps = result.output_tokens / decode_s + result.output_chars = len(result.output) + result.output_sha256 = hashlib.sha256(result.output.encode()).hexdigest() + if not result.output: + result.error = "empty generated output" + except urllib.error.HTTPError as error: + detail = error.read().decode("utf-8", errors="replace")[:2000] + result.error = f"HTTP {error.code}: {detail}" + result.total_s = time.perf_counter() - started + except (OSError, ValueError, json.JSONDecodeError) as error: + result.error = f"{type(error).__name__}: {error}" + result.total_s = time.perf_counter() - started + return result + + +def package_versions() -> dict[str, str | None]: + script = ( + "import importlib.metadata as m,json; " + "print(json.dumps({p:m.version(p) for p in " + "['vllm','vllm-metal','mlx','mlx-lm','mlx-vlm']}))" + ) + executable = Path.home() / ".venv-vllm-metal/bin/python" + try: + process = subprocess.run( + [str(executable), "-c", script], capture_output=True, text=True, timeout=30, check=True + ) + return json.loads(process.stdout) + except (OSError, subprocess.SubprocessError, json.JSONDecodeError): + return {} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--runtime", choices=("ollama", "vllm"), required=True) + parser.add_argument("--ollama-url", default="http://127.0.0.1:11434") + parser.add_argument("--ollama-model", default="qwen3.8:27b-mlx") + parser.add_argument("--vllm-url", default="http://127.0.0.1:8001") + parser.add_argument("--vllm-model", default="qwen3.8-27b-nvfp4") + parser.add_argument("--max-tokens", type=int, default=1536) + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--timeout", type=float, default=900) + parser.add_argument("--output-dir", type=Path, default=Path(__file__).parent / "coding-results") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.max_tokens < 1: + raise SystemExit("max-tokens must be positive") + runtime = ( + Runtime("ollama", "ollama", args.ollama_url, args.ollama_model) + if args.runtime == "ollama" + else Runtime("vllm-metal", "openai", args.vllm_url, args.vllm_model) + ) + try: + health(runtime) + except Exception as error: + raise SystemExit(f"{runtime.name} is unavailable: {error}") from error + + warmup = run_task(runtime, "warmup", "Output only the word READY.", 8, args.seed, args.timeout) + if warmup.error: + raise SystemExit(f"warmup failed: {warmup.error}") + print( + f"warmup runtime={runtime.name} total={warmup.total_s:.3f}s " + f"tokens={warmup.output_tokens}", + flush=True, + ) + + results: list[Result] = [] + for index, (task_name, prompt) in enumerate(TASKS.items(), start=1): + result = run_task(runtime, task_name, prompt, args.max_tokens, args.seed, args.timeout) + results.append(result) + print( + f"[{index}/{len(TASKS)}] task={task_name} total={result.total_s:.3f}s " + f"ttft={result.ttft_s:.3f}s tokens={result.output_tokens} " + f"decode={result.decode_tps:.2f}tok/s finish={result.finish_reason} " + f"error={result.error or 'none'}", + flush=True, + ) + + run_id = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + output_dir = args.output_dir / f"{run_id}-{runtime.name}-{runtime.model.replace(':', '_')}" + output_dir.mkdir(parents=True, exist_ok=False) + raw = { + "run_id": run_id, + "created_at": utc_now(), + "runtime": asdict(runtime), + "host": { + "platform": platform.platform(), + "machine": platform.machine(), + "python": platform.python_version(), + }, + "package_versions": package_versions(), + "settings": { + "temperature": 0, + "seed": args.seed, + "max_tokens": args.max_tokens, + "context_tokens": 8192, + "raw_completion": False, + "chat_template": True, + "thinking": False, + "ollama_speculative_drafting": False, + }, + "warmup": asdict(warmup), + "tasks": TASKS, + "results": [asdict(result) for result in results], + } + (output_dir / "results.json").write_text(json.dumps(raw, indent=2), encoding="utf-8") + summary = [ + { + key: value + for key, value in asdict(result).items() + if key not in {"output"} + } + for result in results + ] + (output_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") + print(f"Results: {output_dir}") + return 1 if any(result.error for result in results) else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/inference/direct_mlx_benchmark.py b/benchmarks/inference/direct_mlx_benchmark.py new file mode 100644 index 0000000..aaa5861 --- /dev/null +++ b/benchmarks/inference/direct_mlx_benchmark.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +"""Measure direct MLX-LM generation without an HTTP serving layer.""" + +from __future__ import annotations + +import argparse +import json +import math +import platform +import statistics +import time +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import mlx.core as mx +from benchmark import PROMPTS +from mlx_lm import load, stream_generate +from mlx_lm.sample_utils import make_sampler + + +def percentile(values: list[float], fraction: float) -> float: + ordered = sorted(values) + position = (len(ordered) - 1) * fraction + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) + + +def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: + measured = [row for row in rows if row["phase"] == "measured"] + result: dict[str, Any] = { + "runtime": "mlx-lm-direct", + "model": measured[0]["model"], + "requests": len(measured), + "successes": sum(row["error"] is None for row in measured), + } + for field in ("ttft_s", "total_s", "prompt_tps", "decode_tps", "request_tps"): + values = [float(row[field]) for row in measured if row[field] is not None] + result[f"{field}_mean"] = statistics.fmean(values) + result[f"{field}_p50"] = percentile(values, 0.5) + result[f"{field}_p95"] = percentile(values, 0.95) + result["peak_memory_gb"] = max(float(row["peak_memory_gb"]) for row in measured) + return result + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model", required=True) + parser.add_argument("--label", required=True) + parser.add_argument("--warmups", type=int, default=2) + parser.add_argument("--rounds", type=int, default=5) + parser.add_argument("--max-tokens", type=int, default=128) + parser.add_argument("--seed", type=int, default=42) + parser.add_argument( + "--output-dir", type=Path, default=Path(__file__).parent / "direct-results" + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.warmups < 0 or args.rounds < 1 or args.max_tokens < 1: + raise SystemExit("warmups must be >= 0; rounds and max-tokens must be >= 1") + + run_id = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + output_dir = args.output_dir / f"{run_id}-{args.label}" + output_dir.mkdir(parents=True, exist_ok=False) + + load_started = time.perf_counter() + model, tokenizer = load(args.model, trust_remote_code=True) + load_s = time.perf_counter() - load_started + sampler = make_sampler(temp=0.0) + prompt_items = list(PROMPTS.items()) + rows: list[dict[str, Any]] = [] + + for phase, count in (("warmup", args.warmups), ("measured", args.rounds)): + for round_number in range(count): + prompt_name, prompt = prompt_items[round_number % len(prompt_items)] + mx.random.seed(args.seed) + started = time.perf_counter() + first_token_at: float | None = None + final = None + error = None + try: + for response in stream_generate( + model, + tokenizer, + prompt, + max_tokens=args.max_tokens, + sampler=sampler, + ): + if first_token_at is None: + first_token_at = time.perf_counter() + final = response + except Exception as exception: # Preserve evidence before failing the run. + error = f"{type(exception).__name__}: {exception}" + total_s = time.perf_counter() - started + tokens = final.generation_tokens if final is not None else 0 + row = { + "runtime": "mlx-lm-direct", + "model": args.label, + "source": args.model, + "phase": phase, + "round": round_number + 1, + "prompt_name": prompt_name, + "prompt_tokens": final.prompt_tokens if final is not None else None, + "output_tokens": tokens, + "ttft_s": first_token_at - started if first_token_at is not None else None, + "total_s": total_s, + "prompt_tps": final.prompt_tps if final is not None else None, + "decode_tps": final.generation_tps if final is not None else None, + "request_tps": tokens / total_s if tokens else None, + "peak_memory_gb": final.peak_memory if final is not None else None, + "finish_reason": final.finish_reason if final is not None else None, + "error": error, + } + rows.append(row) + print( + f"{phase} {round_number + 1}/{count} prompt={prompt_name} " + f"total={total_s:.3f}s tokens={tokens} " + f"decode={row['decode_tps'] or 0:.2f}tok/s error={error or 'none'}", + flush=True, + ) + + summary = summarize(rows) + metadata = { + "run_id": run_id, + "created_at": datetime.now(UTC).isoformat(), + "host": { + "platform": platform.platform(), + "machine": platform.machine(), + "python": platform.python_version(), + }, + "arguments": vars(args) | {"output_dir": str(args.output_dir)}, + "load_s": load_s, + "mlx_version": getattr(mx, "__version__", None), + "device": str(mx.default_device()), + "prompts": PROMPTS, + "summary": summary, + } + (output_dir / "raw.json").write_text(json.dumps(rows, indent=2), encoding="utf-8") + (output_dir / "metadata.json").write_text( + json.dumps(metadata, indent=2, default=str), encoding="utf-8" + ) + (output_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") + print(json.dumps(summary, indent=2)) + print(f"Results: {output_dir}") + return 1 if any(row["error"] for row in rows if row["phase"] == "measured") else 0 + + +if __name__ == "__main__": + raise SystemExit(main())