From 9aa4bbac5524975602b2d8d79ed92e82a1cbef6c Mon Sep 17 00:00:00 2001 From: Duyi-Wang Date: Mon, 10 Aug 2026 15:04:17 +0800 Subject: [PATCH] feat(agentx): aggregate E2E normalized interactivity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compute slow-tail E2E normalized interactivity from per-request E2EL/OSL ratios and emit it in compact AgentX benchmark results. Reject incomplete, nonpositive, and nonfinite samples while preserving a stable empty schema. Add end-to-end and edge-case coverage. 中文：从逐请求 E2EL/OSL 比值计算慢尾端到端归一化交互性，并将其写入紧凑的 AgentX 基准测试结果。过滤字段不完整、非正数及非有限值样本，同时保持稳定的空对象 schema；补充端到端与边界条件测试。 --- .../agentic/aggregation/aggregation_common.py | 2 +- utils/agentic/aggregation/request_metrics.py | 42 ++++- .../test_process_agentic_result.py | 148 +++++++++++++++++- 3 files changed, 187 insertions(+), 5 deletions(-) diff --git a/utils/agentic/aggregation/aggregation_common.py b/utils/agentic/aggregation/aggregation_common.py index 1fc13f951e..5a3f55d3c3 100644 --- a/utils/agentic/aggregation/aggregation_common.py +++ b/utils/agentic/aggregation/aggregation_common.py @@ -47,7 +47,7 @@ def to_int(value: Any) -> int | None: return None try: return int(value) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): return None diff --git a/utils/agentic/aggregation/request_metrics.py b/utils/agentic/aggregation/request_metrics.py index 4ac8f454e7..b74dfb963e 100644 --- a/utils/agentic/aggregation/request_metrics.py +++ b/utils/agentic/aggregation/request_metrics.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import math import statistics from collections.abc import Iterable from pathlib import Path @@ -104,7 +105,11 @@ def extract_per_record_ints(records: list[dict[str, Any]], key: str) -> list[int def _ms_to_s(values_ms: Iterable[float]) -> list[float]: - return [value / 1000.0 for value in values_ms if value is not None and value > 0] + return [ + value / 1000.0 + for value in values_ms + if value is not None and math.isfinite(value) and value > 0 + ] def _distribution(prefix: str, values: list[int]) -> dict[str, float]: @@ -136,7 +141,7 @@ def _interactivity_stats( itl_prefix: str = "itl", intvty_prefix: str = "intvty", ) -> dict[str, float]: - """Derive slow-tail interactivity from the matching ITL statistic.""" + """Derive slow-tail interactivity from the matching latency statistic.""" out: dict[str, float] = {} for key in ("mean", "p50", "p75", "p90", "p95"): value = itl_stats.get(f"{key}_{itl_prefix}") @@ -151,6 +156,34 @@ def _interactivity_stats( return out +def _e2e_normalized_interactivity_stats( + records: list[dict[str, Any]], +) -> dict[str, float]: + """Derive per-user output rate from each request's E2EL/OSL ratio.""" + e2el_per_osl: list[float] = [] + for record in records: + e2el_ms = to_float(_metric_value(record, "request_latency")) + osl = to_float(_metric_value(record, "output_sequence_length")) + if ( + e2el_ms is None + or osl is None + or not math.isfinite(e2el_ms) + or not math.isfinite(osl) + or e2el_ms <= 0 + or osl <= 0 + ): + continue + e2el_per_osl.append(e2el_ms / 1000.0 / osl) + + ratio_stats = stats_for("e2el_per_osl", e2el_per_osl) + return _interactivity_stats( + ratio_stats, + e2el_per_osl, + itl_prefix="e2el_per_osl", + intvty_prefix="e2e_norm_intvty", + ) + + def compute_latency_stats(records: list[dict[str, Any]]) -> tuple[dict[str, Any], 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")) @@ -163,6 +196,7 @@ def compute_latency_stats(records: list[dict[str, Any]]) -> tuple[dict[str, Any] itl_stats = stats_for("itl", itls) tpot_stats = stats_for("tpot", itls) intvty_stats = _interactivity_stats(itl_stats, itls) + e2e_norm_intvty_stats = _e2e_normalized_interactivity_stats(records) full_response_itl_stats = stats_for("full_response_itl", full_response_itls) full_response_intvty_stats = _interactivity_stats( full_response_itl_stats, @@ -177,6 +211,7 @@ 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(e2e_norm_intvty_stats) flat.update(full_response_itl_stats) flat.update(full_response_intvty_stats) @@ -186,6 +221,9 @@ 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), + "e2e_norm_intvty": _nest_stats( + "e2e_norm_intvty", e2e_norm_intvty_stats + ), "full_response_itl": _nest_stats( "full_response_itl", full_response_itl_stats ), diff --git a/utils/agentic/aggregation/test_process_agentic_result.py b/utils/agentic/aggregation/test_process_agentic_result.py index 4649c870c0..7f5b0395bf 100644 --- a/utils/agentic/aggregation/test_process_agentic_result.py +++ b/utils/agentic/aggregation/test_process_agentic_result.py @@ -129,6 +129,7 @@ "itl", "tpot", "intvty", + "e2e_norm_intvty", "full_response_itl", "full_response_intvty", } @@ -494,6 +495,146 @@ def test_processor_derives_interactivity_from_matching_itl_percentile( assert latency["intvty"]["p90"] < 20 +def test_processor_aggregates_e2e_normalized_interactivity_from_slow_tail( + tmp_path: Path, +): + result_dir = tmp_path / "results" + artifact = result_dir / "aiperf_artifacts" + artifact.mkdir(parents=True) + + # E2EL / OSL ratios are 0.02 and 0.04 seconds per output token. + records = [ + _make_record( + conv_id="trace-fast", + turn_index=0, + isl=100, + osl=50, + ttft_ms=30.0, + e2e_ms=1_000.0, + itl_ms=10.0, + start_ns=1_000_000_000, + end_ns=2_000_000_000, + ), + _make_record( + conv_id="trace-slow", + turn_index=0, + isl=100, + osl=50, + ttft_ms=30.0, + e2e_ms=2_000.0, + itl_ms=10.0, + start_ns=2_000_000_000, + end_ns=4_000_000_000, + ), + ] + with open(artifact / "profile_export.jsonl", "w") as f: + for record in records: + f.write(json.dumps(record) + "\n") + with open(artifact / "profile_export_aiperf.json", "w") as f: + json.dump({"request_count": len(records)}, f) + + agg = _run_processor(result_dir, tmp_path / "out") + metric = agg["request_metrics"]["latency"]["e2e_norm_intvty"] + + assert metric["mean"] == pytest.approx(1 / 0.03) + assert metric["p75"] == pytest.approx(1 / 0.035) + assert metric["p90"] == pytest.approx(1 / 0.038) + assert metric["std"] == pytest.approx(12.5) + assert metric["p50"] >= metric["p75"] >= metric["p90"] >= metric["p95"] + assert "p99" not in metric + _assert_stable_request_metrics_schema(agg) + + +def test_e2e_normalized_interactivity_pairs_metrics_within_each_record(): + missing_osl = _make_record( + conv_id="trace-missing-osl", + turn_index=0, + isl=100, + osl=50, + ttft_ms=30.0, + e2e_ms=1_000.0, + itl_ms=10.0, + start_ns=1_000_000_000, + end_ns=2_000_000_000, + ) + del missing_osl["metrics"]["output_sequence_length"] + missing_e2el = _make_record( + conv_id="trace-missing-e2el", + turn_index=0, + isl=100, + osl=50, + ttft_ms=30.0, + e2e_ms=1_000.0, + itl_ms=10.0, + start_ns=2_000_000_000, + end_ns=3_000_000_000, + ) + del missing_e2el["metrics"]["request_latency"] + + _, nested = compute_request_metrics([missing_osl, missing_e2el]) + + assert nested["latency"]["e2e_norm_intvty"] == {} + + +def test_e2e_normalized_interactivity_skips_nonpositive_and_nonfinite_values(): + invalid_pairs = [ + (0.0, 50), + (-1.0, 50), + (float("nan"), 50), + (float("inf"), 50), + (1_000.0, 0), + (1_000.0, -1), + (1_000.0, float("nan")), + (1_000.0, float("inf")), + ] + records = [] + for idx, (e2e_ms, osl) in enumerate(invalid_pairs): + records.append( + _make_record( + conv_id=f"trace-invalid-{idx}", + turn_index=0, + isl=100, + osl=osl, + ttft_ms=30.0, + e2e_ms=e2e_ms, + itl_ms=10.0, + start_ns=(idx + 1) * 1_000_000_000, + end_ns=(idx + 2) * 1_000_000_000, + ) + ) + records.append( + _make_record( + conv_id="trace-valid", + turn_index=0, + isl=100, + osl=50, + ttft_ms=30.0, + e2e_ms=1_000.0, + itl_ms=10.0, + start_ns=10_000_000_000, + end_ns=11_000_000_000, + ) + ) + + _, nested = compute_request_metrics(records) + metric = nested["latency"]["e2e_norm_intvty"] + + assert metric == { + "mean": 50.0, + "p50": 50.0, + "p75": 50.0, + "p90": 50.0, + "p95": 50.0, + "std": 0.0, + } + + +def test_e2e_normalized_interactivity_empty_without_valid_samples(): + _, nested = compute_request_metrics([]) + + assert nested["latency"]["e2e_norm_intvty"] == {} + + def test_processor_throughput_per_gpu(tmp_path: Path): result_dir = _write_fixture(tmp_path) output_dir = tmp_path / "out" @@ -676,7 +817,7 @@ def test_processor_surfaces_request_accounting(tmp_path: Path): isl=100, osl=50, ttft_ms=30.0, - e2e_ms=1_000.0, + e2e_ms=10_000.0, itl_ms=10.0, start_ns=2_000_000_000, end_ns=3_000_000_000, @@ -688,7 +829,7 @@ def test_processor_surfaces_request_accounting(tmp_path: Path): isl=100, osl=50, ttft_ms=30.0, - e2e_ms=1_000.0, + e2e_ms=20_000.0, itl_ms=10.0, start_ns=3_000_000_000, end_ns=4_000_000_000, @@ -714,6 +855,9 @@ def test_processor_surfaces_request_accounting(tmp_path: Path): "error_categories": {"HTTPStatusError": 1}, } assert agg["server_metrics"]["tokens"]["requests_completed"] == 1 + e2e_norm_intvty = agg["request_metrics"]["latency"]["e2e_norm_intvty"] + assert e2e_norm_intvty["mean"] == pytest.approx(50.0) + assert e2e_norm_intvty["p95"] == pytest.approx(50.0) def test_processor_handles_missing_server_metrics(tmp_path: Path):