Skip to content

Quality API

Quality and timing metrics for compressed attention validation.

KLD (Kullback-Leibler Divergence) quality metric for attention distributions.

Measures the KL divergence between compressed and reference attention distributions. Lower KLD indicates better preservation of attention patterns.

Score mapping: KLD_score = 100 * exp(-mean_kld)

so 0 nats → 100, 0.7 nats → 50, 1.7 nats → ~18.

KLDResult dataclass 

KLDResult(score: float, mean_kld: float, ppl: Optional[float] = None, rms_dp_pct: Optional[float] = None, same_topp_pct: Optional[float] = None, is_self_reference: bool = False)

Result of KLD quality evaluation.

KLDTracker 

KLDTracker()

Online KLD tracker for streaming evaluation.

Source code in turboquant/quality/kld.py 
157
158
159
160
def __init__(self):
    self.kld_sum: float = 0.0
    self.count: int = 0
    self.top1_matches: int = 0

update 

update(candidate_scores: Tensor, reference_scores: Tensor) -> None

Add a batch of attention score pairs.

Source code in turboquant/quality/kld.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def update(
    self,
    candidate_scores: torch.Tensor,
    reference_scores: torch.Tensor,
) -> None:
    """Add a batch of attention score pairs."""
    if candidate_scores.shape != reference_scores.shape:
        raise ValueError("shape mismatch")

    kld = compute_attention_kld(candidate_scores, reference_scores, reduction="sum")
    n = candidate_scores.numel() // candidate_scores.shape[-1]

    self.kld_sum += kld
    self.count += n

    # Track top-1 matches
    cand_top1 = candidate_scores.argmax(dim=-1)
    ref_top1 = reference_scores.argmax(dim=-1)
    self.top1_matches += int((cand_top1 == ref_top1).sum().item())

result 

result() -> KLDResult

Compute final KLD result.

Source code in turboquant/quality/kld.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
def result(self) -> KLDResult:
    """Compute final KLD result."""
    if self.count == 0:
        return KLDResult(score=100.0, mean_kld=0.0)

    mean_kld = self.kld_sum / self.count
    same_topp_pct = (self.top1_matches / self.count) * 100.0 if self.count > 0 else 100.0

    return KLDResult(
        score=kld_to_score(mean_kld),
        mean_kld=mean_kld,
        same_topp_pct=same_topp_pct,
    )

reset 

reset() -> None

Reset tracker state.

Source code in turboquant/quality/kld.py 
196
197
198
199
200
def reset(self) -> None:
    """Reset tracker state."""
    self.kld_sum = 0.0
    self.count = 0
    self.top1_matches = 0

kld_to_score 

kld_to_score(kld: float) -> float

Map mean KLD (nats) → 0–100 with score = 100 * exp(-kld).

Source code in turboquant/quality/kld.py 
33
34
35
36
37
def kld_to_score(kld: float) -> float:
    """Map mean KLD (nats) → 0–100 with score = 100 * exp(-kld)."""
    if kld < 0:
        kld = 0.0
    return 100.0 * math.exp(-kld)

score_to_kld 

score_to_kld(score: float) -> float

Inverse mapping: score → mean KLD (nats).

Source code in turboquant/quality/kld.py 
40
41
42
43
44
45
46
def score_to_kld(score: float) -> float:
    """Inverse mapping: score → mean KLD (nats)."""
    if score <= 0:
        return float("inf")
    if score >= 100:
        return 0.0
    return -math.log(score / 100.0)

compute_attention_kld 

compute_attention_kld(candidate_scores: Tensor, reference_scores: Tensor, reduction: str = 'mean') -> float

Compute KL divergence between attention score distributions.

Args: candidate_scores: Attention logits from compressed cache [, seq_len]. reference_scores: Attention logits from reference cache [, seq_len]. reduction: "mean", "sum", or "none".

Returns: KLD in nats. Lower is better.

Source code in turboquant/quality/kld.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def compute_attention_kld(
    candidate_scores: torch.Tensor,
    reference_scores: torch.Tensor,
    reduction: str = "mean",
) -> float:
    """Compute KL divergence between attention score distributions.

    Args:
        candidate_scores: Attention logits from compressed cache [*, seq_len].
        reference_scores: Attention logits from reference cache [*, seq_len].
        reduction: "mean", "sum", or "none".

    Returns:
        KLD in nats. Lower is better.
    """
    if candidate_scores.shape != reference_scores.shape:
        raise ValueError(f"shape mismatch: {candidate_scores.shape} vs {reference_scores.shape}")

    log_cand = F.log_softmax(candidate_scores.float(), dim=-1)
    ref = F.softmax(reference_scores.float(), dim=-1)

    kld = F.kl_div(log_cand, ref, reduction=reduction)
    return max(0.0, float(kld.item() if kld.numel() == 1 else kld.mean().item()))

compute_trajectory_kld 

compute_trajectory_kld(candidate_logits: Tensor, reference_logits: Tensor) -> KLDResult

Compute per-step KLD across a token trajectory.

For each token position, measures KLD between the candidate and reference next-token distributions.

Args: candidate_logits: [seq_len, vocab_size] or [batch, seq_len, vocab_size] reference_logits: Same shape as candidate.

Returns: KLDResult with mean KLD and derived score.

