diff --git a/scripts/analysis/recompute_summary.py b/scripts/analysis/recompute_summary.py new file mode 100644 index 0000000..a2f4bb3 --- /dev/null +++ b/scripts/analysis/recompute_summary.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Re-derive summary.json from existing metrics.jsonl using the fixed metrics.py. + +Bug fixed: requests aborted by SGLang (e.g. input > max-input-len returns +a fast 400 with latency_s ~ 0.08s) were previously counted in latency_stats +as if successful, deflating mean/p50/p90. The fixed metrics.py excludes +all failed requests (errors or aborts) from latency/ttft/tpot stats and +exposes abort_count / failure_count. + +Usage: + python3 scripts/analysis/recompute_summary.py path/to/metrics.jsonl ... + python3 scripts/analysis/recompute_summary.py --diff path/to/metrics.jsonl path/to/old_summary.json +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src")) + +from agentic_pd_hybrid.metrics import RequestMetrics, write_summary_json + + +def load_rows(metrics_path: Path) -> list[RequestMetrics]: + rows = [] + field_names = {f for f in RequestMetrics.__dataclass_fields__} + with metrics_path.open() as handle: + for line in handle: + line = line.strip() + if not line: + continue + raw = json.loads(line) + kwargs = {k: raw.get(k) for k in field_names} + rows.append(RequestMetrics(**kwargs)) + return rows + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("metrics_paths", nargs="+", type=Path) + parser.add_argument( + "--out", + type=Path, + default=None, + help="output summary path (default: alongside metrics with .recomputed_summary.json)", + ) + parser.add_argument( + "--diff", + action="store_true", + help="print before/after diff against the old .summary.json", + ) + args = parser.parse_args() + + for metrics_path in args.metrics_paths: + rows = load_rows(metrics_path) + out_path = args.out or metrics_path.with_suffix(".recomputed_summary.json") + write_summary_json( + out_path, + rows, + trace_path=metrics_path, + router_url=None, + ) + new = json.load(out_path.open()) + print(f"\n=== {metrics_path} ===") + print(f" written: {out_path}") + print(f" total rows: {new['request_count']}") + print(f" error_count: {new['error_count']}") + print(f" abort_count: {new.get('abort_count', '?')}") + print(f" failure_count: {new.get('failure_count', '?')}") + ls = new.get("latency_stats_s", {}) or {} + ts = new.get("ttft_stats_s", {}) or {} + print(f" lat: n={ls.get('count')} mean={ls.get('mean'):.4f} p50={ls.get('p50'):.4f} p90={ls.get('p90'):.4f} p99={ls.get('p99'):.4f}") + print(f" ttft: n={ts.get('count')} mean={ts.get('mean'):.4f} p50={ts.get('p50'):.4f} p90={ts.get('p90'):.4f} p99={ts.get('p99'):.4f}") + + if args.diff: + # find old summary (sibling file) + candidates = [ + metrics_path.parent / f"{metrics_path.stem}.summary.json", + metrics_path.with_suffix(".summary.json"), + ] + old_path = next((p for p in candidates if p.exists()), None) + if old_path: + old = json.load(old_path.open()) + print(f" vs old {old_path}:") + old_ls = old.get("latency_stats_s", {}) or {} + old_ts = old.get("ttft_stats_s", {}) or {} + for k in ("count", "mean", "p50", "p90", "p99"): + o = old_ls.get(k) + n = ls.get(k) + if o is not None and n is not None: + delta = n - o + print(f" lat.{k}: {o:.4f} -> {n:.4f} ({delta:+.4f})") + for k in ("count", "mean", "p50", "p90", "p99"): + o = old_ts.get(k) + n = ts.get(k) + if o is not None and n is not None: + delta = n - o + print(f" ttft.{k}: {o:.4f} -> {n:.4f} ({delta:+.4f})") + + +if __name__ == "__main__": + main() diff --git a/src/agentic_pd_hybrid/metrics.py b/src/agentic_pd_hybrid/metrics.py index b7d9bba..c3b9a8a 100644 --- a/src/agentic_pd_hybrid/metrics.py +++ b/src/agentic_pd_hybrid/metrics.py @@ -114,6 +114,16 @@ def write_metrics_jsonl(path: Path, rows: list[RequestMetrics]) -> None: handle.write(json.dumps(asdict(row), sort_keys=True) + "\n") +def _is_failed_request(row: RequestMetrics) -> bool: + if row.error is not None: + return True + if row.finish_reason is not None: + fr = str(row.finish_reason).lower() + if "abort" in fr or "badrequest" in fr: + return True + return False + + def write_summary_json( path: Path, rows: list[RequestMetrics], @@ -121,9 +131,10 @@ def write_summary_json( trace_path: Path, router_url: str | None, ) -> None: - latencies = [row.latency_s for row in rows if row.latency_s is not None] - ttfts = [row.ttft_s for row in rows if row.ttft_s is not None] - tpots = [row.tpot_s for row in rows if row.tpot_s is not None] + successful = [row for row in rows if not _is_failed_request(row)] + latencies = [row.latency_s for row in successful if row.latency_s is not None] + ttfts = [row.ttft_s for row in successful if row.ttft_s is not None] + tpots = [row.tpot_s for row in successful if row.tpot_s is not None] per_decode_load = Counter(row.assigned_decode_node for row in rows) per_prefill_load = Counter(row.assigned_prefill_node for row in rows) prefill_priorities = Counter( @@ -167,6 +178,17 @@ def write_summary_json( str(key): value for key, value in sorted(decode_priorities.items()) }, "error_count": sum(1 for row in rows if row.error is not None), + "abort_count": sum( + 1 + for row in rows + if row.error is None + and row.finish_reason is not None + and ( + "abort" in str(row.finish_reason).lower() + or "badrequest" in str(row.finish_reason).lower() + ) + ), + "failure_count": sum(1 for row in rows if _is_failed_request(row)), "truncated_request_count": sum( 1 for row in rows