christian.meyer/lai_automatisierung
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add native MLX inference benchmarks

Browse Source
This commit is contained in:
Christian Meyer
2026-08-31 16:38:59 +02:00
parent 1097fa7ae2
commit 75b2c8958c
5 changed files with 1097 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
benchmarks/inference/README.md Normal file
View File

@@ -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.

500
benchmarks/inference/benchmark.py Normal file
View File

@@ -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())

364
benchmarks/inference/coding_benchmark.py Normal file
View File

@@ -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())

155
benchmarks/inference/direct_mlx_benchmark.py Normal file
View File

@@ -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())