Source code in turboquant/quality/kld.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def compute_trajectory_kld(
    candidate_logits: torch.Tensor,
    reference_logits: torch.Tensor,
) -> KLDResult:
    """Compute per-step KLD across a token trajectory.

    For each token position, measures KLD between the candidate and
    reference next-token distributions.

    Args:
        candidate_logits: [seq_len, vocab_size] or [batch, seq_len, vocab_size]
        reference_logits: Same shape as candidate.

    Returns:
        KLDResult with mean KLD and derived score.
    """
    if candidate_logits.shape != reference_logits.shape:
        raise ValueError(f"shape mismatch: {candidate_logits.shape} vs {reference_logits.shape}")

    if candidate_logits.ndim == 2:
        candidate_logits = candidate_logits.unsqueeze(0)
        reference_logits = reference_logits.unsqueeze(0)

    # Flatten to [N, vocab]
    batch, seq_len, vocab = candidate_logits.shape
    cand_flat = candidate_logits.reshape(-1, vocab).float()
    ref_flat = reference_logits.reshape(-1, vocab).float()

    # Per-position KLD
    log_cand = F.log_softmax(cand_flat, dim=-1)
    ref_probs = F.softmax(ref_flat, dim=-1)

    kld_per_pos = F.kl_div(log_cand, ref_probs, reduction="none").sum(dim=-1)
    mean_kld = float(kld_per_pos.mean().item())

    # Same-top-p metric: what fraction of positions have same top-1 token
    cand_top1 = cand_flat.argmax(dim=-1)
    ref_top1 = ref_flat.argmax(dim=-1)
    same_topp = float((cand_top1 == ref_top1).float().mean().item()) * 100.0

    return KLDResult(
        score=kld_to_score(mean_kld),
        mean_kld=mean_kld,
        same_topp_pct=same_topp,
        is_self_reference=False,
    )

compute_self_reference_kld 

compute_self_reference_kld(logits: Tensor, perturbation: float = 1e-06) -> KLDResult

Compute KLD with self as reference (should be ~0).

Useful for validating that the metric pipeline works correctly.

Source code in turboquant/quality/kld.py 
122
123
124
125
126
127
128
129
130
131
def compute_self_reference_kld(
    logits: torch.Tensor,
    perturbation: float = 1e-6,
) -> KLDResult:
    """Compute KLD with self as reference (should be ~0).

    Useful for validating that the metric pipeline works correctly.
    """
    perturbed = logits + torch.randn_like(logits) * perturbation
    return compute_trajectory_kld(perturbed, logits)

compute_rms_delta_probability 

compute_rms_delta_probability(candidate_logits: Tensor, reference_logits: Tensor) -> float

Compute RMS of probability deltas (percentage).

Measures the root-mean-square difference in probabilities between candidate and reference distributions.

Returns: RMS delta probability as a percentage (0-100).

Source code in turboquant/quality/kld.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def compute_rms_delta_probability(
    candidate_logits: torch.Tensor,
    reference_logits: torch.Tensor,
) -> float:
    """Compute RMS of probability deltas (percentage).

    Measures the root-mean-square difference in probabilities
    between candidate and reference distributions.

    Returns:
        RMS delta probability as a percentage (0-100).
    """
    cand_probs = F.softmax(candidate_logits.float(), dim=-1)
    ref_probs = F.softmax(reference_logits.float(), dim=-1)

    delta = (cand_probs - ref_probs).pow(2)
    rms = delta.mean().sqrt()
    return float(rms.item()) * 100.0

Prompt-set loading helpers for reference/candidate evaluations.

Needle-style retrieval prompt helpers.

Token-trajectory and perturbation metrics for generation checks.

sequence_match 

sequence_match(reference: Sequence[int], candidate: Sequence[int]) -> SequenceMatch

Compare two token-id sequences in model-token units.

Source code in turboquant/quality/trajectory.py 
70
71
72
73
74
75
76
77
78
def sequence_match(reference: Sequence[int], candidate: Sequence[int]) -> SequenceMatch:
    """Compare two token-id sequences in model-token units."""
    n = min(len(reference), len(candidate))
    for idx in range(n):
        if reference[idx] != candidate[idx]:
            return SequenceMatch(idx, idx, len(reference), len(candidate), False)
    if len(reference) == len(candidate):
        return SequenceMatch(None, n, len(reference), len(candidate), True)
    return SequenceMatch(n, n, len(reference), len(candidate), False)

trajectory_metrics 

trajectory_metrics(references: Sequence[Sequence[int]], candidates: Sequence[Sequence[int]], *, expected_tokens: int | None = None) -> TrajectoryMetrics

Compute prefix trajectory agreement over paired token-id sequences.

