diff --git a/utils/agentic/aggregation/request_metrics.py b/utils/agentic/aggregation/request_metrics.py index 26fc49df17..4ac8f454e7 100644 --- a/utils/agentic/aggregation/request_metrics.py +++ b/utils/agentic/aggregation/request_metrics.py @@ -129,17 +129,23 @@ def _nest_stats(prefix: str, flat: dict[str, Any]) -> dict[str, Any]: } -def _interactivity_stats(itl_stats: dict[str, Any], itls: list[float]) -> dict[str, float]: +def _interactivity_stats( + itl_stats: dict[str, Any], + itls: list[float], + *, + itl_prefix: str = "itl", + intvty_prefix: str = "intvty", +) -> dict[str, float]: """Derive slow-tail interactivity from the matching ITL statistic.""" out: dict[str, float] = {} for key in ("mean", "p50", "p75", "p90", "p95"): - value = itl_stats.get(f"{key}_itl") + value = itl_stats.get(f"{key}_{itl_prefix}") if isinstance(value, int | float) and not isinstance(value, bool) and value > 0: - out[f"{key}_intvty"] = 1.0 / value + out[f"{key}_{intvty_prefix}"] = 1.0 / value per_request = [1.0 / value for value in itls if value > 0] if per_request: - out["std_intvty"] = ( + out[f"std_{intvty_prefix}"] = ( statistics.pstdev(per_request) if len(per_request) > 1 else 0.0 ) return out @@ -149,11 +155,21 @@ def compute_latency_stats(records: list[dict[str, Any]]) -> tuple[dict[str, Any] ttfts = _ms_to_s(extract_per_record_floats(records, "time_to_first_token")) e2els = _ms_to_s(extract_per_record_floats(records, "request_latency")) itls = _ms_to_s(extract_per_record_floats(records, "inter_token_latency")) + full_response_itls = _ms_to_s( + extract_per_record_floats(records, "full_response_inter_token_latency") + ) ttft_stats = stats_for("ttft", ttfts) e2el_stats = stats_for("e2el", e2els) itl_stats = stats_for("itl", itls) tpot_stats = stats_for("tpot", itls) intvty_stats = _interactivity_stats(itl_stats, itls) + full_response_itl_stats = stats_for("full_response_itl", full_response_itls) + full_response_intvty_stats = _interactivity_stats( + full_response_itl_stats, + full_response_itls, + itl_prefix="full_response_itl", + intvty_prefix="full_response_intvty", + ) flat: dict[str, Any] = {} flat.update(ttft_stats) @@ -161,6 +177,8 @@ def compute_latency_stats(records: list[dict[str, Any]]) -> tuple[dict[str, Any] flat.update(itl_stats) flat.update(tpot_stats) flat.update(intvty_stats) + flat.update(full_response_itl_stats) + flat.update(full_response_intvty_stats) nested = { "ttft": _nest_stats("ttft", ttft_stats), @@ -168,6 +186,12 @@ def compute_latency_stats(records: list[dict[str, Any]]) -> tuple[dict[str, Any] "itl": _nest_stats("itl", itl_stats), "tpot": _nest_stats("tpot", tpot_stats), "intvty": _nest_stats("intvty", intvty_stats), + "full_response_itl": _nest_stats( + "full_response_itl", full_response_itl_stats + ), + "full_response_intvty": _nest_stats( + "full_response_intvty", full_response_intvty_stats + ), } return flat, nested diff --git a/utils/agentic/aggregation/test_process_agentic_result.py b/utils/agentic/aggregation/test_process_agentic_result.py index 21a7307630..4649c870c0 100644 --- a/utils/agentic/aggregation/test_process_agentic_result.py +++ b/utils/agentic/aggregation/test_process_agentic_result.py @@ -123,7 +123,15 @@ "raw", } REQUEST_METRICS_KEYS = {"qps", "latency", "tokens", "throughput", "cache"} -REQUEST_LATENCY_KEYS = {"ttft", "e2el", "itl", "tpot", "intvty"} +REQUEST_LATENCY_KEYS = { + "ttft", + "e2el", + "itl", + "tpot", + "intvty", + "full_response_itl", + "full_response_intvty", +} REQUEST_TOKEN_KEYS = {"input", "output_actual", "output_expected"} REQUEST_THROUGHPUT_KEYS = { "input", @@ -510,6 +518,48 @@ def test_processor_throughput_per_gpu(tmp_path: Path): ) +def test_processor_aggregates_full_response_itl_and_interactivity(tmp_path: Path): + result_dir = tmp_path / "results" + artifact = result_dir / "aiperf_artifacts" + artifact.mkdir(parents=True) + + full_response_itls_ms = (5.469791, 5.0, 4.0) + with open(artifact / "profile_export.jsonl", "w") as f: + for idx, full_response_itl_ms in enumerate(full_response_itls_ms): + record = _make_record( + conv_id=f"trace-{idx}", + turn_index=0, + isl=100, + osl=26_571, + ttft_ms=529.058811, + e2e_ms=610.559573, + itl_ms=0.003067398, + start_ns=(idx + 1) * 1_000_000_000, + end_ns=(idx + 1) * 1_000_000_000 + 145_861_451_008, + ) + record["metrics"]["full_response_inter_token_latency"] = { + "value": full_response_itl_ms, + "unit": "ms", + } + f.write(json.dumps(record) + "\n") + + with open(artifact / "profile_export_aiperf.json", "w") as f: + json.dump({"request_count": len(full_response_itls_ms)}, f) + + agg = _run_processor(result_dir, tmp_path / "out") + latency = agg["request_metrics"]["latency"] + full_response_itl = latency["full_response_itl"] + full_response_intvty = latency["full_response_intvty"] + + assert full_response_itl["p50"] == pytest.approx(0.005) + assert full_response_itl["p75"] == pytest.approx(0.00523) + assert full_response_intvty["p50"] == pytest.approx( + 1 / full_response_itl["p50"] + ) + assert full_response_intvty["p75"] == pytest.approx(1 / 0.0052348955) + assert full_response_intvty["p75"] < full_response_intvty["p50"] + + def test_processor_surfaces_allocated_cpu_dram(tmp_path: Path): result_dir = _write_fixture(tmp_path) diff --git a/utils/aiperf b/utils/aiperf index b7b16cf851..743fb82ab7 160000 --- a/utils/aiperf +++ b/utils/aiperf @@ -1 +1 @@ -Subproject commit b7b16cf851885567988a643282266bce74e34437 +Subproject commit 743fb82ab798dc898fce98d0321db61e7564942a diff --git a/utils/process_changelog.py b/utils/process_changelog.py index 735284750a..577b109b0c 100644 --- a/utils/process_changelog.py +++ b/utils/process_changelog.py @@ -68,8 +68,9 @@ def trim_conc(entries: list[dict]) -> list[dict]: ``int`` (single-node) or ``list`` (multi-node). Other fields may contain nested dictionaries or lists, such as KV-offload backend metadata. - - Single-node entries: group by every other field and keep only the entry - with the lowest ``conc`` per group. + - Single-node entries: group by every configuration field other than + ``conc`` and the generated ``exp-name``, then keep only the entry with + the lowest ``conc`` per group. - Multi-node entries: trim the ``conc`` list in place to ``[min(conc)]``. """ groups: dict[tuple, list[int]] = {} @@ -87,7 +88,7 @@ def trim_conc(entries: list[dict]) -> list[dict]: sorted( (k, _freeze_config_value(v)) for k, v in entry.items() - if k != "conc" + if k not in {"conc", "exp-name"} ) ) groups.setdefault(key, []).append(len(out)) diff --git a/utils/test_process_changelog.py b/utils/test_process_changelog.py index 146fd9e568..9febfa9444 100644 --- a/utils/test_process_changelog.py +++ b/utils/test_process_changelog.py @@ -25,12 +25,13 @@ def test_trim_conc_supports_nested_backend_metadata(): }, } entries = [ - {**common, "conc": 8}, - {**common, "conc": 2}, + {**common, "conc": 8, "exp-name": "kimi_tp8_conc8_kvdram"}, + {**common, "conc": 2, "exp-name": "kimi_tp8_conc2_kvdram"}, { **common, "kv-offload-backend": {"name": "lmcache"}, "conc": 4, + "exp-name": "kimi_tp8_conc4_lmcache", }, ]