from __future__ import annotations from typing import Any def _f(value: Any, digits: int = 1) -> str: try: return f"{float(value):,.{digits}f}" except (TypeError, ValueError): return "N/A" def _pct(value: Any, digits: int = 1) -> str: try: return f"{float(value) * 100:.{digits}f}%" except (TypeError, ValueError): return "N/A" def generate_research_report(robust: dict[str, Any], calibration: dict[str, Any] | None = None) -> str: lines: list[str] = [] lines.append("# InferScale Research Consolidation") lines.append("") lines.append("> Results in this report are simulator outputs unless an external-measurement validation section is present. ") lines.append("> Public device/model timings use analytical reference profiles rather than measured hardware benchmarks.") lines.append("") lines.append("## Study protocol") lines.append("") lines.append(f"- Matched workload seeds: **{robust.get('repetitions', 'N/A')}**") lines.append(f"- Bootstrap resamples: **{robust.get('bootstrap_samples', 'N/A')}**") lines.append("- Robust Pareto objectives: p95 step TTFT, unused speculative prefetch, and mean prefix-HBM residency.") lines.append("- Offline reference: bounded full-trace serving oracle over the declared policy/action-plan family.") lines.append("") lines.append("## Robust policy ranking") lines.append("") lines.append(f"**Robust winner:** {robust.get('robust_winner', 'N/A')}") lines.append("") lines.append(f"**Nominal first-seed winner:** {robust.get('nominal_winner_first_seed', 'N/A')}") lines.append("") lines.append("| Rank | Policy | Median p95 TTFT | 95% CI of mean | TTFT wins | Pareto stability | Median oracle regret | Worst seed |") lines.append("|---:|---|---:|---:|---:|---:|---:|---:|") for row in sorted(robust.get("policies", []), key=lambda item: item.get("robust_rank", 999)): ci = f"{_f(row.get('ttft_ci95_low_ms'))}-{_f(row.get('ttft_ci95_high_ms'))} ms" lines.append( f"| {row.get('robust_rank', '')} | {row.get('label', '')} | {_f(row.get('median_ttft_ms'))} ms | {ci} | " f"{_pct(row.get('ttft_win_rate'))} | {_pct(row.get('pareto_stability'))} | " f"{_f(row.get('median_oracle_regret_ms'))} ms | {_f(row.get('worst_seed_ttft_ms'))} ms |" ) lines.append("") oracle = robust.get("oracle", {}) lines.append("## Offline constrained oracle") lines.append("") lines.append(f"- Candidate plans per seed: **{oracle.get('candidate_count_per_seed', 'N/A')}**") lines.append(f"- Median oracle p95 TTFT: **{_f(oracle.get('median_ttft_ms'))} ms**") lines.append(f"- Definition: `{oracle.get('definition', 'N/A')}`") lines.append("") lines.append(oracle.get("note", "")) lines.append("") if calibration: lines.append("## External measurement calibration") lines.append("") lines.append(f"- Validation mode: **{calibration.get('validation_mode', 'N/A')}**") lines.append(f"- Training cases: **{calibration.get('train_count', 0)}**") lines.append(f"- Validation cases: **{calibration.get('holdout_count', 0)}**") scales = calibration.get("fitted_scales", {}) lines.append(f"- Prefill time scale: **{_f(scales.get('prefill_time_scale'), 3)}x**") lines.append(f"- Decode time scale: **{_f(scales.get('decode_time_scale'), 3)}x**") lines.append(f"- Baseline MAPE: **{_f(calibration.get('baseline', {}).get('mape_pct'), 2)}%**") lines.append(f"- Calibrated MAPE: **{_f(calibration.get('calibrated', {}).get('mape_pct'), 2)}%**") lines.append("") lines.append(calibration.get("note", "")) lines.append("") lines.append("## Interpretation guardrails") lines.append("") lines.append("1. Bootstrap intervals quantify variation across simulated workload seeds; they do not quantify real-hardware model error.") lines.append("2. The offline oracle is bounded to an explicit action-plan family and should not be described as a globally optimal serving controller.") lines.append("3. Analytical reference profiles are hypotheses about timing. Hardware claims require independent measured validation.") lines.append("4. External calibration can correct global scale bias but does not prove fidelity for unseen models, devices, or scheduler regimes.") lines.append("") return "\n".join(lines).strip() + "\n"