Source code in turboquant/quality/trajectory.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def trajectory_metrics(
    references: Sequence[Sequence[int]],
    candidates: Sequence[Sequence[int]],
    *,
    expected_tokens: int | None = None,
) -> TrajectoryMetrics:
    """Compute prefix trajectory agreement over paired token-id sequences."""
    if len(references) != len(candidates):
        raise ValueError("references and candidates must have the same length")
    if not references:
        raise ValueError("at least one sequence pair is required")

    rows = [sequence_match(ref, cand) for ref, cand in zip(references, candidates)]
    mean_prefix = statistics.mean(row.prefix_agreement_length for row in rows)
    mean_candidate = statistics.mean(row.candidate_length for row in rows)
    mean_reference = statistics.mean(row.reference_length for row in rows)
    divergences = [row.first_divergence for row in rows if row.first_divergence is not None]
    score = 100.0 * mean_prefix / mean_candidate if mean_candidate > 0 else 0.0
    score = min(100.0, max(0.0, score))
    notes: list[str] = []
    if expected_tokens is not None:
        short = sum(1 for row in rows if row.candidate_length < expected_tokens)
        if short:
            notes.append(f"{short}/{len(rows)} candidates ended before {expected_tokens} generated tokens")
    return TrajectoryMetrics(
        score=score,
        full_match_rate=sum(1 for row in rows if row.matched) / len(rows),
        median_first_divergence=statistics.median(divergences) if divergences else None,
        mean_prefix_agreement_length=mean_prefix,
        mean_candidate_length=mean_candidate,
        mean_reference_length=mean_reference,
        per_sequence=rows,
        notes=notes,
    )

levenshtein 

levenshtein(a: Sequence[int], b: Sequence[int]) -> int

Levenshtein distance over token ids.

Source code in turboquant/quality/trajectory.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def levenshtein(a: Sequence[int], b: Sequence[int]) -> int:
    """Levenshtein distance over token ids."""
    if not a:
        return len(b)
    if not b:
        return len(a)
    prev = list(range(len(b) + 1))
    for idx_a, val_a in enumerate(a, 1):
        cur = [idx_a] + [0] * len(b)
        for idx_b, val_b in enumerate(b, 1):
            cost = 0 if val_a == val_b else 1
            cur[idx_b] = min(prev[idx_b] + 1, cur[idx_b - 1] + 1, prev[idx_b - 1] + cost)
        prev = cur
    return prev[-1]

normalized_drift 

normalized_drift(anchor: Sequence[int], perturbed: Sequence[int]) -> float

Token edit distance divided by anchor length, clipped to [0, 1].

Source code in turboquant/quality/trajectory.py 
133
134
135
136
137
138
139
def normalized_drift(anchor: Sequence[int], perturbed: Sequence[int]) -> float:
    """Token edit distance divided by anchor length, clipped to [0, 1]."""
    if not anchor and not perturbed:
        return 0.0
    if not anchor:
        return 1.0
    return min(1.0, levenshtein(anchor, perturbed) / len(anchor))

eligible_words 

eligible_words(prompt: str) -> list[tuple[int, int, str]]

Return non-stopword word spans from a prompt.

Source code in turboquant/quality/trajectory.py 
170
171
172
173
174
175
176
177
def eligible_words(prompt: str) -> list[tuple[int, int, str]]:
    """Return non-stopword word spans from a prompt."""
    rows: list[tuple[int, int, str]] = []
    for match in _WORD_RE.finditer(prompt):
        word = match.group(0)
        if word.lower() not in _STOPWORDS:
            rows.append((match.start(), match.end(), word))
    return rows

perturb_prompt 

perturb_prompt(prompt: str, kind: str, *, seed: int = 42) -> str | None

Apply one deterministic local prompt perturbation.

Source code in turboquant/quality/trajectory.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def perturb_prompt(prompt: str, kind: str, *, seed: int = 42) -> str | None:
    """Apply one deterministic local prompt perturbation."""
    rng = random.Random(seed)
    if kind == "typo":
        rows = [(s, e, w) for s, e, w in eligible_words(prompt) if len(w) >= 4]
        if not rows:
            return None
        start, end, word = rng.choice(rows)
        pos = rng.randrange(0, len(word) - 1)
        swapped = word[:pos] + word[pos + 1] + word[pos] + word[pos + 2:]
        if swapped == word:
            return None
        return prompt[:start] + swapped + prompt[end:]
    if kind == "case":
        chunks: list[str] = []
        cursor = 0
        changed = False
        for match in _WORD_RE.finditer(prompt):
            chunks.append(prompt[cursor:match.start()])
            word = match.group(0)
            if word[0].isupper():
                chunks.append(word[0].lower() + word[1:])
                changed = True
            else:
                chunks.append(word)
            cursor = match.end()
        chunks.append(prompt[cursor:])
        return "".join(chunks) if changed else None
    if kind == "punct":
        stripped = prompt.rstrip()
        tail = prompt[len(stripped):]
        if stripped.endswith("?") or stripped.endswith("."):
            return stripped[:-1] + tail
        return stripped + "?" + tail
    if kind == "paraphrase":
        rows = [
            (s, e, w) for s, e, w in eligible_words(prompt)
            if len(w) >= 3 and w.lower() in _SYNONYMS
        ]
        if not rows:
            return None
        start, end, word = rng.choice(rows)
        sub = _SYNONYMS[word.lower()]
        if word[0].isupper():
            sub = sub[0].upper() + sub[1:]
        return prompt[:start] + sub + prompt[end:]
    raise ValueError(f"unknown perturbation kind: {kind}")

perturbation_metrics 

