Aggregate E2E normalized interactivity in AgentX CI results / 在 AgentX CI 结果中聚合端到端归一化交互性

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
 
 

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
         ),

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):