356 lines
13 KiB
Python
356 lines
13 KiB
Python
"""Versioned performance contract metadata and stub benchmark runner for DGR-001.
|
|
|
|
This module captures the *contract* first: the model family, architecture
|
|
alignment, benchmark lanes, and stop condition that benchmark runs must
|
|
satisfy. It also runs the contract's lanes through a deterministic stub
|
|
backend so the report data shape exists end to end. It never downloads or
|
|
executes a model; real transformers / llama.cpp backends plug in behind the
|
|
same ``run()`` seam later.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
SCHEMA_VERSION = 1
|
|
CONTRACT_ID = "DGR-001"
|
|
DEFAULT_OUTPUT_PATH = Path(".scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ModelTarget:
|
|
"""Architecture-aligned model target for the DGR-001 benchmark contract."""
|
|
|
|
name: str
|
|
architecture: str
|
|
safetensors_repo: str
|
|
safetensors_precision: str
|
|
gguf_repo: str
|
|
gguf_quant: str
|
|
gguf_size_gb: float
|
|
comparison_policy: str
|
|
rationale: str
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"name": self.name,
|
|
"architecture": self.architecture,
|
|
"safetensors_repo": self.safetensors_repo,
|
|
"safetensors_precision": self.safetensors_precision,
|
|
"gguf_repo": self.gguf_repo,
|
|
"gguf_quant": self.gguf_quant,
|
|
"gguf_size_gb": self.gguf_size_gb,
|
|
"comparison_policy": self.comparison_policy,
|
|
"rationale": self.rationale,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BenchmarkLane:
|
|
"""One side of the comparison the contract requires."""
|
|
|
|
id: str
|
|
runtime: str
|
|
device: str
|
|
recipe: str
|
|
concurrency_levels: tuple[int, ...]
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"id": self.id,
|
|
"runtime": self.runtime,
|
|
"device": self.device,
|
|
"recipe": self.recipe,
|
|
"concurrency_levels": list(self.concurrency_levels),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PerformanceContract:
|
|
"""Machine-readable contract for the DGR-001 benchmark story."""
|
|
|
|
schema_version: int
|
|
story_id: str
|
|
model_target: ModelTarget
|
|
benchmark_lanes: tuple[BenchmarkLane, ...]
|
|
metrics: tuple[str, ...]
|
|
stop_condition: str
|
|
notes: tuple[str, ...] = ()
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"schema_version": self.schema_version,
|
|
"story_id": self.story_id,
|
|
"model_target": self.model_target.to_dict(),
|
|
"benchmark_lanes": [lane.to_dict() for lane in self.benchmark_lanes],
|
|
"metrics": list(self.metrics),
|
|
"stop_condition": self.stop_condition,
|
|
"notes": list(self.notes),
|
|
}
|
|
|
|
def write_json(self, path: str | Path) -> Path:
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(self.to_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
DEFAULT_CONTRACT = PerformanceContract(
|
|
schema_version=SCHEMA_VERSION,
|
|
story_id=CONTRACT_ID,
|
|
model_target=ModelTarget(
|
|
name="DeepSeek-V2-Lite-Chat",
|
|
architecture="deepseek2",
|
|
safetensors_repo="deepseek-ai/DeepSeek-V2-Lite-Chat",
|
|
safetensors_precision="bfloat16",
|
|
gguf_repo="second-state/DeepSeek-V2-Lite-Chat-GGUF",
|
|
gguf_quant="Q2_K",
|
|
gguf_size_gb=6.43,
|
|
comparison_policy=(
|
|
"same model/revision, closest practical low-footprint precision pair: "
|
|
"BF16 safetensors versus Q2_K GGUF"
|
|
),
|
|
rationale=(
|
|
"Smallest DeepSeek-family benchmark anchor that still points toward "
|
|
"DeepSeek-V4-Flash; keeps the runtime on the DeepSeek2 path instead "
|
|
"of falling back to a tiny but architecture-mismatched smoke model."
|
|
),
|
|
),
|
|
benchmark_lanes=(
|
|
BenchmarkLane(
|
|
id="transformers-safetensors-cpu",
|
|
runtime="transformers",
|
|
device="cpu",
|
|
recipe="current safetensors recipe",
|
|
concurrency_levels=(1, 4),
|
|
),
|
|
BenchmarkLane(
|
|
id="llama-cpp-gguf-cpu",
|
|
runtime="llama.cpp",
|
|
device="cpu",
|
|
recipe="whole-model GGUF recipe",
|
|
concurrency_levels=(1, 4),
|
|
),
|
|
BenchmarkLane(
|
|
id="transformers-safetensors-gpu",
|
|
runtime="transformers",
|
|
device="gpu",
|
|
recipe="current safetensors recipe",
|
|
concurrency_levels=(1, 4),
|
|
),
|
|
BenchmarkLane(
|
|
id="llama-cpp-gguf-gpu",
|
|
runtime="llama.cpp",
|
|
device="gpu",
|
|
recipe="whole-model GGUF recipe",
|
|
concurrency_levels=(1, 4),
|
|
),
|
|
),
|
|
metrics=(
|
|
"ttft_ms",
|
|
"prefill_tok_per_sec",
|
|
"decode_tok_per_sec",
|
|
"p50_latency_ms",
|
|
"p95_latency_ms",
|
|
"aggregate_throughput_tok_per_sec",
|
|
"rss_bytes",
|
|
"vram_bytes",
|
|
"artifact_bytes",
|
|
"failure_count",
|
|
"output_drift",
|
|
),
|
|
stop_condition=(
|
|
"Stop if GGUF does not provide a meaningful speed or fit benefit over the "
|
|
"safetensors baseline for the chosen DeepSeek-family model target."
|
|
),
|
|
notes=(
|
|
"Real model execution stays opt-in and must keep model artifacts on the mounted drive.",
|
|
"Use the tiny fallback only for loader plumbing smoke tests; it does not replace the architecture-aligned baseline.",
|
|
),
|
|
)
|
|
|
|
|
|
def build_default_contract() -> PerformanceContract:
|
|
return DEFAULT_CONTRACT
|
|
|
|
|
|
BENCHMARK_SCHEMA_VERSION = 1
|
|
STUB_OUTPUT_TOKENS = ("mesh", "activation", "seam", "baseline")
|
|
# DeepSeek-V2-Lite is ~15.7B params at 2 bytes each; metadata only, nothing downloaded.
|
|
_SAFETENSORS_BF16_ARTIFACT_GB = 31.4
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LaneSample:
|
|
"""Raw single-stream measurements one backend produces for a lane."""
|
|
|
|
ttft_ms: float
|
|
prefill_tok_per_sec: float
|
|
decode_tok_per_sec: float
|
|
rss_bytes: int
|
|
vram_bytes: int
|
|
artifact_bytes: int
|
|
output_tokens: tuple[str, ...]
|
|
failure_count: int = 0
|
|
|
|
|
|
def _gb(value: float) -> int:
|
|
return int(value * 1024**3)
|
|
|
|
|
|
class StubLaneBackend:
|
|
"""Deterministic placeholder measurements until real lane execution lands.
|
|
|
|
The numbers are synthetic but directionally shaped — the Q2_K GGUF loads a
|
|
far smaller artifact and decodes faster than BF16 safetensors — so the
|
|
comparison and stop-condition plumbing can be exercised in CI.
|
|
"""
|
|
|
|
source = "stub-backend"
|
|
|
|
# (runtime, device) -> (ttft_ms, prefill tok/s, decode tok/s, rss GB, vram GB)
|
|
_PROFILES = {
|
|
("transformers", "cpu"): (1800.0, 45.0, 6.0, 33.0, 0.0),
|
|
("llama.cpp", "cpu"): (950.0, 90.0, 14.0, 7.1, 0.0),
|
|
("transformers", "gpu"): (420.0, 850.0, 34.0, 4.0, 33.0),
|
|
("llama.cpp", "gpu"): (260.0, 640.0, 52.0, 1.5, 7.5),
|
|
}
|
|
|
|
def __init__(self, contract: PerformanceContract) -> None:
|
|
self._contract = contract
|
|
|
|
def run(self, lane: BenchmarkLane) -> LaneSample:
|
|
ttft_ms, prefill, decode, rss_gb, vram_gb = self._PROFILES[(lane.runtime, lane.device)]
|
|
artifact_gb = (
|
|
self._contract.model_target.gguf_size_gb
|
|
if lane.runtime == "llama.cpp"
|
|
else _SAFETENSORS_BF16_ARTIFACT_GB
|
|
)
|
|
return LaneSample(
|
|
ttft_ms=ttft_ms,
|
|
prefill_tok_per_sec=prefill,
|
|
decode_tok_per_sec=decode,
|
|
rss_bytes=_gb(rss_gb),
|
|
vram_bytes=_gb(vram_gb),
|
|
artifact_bytes=_gb(artifact_gb),
|
|
output_tokens=STUB_OUTPUT_TOKENS,
|
|
)
|
|
|
|
|
|
def _output_drift(tokens: tuple[str, ...], reference: tuple[str, ...]) -> float:
|
|
"""Fraction of positions where a lane's output diverges from its reference."""
|
|
length = max(len(tokens), len(reference))
|
|
if length == 0:
|
|
return 0.0
|
|
mismatches = sum(a != b for a, b in zip(tokens, reference)) + abs(len(tokens) - len(reference))
|
|
return round(mismatches / length, 4)
|
|
|
|
|
|
def _metrics_for(sample: LaneSample, concurrency: int, output_drift: float) -> dict:
|
|
# Stub concurrency model: batching scales throughput at 85% efficiency and
|
|
# stretches per-request token latency and TTFT accordingly.
|
|
efficiency = 1.0 if concurrency == 1 else 0.85
|
|
p50_latency_ms = round(1000.0 / (sample.decode_tok_per_sec * efficiency), 4)
|
|
return {
|
|
"ttft_ms": round(sample.ttft_ms * (1 + 0.1 * (concurrency - 1)), 4),
|
|
"prefill_tok_per_sec": round(sample.prefill_tok_per_sec * efficiency, 4),
|
|
"decode_tok_per_sec": round(sample.decode_tok_per_sec * efficiency, 4),
|
|
"p50_latency_ms": p50_latency_ms,
|
|
"p95_latency_ms": round(p50_latency_ms * 1.25, 4),
|
|
"aggregate_throughput_tok_per_sec": round(sample.decode_tok_per_sec * concurrency * efficiency, 4),
|
|
"rss_bytes": sample.rss_bytes,
|
|
"vram_bytes": sample.vram_bytes,
|
|
"artifact_bytes": sample.artifact_bytes,
|
|
"failure_count": sample.failure_count,
|
|
"output_drift": output_drift,
|
|
}
|
|
|
|
|
|
def _compare_device(lanes: list[tuple[BenchmarkLane, LaneSample]], device: str) -> dict:
|
|
by_runtime = {lane.runtime: (lane, sample) for lane, sample in lanes if lane.device == device}
|
|
safetensors_lane, safetensors = by_runtime["transformers"]
|
|
gguf_lane, gguf = by_runtime["llama.cpp"]
|
|
memory_metric = "vram_bytes" if device == "gpu" else "rss_bytes"
|
|
decode_speedup = round(gguf.decode_tok_per_sec / safetensors.decode_tok_per_sec, 4)
|
|
artifact_bytes_ratio = round(gguf.artifact_bytes / max(1, safetensors.artifact_bytes), 4)
|
|
return {
|
|
"safetensors_lane": safetensors_lane.id,
|
|
"gguf_lane": gguf_lane.id,
|
|
"decode_speedup": decode_speedup,
|
|
"ttft_speedup": round(safetensors.ttft_ms / max(0.001, gguf.ttft_ms), 4),
|
|
"artifact_bytes_ratio": artifact_bytes_ratio,
|
|
"memory_metric": memory_metric,
|
|
"memory_bytes_ratio": round(
|
|
getattr(gguf, memory_metric) / max(1, getattr(safetensors, memory_metric)), 4
|
|
),
|
|
"output_drift": _output_drift(gguf.output_tokens, safetensors.output_tokens),
|
|
"gguf_benefit": decode_speedup >= 1.10 or artifact_bytes_ratio <= 0.5,
|
|
}
|
|
|
|
|
|
def run_performance_benchmark(
|
|
contract: PerformanceContract = DEFAULT_CONTRACT,
|
|
backend: StubLaneBackend | None = None,
|
|
) -> dict:
|
|
"""Run every contract lane through a backend and compare GGUF to safetensors."""
|
|
backend = backend if backend is not None else StubLaneBackend(contract)
|
|
lanes = [(lane, backend.run(lane)) for lane in contract.benchmark_lanes]
|
|
references = {
|
|
lane.device: sample.output_tokens for lane, sample in lanes if lane.runtime == "transformers"
|
|
}
|
|
lane_reports = []
|
|
for lane, sample in lanes:
|
|
drift = _output_drift(sample.output_tokens, references.get(lane.device, sample.output_tokens))
|
|
lane_reports.append({
|
|
**lane.to_dict(),
|
|
"output_tokens": list(sample.output_tokens),
|
|
"results": [
|
|
{"concurrency": level, "metrics": _metrics_for(sample, level, drift)}
|
|
for level in lane.concurrency_levels
|
|
],
|
|
})
|
|
devices = sorted({lane.device for lane, _ in lanes})
|
|
comparisons = {device: _compare_device(lanes, device) for device in devices}
|
|
gguf_benefit = any(comparison["gguf_benefit"] for comparison in comparisons.values())
|
|
return {
|
|
"schema_version": BENCHMARK_SCHEMA_VERSION,
|
|
"story_id": contract.story_id,
|
|
"source": getattr(backend, "source", "custom-backend"),
|
|
"model_target": contract.model_target.to_dict(),
|
|
"lanes": lane_reports,
|
|
"comparisons": comparisons,
|
|
"stop_condition": {
|
|
"text": contract.stop_condition,
|
|
"gguf_benefit": gguf_benefit,
|
|
"triggered": not gguf_benefit,
|
|
},
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Write the DGR-001 performance contract JSON")
|
|
parser.add_argument("--json-out", type=Path, default=DEFAULT_OUTPUT_PATH, help="output JSON path")
|
|
parser.add_argument(
|
|
"--benchmark-out",
|
|
type=Path,
|
|
default=None,
|
|
help="also run the deterministic stub benchmark and write its JSON report here",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
contract = build_default_contract()
|
|
path = contract.write_json(args.json_out)
|
|
print(path)
|
|
if args.benchmark_out is not None:
|
|
report = run_performance_benchmark(contract)
|
|
args.benchmark_out.parent.mkdir(parents=True, exist_ok=True)
|
|
args.benchmark_out.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
print(args.benchmark_out)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover - CLI entry point
|
|
raise SystemExit(main())
|