perturbation_metrics(*, prompt_ids: Sequence[str], prompts: Sequence[str], reference_anchor: Sequence[Sequence[int]], candidate_anchor: Sequence[Sequence[int]], reference_perturbed: dict[tuple[str, str], Sequence[int]], candidate_perturbed: dict[tuple[str, str], Sequence[int]], perturbations: Sequence[str] = ('typo', 'case', 'punct', 'paraphrase'), alpha: float = 5.0, seed: int = 42) -> PerturbationMetrics

Score excess token drift under small prompt changes.

Source code in turboquant/quality/trajectory.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def perturbation_metrics(
    *,
    prompt_ids: Sequence[str],
    prompts: Sequence[str],
    reference_anchor: Sequence[Sequence[int]],
    candidate_anchor: Sequence[Sequence[int]],
    reference_perturbed: dict[tuple[str, str], Sequence[int]],
    candidate_perturbed: dict[tuple[str, str], Sequence[int]],
    perturbations: Sequence[str] = ("typo", "case", "punct", "paraphrase"),
    alpha: float = 5.0,
    seed: int = 42,
) -> PerturbationMetrics:
    """Score excess token drift under small prompt changes."""
    if not (len(prompt_ids) == len(prompts) == len(reference_anchor) == len(candidate_anchor)):
        raise ValueError("prompt ids, prompts, and anchor outputs must have matching lengths")
    records: list[PerturbationRecord] = []
    per_kind: dict[str, list[float]] = {kind: [] for kind in perturbations}
    skipped = 0
    for idx, prompt_id in enumerate(prompt_ids):
        prompt = prompts[idx]
        for kind in perturbations:
            perturbed_prompt = perturb_prompt(prompt, kind, seed=seed + idx)
            if perturbed_prompt is None:
                skipped += 1
                continue
            key = (prompt_id, kind)
            if key not in reference_perturbed or key not in candidate_perturbed:
                skipped += 1
                continue
            ref_drift = normalized_drift(reference_anchor[idx], reference_perturbed[key])
            cand_drift = normalized_drift(candidate_anchor[idx], candidate_perturbed[key])
            excess = max(0.0, cand_drift - ref_drift)
            score = 100.0 * math.exp(-alpha * excess)
            records.append(
                PerturbationRecord(
                    prompt_id=prompt_id,
                    perturbation=kind,
                    perturbed_prompt=perturbed_prompt,
                    reference_drift=ref_drift,
                    candidate_drift=cand_drift,
                    excess_drift=excess,
                    score=score,
                )
            )
            per_kind[kind].append(score)
    if not records:
        raise ValueError("no perturbation records were scored")
    per_summary = {
        kind: (statistics.mean(scores) if scores else float("nan"))
        for kind, scores in per_kind.items()
    }
    notes = [f"{skipped} perturbation cells were skipped"] if skipped else []
    return PerturbationMetrics(
        score=statistics.mean(row.score for row in records),
        per_perturbation_score=per_summary,
        records=records,
        skipped=skipped,
        notes=notes,
    )

harmonic_mean 

harmonic_mean(values: Sequence[float]) -> float

Harmonic mean clipped to [0, 100].

Source code in turboquant/quality/trajectory.py 
290
291
292
293
294
295
def harmonic_mean(values: Sequence[float]) -> float:
    """Harmonic mean clipped to [0, 100]."""
    clean = [max(0.0, float(value)) for value in values]
    if not clean or any(value <= 0.0 for value in clean):
        return 0.0
    return min(100.0, max(0.0, len(clean) / sum(1.0 / value for value in clean)))

score_band 

score_band(score: float) -> str

Map a 0-100 score to a coarse report band.

Source code in turboquant/quality/trajectory.py 
298
299
300
301
302
303
304
305
306
def score_band(score: float) -> str:
    """Map a 0-100 score to a coarse report band."""
    if score >= 90.0:
        return "match"
    if score >= 80.0:
        return "minor_drift"
    if score >= 60.0:
        return "drift"
    return "mismatch"

Composite quality scoring for reference/candidate checks.

Hardware replay system for TurboQuant diagnostic profiles.

Parses diagnostic output from turbo-hardware-diag.sh, builds structured hardware profiles, and enables comparison/replay across different hardware configurations.

Usage: # Parse a diagnostic output file profile = HardwareProfile.from_diag_file("turbo-diag-20260326.txt")

# Compare two profiles
report = compare_profiles(baseline, target)

# Predict performance for a hardware config
predicted = predict_decode_from_baseline(profile, target_gpu_family_id=1007)

GPUInfo dataclass 

GPUInfo(name: str = 'unknown', family: str = 'unknown', family_id: int = 0, has_tensor: bool = False, has_unified_memory: bool = False, has_bfloat: bool = False, recommended_max_working_set_mb: float = 0.0, metal_version: str = 'unknown', cuda_compute_cap: str = '', cuda_vram_mb: float = 0.0)

GPU hardware capabilities.

SystemInfo dataclass 

SystemInfo(platform: str = 'unknown', os_version: str = 'unknown', arch: str = 'unknown', cpu_brand: str = 'unknown', cpu_cores_physical: int = 0, cpu_cores_logical: int = 0, ram_total_gb: int = 0, apple_silicon: bool = False, chip_model: str = '', l1_dcache: int = 0, l2_cache: int = 0, gpu: GPUInfo = GPUInfo())

System hardware specs (no PII).

BenchResult dataclass 

BenchResult(label: str, cache_type_k: str, cache_type_v: str, context_depth: int = 0, mode: str = '', tok_per_sec: float = 0.0, stddev: float = 0.0, wall_ms: int = 0, env: str = '')

Single benchmark measurement.

LoadSnapshot dataclass 

LoadSnapshot(label: str, timestamp: str = '', load_avg: str = '', free_ram_mb: float = 0.0, swap_used: str = '', process_count: int = 0, thermal: str = '', gpu_util: str = '')

System load at a point in time.

ModelInfo dataclass 

ModelInfo(filename: str = '', filesize_bytes: int = 0, architecture: str = '', name: str = '', file_type: str = '', model_type: str = '', params: str = '', n_layer: int = 0, n_head: int = 0, n_head_kv: int = 0, n_expert: int = 0, n_expert_used: int = 0, n_ctx_train: int = 0, n_embd: int = 0)

Model metadata.

PPLResult dataclass 

PPLResult(cache_type: str, chunks: int, ppl: float, stddev: float, env: str = '')

Perplexity measurement.

HardwareProfile dataclass 

HardwareProfile(diag_version: int = 0, timestamp: str = '', system: SystemInfo = SystemInfo(), model: ModelInfo = ModelInfo(), benchmarks: list[BenchResult] = list(), ppl_results: list[PPLResult] = list(), load_snapshots: list[LoadSnapshot] = list(), build_commit: str = '')

Complete hardware profile from a diagnostic run.

to_json 

to_json() -> str

Serialize to JSON for storage/replay.

Source code in turboquant/quality/hw_replay.py 
128
129
130
def to_json(self) -> str:
    """Serialize to JSON for storage/replay."""
    return json.dumps(asdict(self), indent=2)

save 

save(path: str | Path) -> None

Save profile to JSON file.

Source code in turboquant/quality/hw_replay.py 
132
133
134
def save(self, path: str | Path) -> None:
    """Save profile to JSON file."""
    Path(path).write_text(self.to_json())

from_json classmethod 

from_json(path: str | Path) -> HardwareProfile

Load profile from JSON file.

Source code in turboquant/quality/hw_replay.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@classmethod
def from_json(cls, path: str | Path) -> HardwareProfile:
    """Load profile from JSON file."""
    data = json.loads(Path(path).read_text())
    profile = cls()
    profile.diag_version = data.get("diag_version", 0)
    profile.timestamp = data.get("timestamp", "")
    profile.build_commit = data.get("build_commit", "")

    # System info
    si = data.get("system", {})
    profile.system = SystemInfo(
        platform=si.get("platform", "unknown"),
        os_version=si.get("os_version", "unknown"),
        arch=si.get("arch", "unknown"),
        cpu_brand=si.get("cpu_brand", "unknown"),
        cpu_cores_physical=si.get("cpu_cores_physical", 0),
        cpu_cores_logical=si.get("cpu_cores_logical", 0),
        ram_total_gb=si.get("ram_total_gb", 0),
        apple_silicon=si.get("apple_silicon", False),
        chip_model=si.get("chip_model", ""),
        l1_dcache=si.get("l1_dcache", 0),
        l2_cache=si.get("l2_cache", 0),
        gpu=GPUInfo(**si.get("gpu", {})),
    )

    # Model info
    mi = data.get("model", {})
    profile.model = ModelInfo(**mi)

    # Benchmarks
    for b in data.get("benchmarks", []):
        profile.benchmarks.append(BenchResult(**b))

    # PPL
    for p in data.get("ppl_results", []):
        profile.ppl_results.append(PPLResult(**p))

    # Load snapshots
    for ls in data.get("load_snapshots", []):
        profile.load_snapshots.append(LoadSnapshot(**ls))

    return profile

from_diag_file classmethod 

from_diag_file(path: str | Path) -> HardwareProfile

Parse a turbo-hardware-diag.sh output file into a profile.

Source code in turboquant/quality/hw_replay.py 
180
181
182
183
184
@classmethod
def from_diag_file(cls, path: str | Path) -> HardwareProfile:
    """Parse a turbo-hardware-diag.sh output file into a profile."""
    text = Path(path).read_text()
    return parse_diag_output(text)

get_decode_curve 

get_decode_curve(cache_type: str = 'turbo3', env: str = '') -> dict[int, float]

Extract decode speed vs context depth curve.

Source code in turboquant/quality/hw_replay.py 
186
187
188
189
190
191
192
def get_decode_curve(self, cache_type: str = "turbo3", env: str = "") -> dict[int, float]:
    """Extract decode speed vs context depth curve."""
    curve = {}
    for b in self.benchmarks:
        if b.mode == "decode" and b.cache_type_k == cache_type and b.env == env:
            curve[b.context_depth] = b.tok_per_sec
    return dict(sorted(curve.items()))

get_prefill_curve 

get_prefill_curve(cache_type: str = 'turbo3', env: str = '') -> dict[int, float]

Extract prefill speed vs context depth curve.

Source code in turboquant/quality/hw_replay.py 
194
195
196
197
198
199
200
def get_prefill_curve(self, cache_type: str = "turbo3", env: str = "") -> dict[int, float]:
    """Extract prefill speed vs context depth curve."""
    curve = {}
    for b in self.benchmarks:
        if b.mode == "prefill" and b.cache_type_k == cache_type and b.env == env:
            curve[b.context_depth] = b.tok_per_sec
    return dict(sorted(curve.items()))

get_ratio_curve 

get_ratio_curve(cache_type: str = 'turbo3', baseline: str = 'q8_0', mode: str = 'decode', env: str = '') -> dict[int, float]

Compute turbo3/q8_0 ratio at each context depth.

Source code in turboquant/quality/hw_replay.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def get_ratio_curve(
    self, cache_type: str = "turbo3",
    baseline: str = "q8_0", mode: str = "decode",
    env: str = ""
) -> dict[int, float]:
    """Compute turbo3/q8_0 ratio at each context depth."""
    target = {}
    base = {}
    for b in self.benchmarks:
        if b.mode != mode:
            continue
        if b.cache_type_k == cache_type and b.env == env:
            target[b.context_depth] = b.tok_per_sec
        elif b.cache_type_k == baseline and b.env == "":
            base[b.context_depth] = b.tok_per_sec

    ratios = {}
    for depth in sorted(set(target.keys()) & set(base.keys())):
        if base[depth] > 0:
            ratios[depth] = target[depth] / base[depth]
    return ratios

find_decode_inflection 

find_decode_inflection(cache_type: str = 'turbo3') -> Optional[int]

Find context depth where decode ratio drops most steeply.

Returns the depth where the gradient (ratio change per depth doubling) is most negative, indicating constant cache thrashing onset. Returns None if insufficient data.

Source code in turboquant/quality/hw_replay.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def find_decode_inflection(self, cache_type: str = "turbo3") -> Optional[int]:
    """Find context depth where decode ratio drops most steeply.

    Returns the depth where the gradient (ratio change per depth doubling)
    is most negative, indicating constant cache thrashing onset.
    Returns None if insufficient data.
    """
    ratios = self.get_ratio_curve(cache_type, "q8_0", "decode")
    if len(ratios) < 3:
        return None

    depths = sorted(ratios.keys())
    worst_gradient = 0.0
    inflection_depth = None

    for i in range(1, len(depths)):
        d0, d1 = depths[i - 1], depths[i]
        r0, r1 = ratios[d0], ratios[d1]
        if d0 > 0 and d1 > d0:
            # Gradient: ratio change per context doubling
            gradient = (r1 - r0) / (d1 - d0) * d0  # normalized
            if gradient < worst_gradient:
                worst_gradient = gradient
                inflection_depth = d1

    return inflection_depth

flag_unreliable_measurements 

flag_unreliable_measurements() -> list[str]

Flag measurements at 1K context (Metal async dispatch artifact).

Source code in turboquant/quality/hw_replay.py 
251
252
253
254
255
256
257
258
259
260
def flag_unreliable_measurements(self) -> list[str]:
    """Flag measurements at 1K context (Metal async dispatch artifact)."""
    warnings = []
    for b in self.benchmarks:
        if b.context_depth == 1024 and b.tok_per_sec > 10000:
            warnings.append(
                f"Unreliable measurement: {b.label} at 1K context "
                f"({b.tok_per_sec:.0f} tok/s — Metal async dispatch artifact)"
            )
    return warnings

ComparisonReport dataclass 

ComparisonReport(baseline_name: str, target_name: str, hardware_diff: dict = dict(), decode_ratio_curve: dict = dict(), prefill_ratio_curve: dict = dict(), ppl_comparison: dict = dict(), anomalies: list[str] = list())

Result of comparing two hardware profiles.

to_markdown 

to_markdown() -> str

Render comparison as markdown table.

Source code in turboquant/quality/hw_replay.py 
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
def to_markdown(self) -> str:
    """Render comparison as markdown table."""
    lines = [f"# TurboQuant Hardware Comparison: {self.baseline_name} vs {self.target_name}\n"]

    if self.hardware_diff:
        lines.append("## Hardware Differences\n")
        lines.append("| Property | Baseline | Target |")
        lines.append("|----------|----------|--------|")
        for key, (bval, tval) in self.hardware_diff.items():
            lines.append(f"| {key} | {bval} | {tval} |")
        lines.append("")

    if self.decode_ratio_curve:
        lines.append("## Decode Speed Ratio (turbo3/q8_0)\n")
        lines.append("| Context | Baseline | Target | Delta |")
        lines.append("|---------|----------|--------|-------|")
        for depth, (br, tr) in sorted(self.decode_ratio_curve.items()):
            delta = tr - br if tr and br else 0
            flag = " ⚠️" if delta < -0.1 else ""
            lines.append(f"| {depth:,} | {br:.3f}x | {tr:.3f}x | {delta:+.3f}{flag} |")
        lines.append("")

    if self.anomalies:
        lines.append("## Anomalies\n")
        for a in self.anomalies:
            lines.append(f"- {a}")
        lines.append("")

    return "\n".join(lines)

parse_diag_output 

parse_diag_output(text: str) -> HardwareProfile

Parse raw turbo-hardware-diag.sh output into a HardwareProfile.

Source code in turboquant/quality/hw_replay.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def parse_diag_output(text: str) -> HardwareProfile:
    """Parse raw turbo-hardware-diag.sh output into a HardwareProfile."""
    profile = HardwareProfile()
    lines = text.split('\n')

    # Header
    for line in lines:
        if line.startswith("TURBO_DIAG_VERSION="):
            profile.diag_version = int(line.split("=", 1)[1])
        elif line.startswith("TURBO_DIAG_TIMESTAMP="):
            profile.timestamp = line.split("=", 1)[1]
        elif line.startswith("TURBO_DIAG_MODEL="):
            profile.model.filename = line.split("=", 1)[1]

    # Hardware tags
    for line in lines:
        if not line.startswith("[HW]"):
            continue
        kv = line[len("[HW] "):]
        if "=" not in kv:
            continue
        key, val = kv.split("=", 1)
        key = key.strip()
        val = val.strip()
        if key == "os":
            parts = kv.split()
            for part in parts:
                if "=" in part:
                    k, v = part.split("=", 1)
                    if k == "os":
                        profile.system.platform = v
                    elif k == "os_version":
                        profile.system.os_version = v
                    elif k == "arch":
                        profile.system.arch = v
        elif key == "cpu_brand":
            profile.system.cpu_brand = val
        elif key == "cpu_cores_physical":
            profile.system.cpu_cores_physical = _int(val)
        elif key == "cpu_cores_logical":
            profile.system.cpu_cores_logical = _int(val)
        elif key == "ram_total_gb":
            profile.system.ram_total_gb = _int(val)
        elif key == "apple_silicon":
            profile.system.apple_silicon = val.lower() == "true"
        elif key == "chip_model":
            profile.system.chip_model = val
        elif key == "l1_dcache":
            profile.system.l1_dcache = _int(val)
        elif key == "l2_cache":
            profile.system.l2_cache = _int(val)

    # GPU info
    for line in lines:
        if "[GPU]" in line or "[METAL]" in line:
            content = re.sub(r'^\[(GPU|METAL)\]\s*', '', line)
            if "GPU name:" in content:
                profile.system.gpu.name = content.split("GPU name:")[-1].strip()
            elif "GPU family:" in content:
                fam = content.split("GPU family:")[-1].strip()
                profile.system.gpu.family = fam
                m = re.search(r'\((\d+)\)', fam)
                if m:
                    profile.system.gpu.family_id = int(m.group(1))
            elif "has tensor" in content:
                profile.system.gpu.has_tensor = "true" in content.lower()
            elif "has unified" in content:
                profile.system.gpu.has_unified_memory = "true" in content.lower()
            elif "has bfloat" in content:
                profile.system.gpu.has_bfloat = "true" in content.lower()
            elif "recommendedMax" in content:
                m = re.search(r'([\d.]+)\s*MB', content)
                if m:
                    profile.system.gpu.recommended_max_working_set_mb = float(m.group(1))
        if "[METAL_TENSOR]" in line and "has tensor" in line:
            profile.system.gpu.has_tensor = "true" in line.lower()

    # Model info
    for line in lines:
        if not line.startswith("[MODEL]"):
            continue
        content = line[len("[MODEL] "):]
        if "general.name" in content:
            profile.model.name = content.split("=")[-1].strip()
        elif "general.architecture" in content or "arch " in content:
            profile.model.architecture = content.split("=")[-1].strip()
        elif "file type" in content and "file format" not in content:
            profile.model.file_type = content.split("=")[-1].strip()
        elif "model type" in content:
            profile.model.model_type = content.split("=")[-1].strip()
        elif "model params" in content:
            profile.model.params = content.split("=")[-1].strip()
        elif "n_layer" in content:
            profile.model.n_layer = _int(content.split("=")[-1].strip())
        elif "n_head " in content and "n_head_kv" not in content:
            profile.model.n_head = _int(content.split("=")[-1].strip())
        elif "n_head_kv" in content:
            profile.model.n_head_kv = _int(content.split("=")[-1].strip())
        elif "n_expert " in content and "used" not in content:
            profile.model.n_expert = _int(content.split("=")[-1].strip())
        elif "n_expert_used" in content:
            profile.model.n_expert_used = _int(content.split("=")[-1].strip())
        elif "n_ctx_train" in content:
            profile.model.n_ctx_train = _int(content.split("=")[-1].strip())
        elif "n_embd" in content:
            profile.model.n_embd = _int(content.split("=")[-1].strip())
        elif "filename=" in content:
            profile.model.filename = content.split("=", 1)[1]
        elif "filesize_bytes=" in content:
            profile.model.filesize_bytes = _int(content.split("=", 1)[1])

    # Benchmarks — parse llama-bench table output
    _parse_bench_results(lines, profile)

    # PPL results
    _parse_ppl_results(lines, profile)

    # Load snapshots
    _parse_load_snapshots(lines, profile)

    # Build info
    for line in lines:
        if line.startswith("[BUILD]"):
            profile.build_commit = line[len("[BUILD] "):].strip()

    return profile

compare_profiles 

compare_profiles(baseline: HardwareProfile, target: HardwareProfile) -> ComparisonReport

Compare two hardware profiles and identify differences.

Source code in turboquant/quality/hw_replay.py 
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
def compare_profiles(baseline: HardwareProfile, target: HardwareProfile) -> ComparisonReport:
    """Compare two hardware profiles and identify differences."""
    report = ComparisonReport(
        baseline_name=baseline.system.chip_model or baseline.system.cpu_brand,
        target_name=target.system.chip_model or target.system.cpu_brand,
    )

    # Hardware differences
    hw_fields = [
        ("CPU", baseline.system.cpu_brand, target.system.cpu_brand),
        ("RAM (GB)", str(baseline.system.ram_total_gb), str(target.system.ram_total_gb)),
        ("GPU Family", baseline.system.gpu.family, target.system.gpu.family),
        ("GPU Family ID", str(baseline.system.gpu.family_id), str(target.system.gpu.family_id)),
        ("Tensor API", str(baseline.system.gpu.has_tensor), str(target.system.gpu.has_tensor)),
        ("Max Working Set (MB)", f"{baseline.system.gpu.recommended_max_working_set_mb:.0f}",
         f"{target.system.gpu.recommended_max_working_set_mb:.0f}"),
        ("Apple Silicon", str(baseline.system.apple_silicon), str(target.system.apple_silicon)),
    ]
    for name, bval, tval in hw_fields:
        if bval != tval:
            report.hardware_diff[name] = (bval, tval)

    # Decode ratio curves
    base_ratios = baseline.get_ratio_curve("turbo3", "q8_0", "decode")
    target_ratios = target.get_ratio_curve("turbo3", "q8_0", "decode")
    all_depths = sorted(set(base_ratios.keys()) | set(target_ratios.keys()))
    for depth in all_depths:
        br = base_ratios.get(depth, 0)
        tr = target_ratios.get(depth, 0)
        report.decode_ratio_curve[depth] = (br, tr)

    # Prefill ratio curves
    base_pf = baseline.get_ratio_curve("turbo3", "q8_0", "prefill")
    target_pf = target.get_ratio_curve("turbo3", "q8_0", "prefill")
    for depth in sorted(set(base_pf.keys()) | set(target_pf.keys())):
        bp = base_pf.get(depth, 0)
        tp = target_pf.get(depth, 0)
        report.prefill_ratio_curve[depth] = (bp, tp)

    # PPL comparison
    for bp in baseline.ppl_results:
        for tp in target.ppl_results:
            if bp.cache_type == tp.cache_type and bp.env == tp.env:
                report.ppl_comparison[f"{bp.cache_type}_{bp.env or 'uniform'}"] = (bp.ppl, tp.ppl)

    # Detect anomalies
    for depth, (br, tr) in report.decode_ratio_curve.items():
        if br > 0 and tr > 0:
            if tr < br * 0.5:
                report.anomalies.append(
                    f"Decode ratio at {depth:,} is {tr:.3f}x on target vs {br:.3f}x on baseline "
                    f"({tr/br:.0%} of expected). Constant cache thrashing suspected."
                )
            elif tr < 0.5:
                report.anomalies.append(
                    f"Decode ratio at {depth:,} is {tr:.3f}x — below 0.5x threshold. "
                    f"Hardware may not support turbo3 decode at this context depth."
                )

    # Tensor API warning
    if baseline.system.gpu.has_tensor and not target.system.gpu.has_tensor:
        report.anomalies.append(
            "Target lacks Tensor API (M1/M2/M3/M4). "
            "Turbo3 constant cache performance will be significantly worse."
        )

    return report

predict_decode_from_baseline 

predict_decode_from_baseline(baseline: HardwareProfile, target_gpu_family_id: int, target_has_tensor: bool) -> dict[int, float]

Predict target decode ratios from baseline profile using a simple model.

The model: constant cache throughput scales with GPU generation. M1 (family 1007): ~3x worse constant cache than M5 (family 1010) for divergent access patterns. Missing Tensor API adds additional penalty.

Source code in turboquant/quality/hw_replay.py 
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
def predict_decode_from_baseline(
    baseline: HardwareProfile,
    target_gpu_family_id: int,
    target_has_tensor: bool
) -> dict[int, float]:
    """Predict target decode ratios from baseline profile using a simple model.

    The model: constant cache throughput scales with GPU generation.
    M1 (family 1007): ~3x worse constant cache than M5 (family 1010)
    for divergent access patterns. Missing Tensor API adds additional penalty.
    """
    base_ratios = baseline.get_ratio_curve("turbo3", "q8_0", "decode")
    if not base_ratios:
        return {}

    base_family = baseline.system.gpu.family_id
    if base_family == 0 or target_gpu_family_id == 0:
        return base_ratios  # Can't predict without family info

    # Family ID gap (e.g., 1010 - 1007 = 3 generations)
    gen_gap = base_family - target_gpu_family_id

    # Tensor API penalty
    tensor_penalty = 1.0
    if baseline.system.gpu.has_tensor and not target_has_tensor:
        tensor_penalty = 1.3  # 30% additional overhead for missing tensor API

    predicted = {}
    for depth, ratio in base_ratios.items():
        inherent_overhead = 1.0 - ratio
        cache_fraction = min(1.0, (depth / 16384) ** 1.5) if depth > 0 else 0.0
        cache_penalty = inherent_overhead * cache_fraction
        non_cache_penalty = inherent_overhead - cache_penalty

        scaled_cache_penalty = cache_penalty * (1.8 ** gen_gap) * tensor_penalty

        predicted_ratio = max(0.01, 1.0 - non_cache_penalty - scaled_cache_penalty)
        predicted[depth] = round(predicted_ratio, 3)

    return predicted