From 62d433e1468a538da0f93a9e90c2c0645ce7faf9 Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Mon, 7 Sep 2026 13:34:23 +0000 Subject: [PATCH 1/7] [Frontend][Benchmark] Add duplex metrics for OmniInteract benchmark Signed-off-by: ZacheryAU Co-authored-by: Ruirui Yang | Rein <73573651+R2-Y@users.noreply.github.com> --- docs/cli/bench/serve.md | 8 +- tests/benchmarks/patch/test_patch.py | 3 + tests/benchmarks/test_omniinteract.py | 66 ++++++ tests/clients/test_duplex_client.py | 146 ++++++++++++ .../test_minicpmo_realtime_duplex_drivers.py | 4 + tests/engine/test_output_processor.py | 100 +++++++++ .../openai_api/test_duplex_handler.py | 65 +++++- vllm_omni/benchmarks/omniinteract.py | 30 ++- vllm_omni/benchmarks/patch/patch.py | 15 ++ vllm_omni/clients/duplex.py | 211 ++++++++++++++++-- vllm_omni/entrypoints/duplex/protocol.py | 93 +++++++- .../entrypoints/duplex/runtime_bridge.py | 25 ++- .../entrypoints/duplex/session_runner.py | 2 + vllm_omni/outputs/output_processor.py | 11 +- 14 files changed, 734 insertions(+), 45 deletions(-) diff --git a/docs/cli/bench/serve.md b/docs/cli/bench/serve.md index a162c001a0f..8dc5e1ed89b 100644 --- a/docs/cli/bench/serve.md +++ b/docs/cli/bench/serve.md @@ -316,8 +316,12 @@ arrives, checkpointing the response's history position so a later committed user outputs are ineligible and omitted from the official manifest; `audio_clipped_bytes` records output beyond the rounded video horizon. -TTFT, TTFP, and RTF start at client receipt of `response.created`. TPOT/ITL use engine stage-0 timing; ITL is emitted only when -every token interval is present. +Per-response TTFT and TTFP start when the server begins executing the native model-turn request that owns the response. RTF +continues to use client receipt of `response.created` through the last audio packet, divided by emitted audio duration. Global +TTFT, TTFP, and RTF cover the complete input-stream window. TPOT/ITL use engine stage-0 timing; ITL is emitted only when every +token interval is present within a continuous generation segment. Model-unit pacing and gaps between generation segments are +excluded from TPOT/ITL. Raw request metrics retain `response_created_to_first_text_ms` and +`response_created_to_first_audio_ms` as client-envelope diagnostics. The checked-in local performance configuration measures four deterministic cases from each OmniInteract subset (12 videos total), with no benchmark warmups and a maximum concurrency of two. Each subset also sends one readiness request before its diff --git a/tests/benchmarks/patch/test_patch.py b/tests/benchmarks/patch/test_patch.py index accb3010077..664842d275c 100644 --- a/tests/benchmarks/patch/test_patch.py +++ b/tests/benchmarks/patch/test_patch.py @@ -287,8 +287,10 @@ async def close_session(self, **_kwargs): "ttft": "conversation.item.create client send to first non-empty text delta", "ttfp": "conversation.item.create client send to first audio packet", "rtf": "request-start-to-last-audio receive time divided by emitted audio duration", + "tpot": "Stage-0 engine mean time per output token", }, "ttft_ms": pytest.approx(20.0, abs=2.0), + "tpot_ms": 10.0, "ttfp_ms": pytest.approx(30.0, abs=2.0), "rtf": pytest.approx(0.3, abs=0.03), "audio_generation_ms": pytest.approx(30.0, abs=2.0), @@ -298,6 +300,7 @@ async def close_session(self, **_kwargs): "session_id": session_id, "audio_turn_count": 4, "mean_ttft_ms": pytest.approx(20.0, abs=2.0), + "mean_tpot_ms": 10.0, "mean_ttfp_ms": pytest.approx(30.0, abs=2.0), "mean_rtf": pytest.approx(0.3, abs=0.03), } diff --git a/tests/benchmarks/test_omniinteract.py b/tests/benchmarks/test_omniinteract.py index f9488b11cb8..b93c0ed0b6e 100644 --- a/tests/benchmarks/test_omniinteract.py +++ b/tests/benchmarks/test_omniinteract.py @@ -148,6 +148,72 @@ def _successful_output( return case, collector, result +def test_response_metrics_include_engine_tpot_and_global_stream_window(): + first_audio = _audio("r1") + first_audio["metadata"] = { + "audio_duration_ms": 100, + "vllm_omni": { + "stage_metrics": { + "0": { + "num_tokens_out": 3, + "vllm_tpot_ms": 10.0, + "vllm_itls_ms": [10.0, 10.0], + } + } + }, + } + second_audio = _audio("r2") + second_audio["metadata"] = { + "audio_duration_ms": 200, + "vllm_omni": { + "stage_metrics": { + "0": { + "num_tokens_out": 2, + "vllm_tpot_ms": 20.0, + "vllm_itls_ms": [20.0], + } + } + }, + } + collector = _collector( + (_created("r1"), 10.1), + (_text("r1", "first"), 10.2), + (first_audio, 10.3), + (_done("r1"), 10.4), + (_created("r2"), 12.1), + (_text("r2", "second"), 12.2), + (second_audio, 12.5), + (_done("r2"), 12.6), + ) + result = oi.OmniInteractCaseResult("1q1a", "video.mp4", "", session_id="session") + + oi._populate_response_metrics(result, collector, stream_start=10.0) + + assert result.output_tokens == 5 + assert [metric["tpot_ms"] for metric in result.duplex_request_metrics] == [10.0, 20.0] + assert result.duplex_session_metrics == { + "session_id": "session", + "audio_turn_count": 2, + "mean_ttft_ms": 100.0, + "mean_tpot_ms": 15.0, + "mean_ttfp_ms": 300.0, + "mean_rtf": 2.0, + "global_ttft_ms": 200.0, + "global_ttfp_ms": 300.0, + "global_rtf": 8.333333, + "global_audio_generation_ms": 2500.0, + "global_audio_duration_ms": 300.0, + "global_measurement_origin": { + "ttft": "input stream start to first non-empty text delta", + "ttfp": "input stream start to first audio packet", + "rtf": ( + "input stream start-to-last-audio receive time divided by total emitted audio duration; " + "includes concurrent realtime input" + ), + }, + } + + def _write_success( root: Path, case: data.OmniInteractCase, diff --git a/tests/clients/test_duplex_client.py b/tests/clients/test_duplex_client.py index 1be3eb9432e..26beccceb6c 100644 --- a/tests/clients/test_duplex_client.py +++ b/tests/clients/test_duplex_client.py @@ -736,11 +736,157 @@ def test_event_collector_reports_engine_token_and_audio_intervals(): "measurement_origin": { "ttft": "input_audio_buffer.commit client send to first non-empty text delta", "ttfp": "input_audio_buffer.commit client send to first audio packet", + "tpot": "Stage-0 engine mean time per output token", }, "ttft_ms": 250.0, "ttfp_ms": 300.0, "audio_generation_ms": 460.0, "audio_duration_ms": 240.0, + "tpot_ms": 15.0, + } + + +def test_session_summary_omits_unmeasured_zero_tpot(): + summary = summarize_session_request_metrics( + [ + {"ttft_ms": 100.0, "tpot_ms": 0.0}, + {"ttft_ms": 200.0, "tpot_ms": None}, + ], + session_id="session", + ) + + assert summary["mean_ttft_ms"] == 150.0 + assert "mean_tpot_ms" not in summary + + +def test_event_collector_reports_global_metrics_across_responses(): + collector = EventCollector() + for response_id, created_at_s, text_at_s, audio_times, durations in ( + ("resp-a", 10.1, 10.2, (10.3, 10.4), (100, 200)), + ("resp-b", 12.1, 12.2, (12.3, 12.5), (150, 300)), + ): + collector.add( + {"type": "response.created", "response": {"id": response_id}}, + received_at_s=created_at_s, + ) + collector.add( + { + "type": "response.audio_transcript.delta", + "response_id": response_id, + "delta": response_id, + }, + received_at_s=text_at_s, + ) + for received_at_s, duration_ms in zip(audio_times, durations, strict=True): + collector.add( + { + "type": "response.audio.delta", + "response_id": response_id, + "delta": base64.b64encode(b"audio").decode("ascii"), + "sample_rate_hz": 16_000, + "metadata": {"audio_duration_ms": duration_ms}, + }, + received_at_s=received_at_s, + ) + + metrics = collector.global_timing_summary( + after_s=10.0, + window_started_at_s=10.0, + response_ids=["resp-a", "resp-b"], + measurement_origin={ + "ttft": "input stream start to first non-empty text delta", + "ttfp": "input stream start to first audio packet", + "rtf": "input stream start-to-last-audio receive time divided by total emitted audio duration", + }, + ) + + assert metrics == { + "source": "client_monotonic_receive", + "response_ids": ["resp-a", "resp-b"], + "measurement_origin": { + "ttft": "input stream start to first non-empty text delta", + "ttfp": "input stream start to first audio packet", + "rtf": "input stream start-to-last-audio receive time divided by total emitted audio duration", + }, + "ttft_ms": 200.0, + "ttfp_ms": 300.0, + "audio_generation_ms": 2500.0, + "audio_duration_ms": 500.0, + } + + +def test_event_collector_prefers_server_request_start_for_ttf(): + collector = EventCollector() + collector.add( + { + "type": "response.created", + "response": { + "id": "resp-a", + "metadata": { + "duplex_event": { + "response_request_metrics": { + "source": "server_monotonic_request_start", + "measurement_origin": { + "ttft": "native request start to first text output", + "ttfp": "native request start to first audio output", + }, + "ttft_ms": 125.0, + "ttfp_ms": 175.0, + } + } + }, + }, + }, + received_at_s=10.0, + ) + stage_metrics = { + "0": { + "num_tokens_out": 4, + "vllm_ttft_ms": 120.0, + "vllm_tpot_ms": 15.0, + "vllm_itl_ms": 14.0, + "vllm_itls_ms": [10.0, 14.0, 18.0], + } + } + for received_at_s, cumulative_audio_ms in ((10.2, 80), (10.25, 160), (10.36, 240)): + collector.add( + { + "type": "response.audio.delta", + "response_id": "resp-a", + "delta": base64.b64encode(b"audio").decode("ascii"), + "sample_rate_hz": 16_000, + "metadata": { + "audio_duration_ms": cumulative_audio_ms, + "vllm_omni": {"stage_metrics": stage_metrics}, + }, + }, + received_at_s=received_at_s, + ) + collector.add( + {"type": "response.audio_transcript.delta", "response_id": "resp-a", "delta": "hello"}, + received_at_s=10.15, + ) + + timing = collector.timing_summary( + after_s=10.0, + input_committed_at_s=9.9, + response_id="resp-a", + ) + + assert timing["request_metrics"] == { + "source": "server_request_start_and_client_receive", + "measurement_origin": { + "ttft": "native request start to first text output", + "ttfp": "native request start to first audio output", + "tpot": "Stage-0 engine mean time per output token", + }, + "ttft_ms": 125.0, + "ttfp_ms": 175.0, + "audio_generation_ms": 460.0, + "audio_duration_ms": 240.0, + "response_created_to_first_text_ms": 150.0, + "response_created_to_first_audio_ms": 200.0, + "tpot_ms": 15.0, } diff --git a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py index 22c8e6b34a6..821e505741c 100644 --- a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py +++ b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py @@ -1174,7 +1174,9 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity(): "ttft": "input_audio_buffer.commit client send to first non-empty text delta", "ttfp": "input_audio_buffer.commit client send to first audio packet", "rtf": "commit-to-last-audio receive time divided by emitted audio duration", + "tpot": "Stage-0 engine mean time per output token", }, + "tpot_ms": None, }, { "session_id": "seed-tts-session", @@ -1190,7 +1192,9 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity(): "ttft": "input_audio_buffer.commit client send to first non-empty text delta", "ttfp": "input_audio_buffer.commit client send to first audio packet", "rtf": "commit-to-last-audio receive time divided by emitted audio duration", + "tpot": "Stage-0 engine mean time per output token", }, + "tpot_ms": None, }, ] assert session_metrics == { diff --git a/tests/engine/test_output_processor.py b/tests/engine/test_output_processor.py index 62ed8334b66..351beb297a4 100644 --- a/tests/engine/test_output_processor.py +++ b/tests/engine/test_output_processor.py @@ -2,6 +2,7 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """Regression tests for OmniRequestState multimodal DELTA drain and consolidation guard.""" +from dataclasses import dataclass from types import SimpleNamespace from unittest.mock import MagicMock @@ -38,6 +39,21 @@ ) _LOGPROBS = MagicMock(logprobs=None, cumulative_logprob=None, prompt_logprobs=None) + +@dataclass +class _StreamingUpdate: + final: bool + prompt: object | None + prompt_token_ids: list[int] + arrival_time: float + + +@dataclass +class _FinishedRequestMetric: + request_id: str + mean_time_per_output_token: float + + _DEFAULT_STATE_KWARGS = dict( request_id="r", external_req_id="r", @@ -130,6 +146,90 @@ def update_native_stats(_output, _timestamp, _was_prefilling, native_stats, *_ar assert processor.pop_native_text_metrics("r")["num_generation_tokens"] == 27 +def test_native_text_metrics_exclude_cross_segment_gap(monkeypatch): + monkeypatch.setattr(VLLMOutputProcessor, "_update_stats_from_output", lambda *args, **kwargs: None) + processor = object.__new__(MultimodalOutputProcessor) + processor._native_text_metrics_by_request = {} + processor.lora_states = {} + state = _make_state(RequestOutputKind.DELTA) + iteration_stats = MagicMock() + + def update_segment(_output, timestamp, _was_prefilling, native_stats, *_args): + native_stats.num_generation_tokens = 2 if timestamp == 10.03 else 1 + native_stats.first_token_ts = 10.02 if timestamp == 10.03 else timestamp + native_stats.last_token_ts = timestamp + native_stats.first_token_latency = 0.01 + + iteration_stats.update_from_output.side_effect = update_segment + processor._update_stats_from_output( + state, + MagicMock(), + 10.0, + iteration_stats, + ) + first_segment = processor.pop_native_text_metrics("r") + assert first_segment["vllm_tpot_ms"] == 0.0 + assert first_segment["vllm_itls_ms"] == [] + + state.apply_streaming_update( + _StreamingUpdate( + final=False, + prompt=None, + prompt_token_ids=[], + arrival_time=10.01, + ) + ) + state.is_prefilling = True + processor._update_stats_from_output( + state, + MagicMock(), + 10.02, + iteration_stats, + ) + state.is_prefilling = False + processor._update_stats_from_output( + state, + MagicMock(), + 10.03, + iteration_stats, + ) + second_segment = processor.pop_native_text_metrics("r") + + assert second_segment["num_generation_tokens"] == 2 + assert second_segment["vllm_itls_ms"] == pytest.approx([10.0]) + assert second_segment["vllm_tpot_ms"] == pytest.approx(10.0) + + +@pytest.mark.parametrize( + ("finished_tpot_s", "expected_tpot_ms"), + [(0.0, 19.0), (0.023, 23.0)], +) +def test_native_text_tpot_only_accepts_positive_finished_metric( + monkeypatch, + finished_tpot_s, + expected_tpot_ms, +): + monkeypatch.setattr(VLLMOutputProcessor, "_update_stats_from_finished", lambda *args, **kwargs: None) + processor = object.__new__(MultimodalOutputProcessor) + processor._native_text_metrics_by_request = {"r": {"vllm_tpot_ms": 19.0}} + state = _make_state(RequestOutputKind.DELTA) + iteration_stats = MagicMock() + iteration_stats.finished_requests = [ + _FinishedRequestMetric( + request_id="r", + mean_time_per_output_token=finished_tpot_s, + ) + ] + + processor._update_stats_from_finished( + state, + FinishReason.STOP, + iteration_stats, + ) + + assert processor.pop_native_text_metrics("r")["vllm_tpot_ms"] == expected_tpot_ms + + def test_delta_drains_output_modality_per_step(): """DELTA drains the mm_type key (output modality) but preserves hidden-state keys.""" s = _make_state(RequestOutputKind.DELTA) diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index d3bbc5d01d2..c0e2bb5376e 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -8759,13 +8759,56 @@ async def emit(output: object) -> None: assert session.turn_id == 1 +def test_response_timing_binds_latest_request_start_for_model_turn(): + session = DuplexSession( + session_id="sid-response-request-timing", + config=DuplexSessionConfig(extra_body={"auto_response": True}), + ) + session.mark_model_turn_request_started(0, 10.0) + session.mark_model_turn_request_started(0, 11.0) + session.begin_response(turn_id=0) + + assert session.mark_response_first_outputs( + observed_at_s=11.2, + has_text=True, + has_audio=False, + )["ttft_ms"] == pytest.approx(200.0) + + session.mark_model_turn_request_started(0, 12.0) + assert session.mark_response_first_outputs( + observed_at_s=12.3, + has_text=False, + has_audio=True, + )["ttfp_ms"] == pytest.approx(1300.0) + + +def test_response_tpot_fallback_ignores_single_token_segment(): + session = DuplexSession( + session_id="sid-response-tpot", + config=DuplexSessionConfig(extra_body={"auto_response": True}), + ) + session.begin_response(turn_id=0) + + first_metrics = session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 1, "vllm_tpot_ms": 900.0}}) + assert "vllm_tpot_ms" not in first_metrics["0"] + + second_metrics = session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 2, "vllm_tpot_ms": 15.0}}) + assert second_metrics["0"]["num_tokens_out"] == 3 + assert second_metrics["0"]["vllm_tpot_ms"] == 15.0 + + @pytest.mark.asyncio -async def test_continuous_response_metrics_accumulate_only_owned_model_units(): +async def test_continuous_response_metrics_accumulate_only_owned_model_units(monkeypatch): handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient())) session = DuplexSession( session_id="sid-response-metrics", config=DuplexSessionConfig(extra_body={"auto_response": True}), ) + session.mark_model_turn_request_started(0, 10.0) + monkeypatch.setattr( + "vllm_omni.entrypoints.duplex.runtime_bridge.time.monotonic", + lambda: 10.25, + ) sent: list[dict[str, Any]] = [] async def send_json(payload: dict[str, Any]) -> None: @@ -8798,17 +8841,28 @@ async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list session=session, ) - await emit(text="first", audio="audio-a", tokens=4, ttft_ms=50.0, itls=[10.0, 11.0]) - await emit(text="second", audio="audio-b", tokens=6, ttft_ms=70.0, itls=[12.0, 13.0, 14.0]) + await emit(text="first", audio="audio-a", tokens=3, ttft_ms=50.0, itls=[10.0, 11.0]) + await emit(text="second", audio="audio-b", tokens=4, ttft_ms=70.0, itls=[12.0, 13.0, 14.0]) deltas = [payload for payload in sent if payload.get("type") == "response.output_audio.delta"] assert len(deltas) == 2 first_metrics = deltas[0]["vllm_omni"]["stage_metrics"]["0"] second_metrics = deltas[1]["vllm_omni"]["stage_metrics"]["0"] - assert first_metrics["num_tokens_out"] == 4 - assert second_metrics["num_tokens_out"] == 10 + assert first_metrics["num_tokens_out"] == 3 + assert first_metrics["vllm_tpot_ms"] == pytest.approx(10.5) + assert deltas[0]["vllm_omni"]["response_request_metrics"] == { + "source": "server_monotonic_request_start", + "measurement_origin": { + "ttft": "native model-turn request execution start to first non-empty text output", + "ttfp": "native model-turn request execution start to first audio output", + }, + "ttft_ms": 250.0, + "ttfp_ms": 250.0, + } + assert second_metrics["num_tokens_out"] == 7 assert second_metrics["vllm_ttft_ms"] == 50.0 assert second_metrics["vllm_itls_ms"] == [10.0, 11.0, 12.0, 13.0, 14.0] + assert second_metrics["vllm_tpot_ms"] == pytest.approx(12.0) first_response_id = deltas[0]["response_id"] session.end_response() @@ -8820,6 +8874,7 @@ async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list assert third_metrics["num_tokens_out"] == 3 assert third_metrics["vllm_ttft_ms"] == 90.0 assert third_metrics["vllm_itls_ms"] == [15.0, 16.0] + assert third_metrics["vllm_tpot_ms"] == pytest.approx(15.5) @pytest.mark.asyncio diff --git a/vllm_omni/benchmarks/omniinteract.py b/vllm_omni/benchmarks/omniinteract.py index b821d02356d..792a61e2bf2 100644 --- a/vllm_omni/benchmarks/omniinteract.py +++ b/vllm_omni/benchmarks/omniinteract.py @@ -962,8 +962,7 @@ def _populate_response_metrics( stream_start: float, ) -> None: measurement_origin = { - "ttft": "response.created client receive to first non-empty text delta", - "ttfp": "response.created client receive to first audio packet", + "tpot": "Stage-0 engine mean time per output token", "rtf": "response.created client receive to last audio packet divided by emitted audio duration", } request_metrics: list[dict[str, object]] = [] @@ -996,10 +995,35 @@ def _populate_response_metrics( result.duplex_request_metrics = request_metrics from vllm_omni.clients.duplex import summarize_session_request_metrics - result.duplex_session_metrics = summarize_session_request_metrics( + session_metrics = summarize_session_request_metrics( request_metrics, session_id=result.session_id, ) + global_metrics = collector.global_timing_summary( + after_s=stream_start, + window_started_at_s=stream_start, + response_ids=list(collector.response_ids), + measurement_origin={ + "ttft": "input stream start to first non-empty text delta", + "ttfp": "input stream start to first audio packet", + "rtf": ( + "input stream start-to-last-audio receive time divided by total emitted audio duration; " + "includes concurrent realtime input" + ), + }, + ) + if global_metrics: + session_metrics.update( + { + "global_ttft_ms": global_metrics.get("ttft_ms"), + "global_ttfp_ms": global_metrics.get("ttfp_ms"), + "global_rtf": _audio_rtf_from_raw_metric(global_metrics), + "global_audio_generation_ms": global_metrics.get("audio_generation_ms"), + "global_audio_duration_ms": global_metrics.get("audio_duration_ms"), + "global_measurement_origin": global_metrics.get("measurement_origin"), + } + ) + result.duplex_session_metrics = session_metrics async def run_omniinteract_case( diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index 3313e691f1f..c0fafa3da57 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -2998,6 +2998,21 @@ def measured_ttft(output: RequestFuncOutput) -> float | None: ] if duplex_session_metrics: result["duplex_session_metrics"] = duplex_session_metrics + for session_key, result_key in ( + ("global_ttft_ms", "mean_duplex_global_ttft_ms"), + ("global_ttfp_ms", "mean_duplex_global_ttfp_ms"), + ("global_rtf", "mean_duplex_global_rtf"), + ): + values = [ + float(value) + for metric in duplex_session_metrics + if isinstance((value := metric.get(session_key)), int | float) + and not isinstance(value, bool) + and np.isfinite(value) + and value >= 0 + ] + if values: + result[result_key] = sum(values) / len(values) if omniinteract_summary is not None: result["omniinteract"] = omniinteract_summary diff --git a/vllm_omni/clients/duplex.py b/vllm_omni/clients/duplex.py index eb8a6ded0d7..d6917b85f30 100644 --- a/vllm_omni/clients/duplex.py +++ b/vllm_omni/clients/duplex.py @@ -1250,6 +1250,15 @@ def _rounded_ms(value: float) -> float: return round(float(value), 3) +def _finite_number(value: object, *, nonnegative: bool = False) -> float | None: + if not isinstance(value, int | float) or isinstance(value, bool) or not math.isfinite(float(value)): + return None + number = float(value) + if nonnegative and number < 0: + return None + return number + + def _interval_summary(values: list[float]) -> dict[str, float | int]: clean = sorted(_rounded_ms(value) for value in values if math.isfinite(value) and value >= 0) if not clean: @@ -1287,6 +1296,34 @@ def _event_stage_metrics(event: dict[str, object]) -> dict[str, object] | None: return None +def _event_response_request_metrics(event: dict[str, object]) -> dict[str, object] | None: + direct_metrics = event.get("response_request_metrics") + if isinstance(direct_metrics, dict): + return direct_metrics + candidates: list[object] = [event.get("vllm_omni")] + metadata = event.get("metadata") + if isinstance(metadata, dict): + candidates.extend((metadata, metadata.get("vllm_omni"), metadata.get("duplex_event"))) + response = event.get("response") + if isinstance(response, dict): + response_metadata = response.get("metadata") + if isinstance(response_metadata, dict): + candidates.extend( + ( + response_metadata, + response_metadata.get("vllm_omni"), + response_metadata.get("duplex_event"), + ) + ) + for candidate in candidates: + if not isinstance(candidate, dict): + continue + request_metrics = candidate.get("response_request_metrics") + if isinstance(request_metrics, dict): + return request_metrics + return None + + class EventCollector: """Accumulate events for assertions and latency summaries (tests/benchmarks). @@ -1381,6 +1418,85 @@ def last_received_at(self, event_type: str) -> float | None: return received_at_s return None + def global_timing_summary( + self, + *, + after_s: float, + window_started_at_s: float, + response_ids: list[str], + measurement_origin: dict[str, str], + ) -> dict[str, object]: + """Summarize one client-observed timing window across responses. + + Returns raw measurements only. Derived metrics such as RTF are computed + by the caller — e.g. with ``vllm_omni.metrics.definitions.compute_audio_rtf``. + """ + selected_response_ids = set(response_ids) + first_text_received_at_s: float | None = None + audio_received_at_s: list[float] = [] + response_audio_duration_ms: dict[str, float] = {} + + for event, received_at_s in zip(self.events, self.event_received_at_s, strict=True): + if received_at_s < after_s: + continue + response_id = self.response_id(event) + if response_id is None or response_id not in selected_response_ids: + continue + if ( + event.get("type") + in { + "response.audio_transcript.delta", + "response.output_text.delta", + "response.text.delta", + } + and isinstance(event.get("delta"), str) + and bool(event["delta"]) + and first_text_received_at_s is None + ): + first_text_received_at_s = received_at_s + if event.get("type") not in _AUDIO_DELTA_TYPES: + continue + delta = event.get("delta") or event.get("audio") + if not isinstance(delta, str) or not delta: + continue + audio_received_at_s.append(received_at_s) + metadata = event.get("metadata") + duration_ms = metadata.get("audio_duration_ms") if isinstance(metadata, dict) else None + parsed_duration_ms = _finite_number(duration_ms, nonnegative=True) + if parsed_duration_ms is not None: + response_audio_duration_ms[response_id] = max( + response_audio_duration_ms.get(response_id, 0.0), + parsed_duration_ms, + ) + + if not audio_received_at_s: + return {} + + audio_duration_ms = sum( + response_audio_duration_ms.get( + response_id, + self._format_for_bytes().duration_ms(len(self.audio_bytes(response_id))), + ) + for response_id in response_ids + ) + audio_generation_ms = max( + 0.0, + (audio_received_at_s[-1] - window_started_at_s) * 1000.0, + ) + return { + "source": "client_monotonic_receive", + "response_ids": list(response_ids), + "measurement_origin": dict(measurement_origin), + "ttft_ms": ( + _rounded_ms((first_text_received_at_s - window_started_at_s) * 1000.0) + if first_text_received_at_s is not None + else None + ), + "ttfp_ms": _rounded_ms((audio_received_at_s[0] - window_started_at_s) * 1000.0), + "audio_generation_ms": _rounded_ms(audio_generation_ms), + "audio_duration_ms": _rounded_ms(audio_duration_ms), + } + def timing_summary( self, *, @@ -1391,6 +1507,7 @@ def timing_summary( ) -> dict[str, object]: """Summarize engine token metrics and client-observed audio cadence.""" stage0_metrics: dict[str, object] | None = None + response_request_metrics: dict[str, object] = {} response_created_at_s: float | None = None first_text_received_at_s: float | None = None audio_received_at_s: list[float] = [] @@ -1421,6 +1538,10 @@ def timing_summary( if isinstance(stage0, dict): stage0_metrics = stage0 + event_request_metrics = _event_response_request_metrics(event) + if event_request_metrics is not None: + response_request_metrics.update(event_request_metrics) + if event.get("type") not in _AUDIO_DELTA_TYPES: continue delta = event.get("delta") or event.get("audio") @@ -1480,6 +1601,31 @@ def timing_summary( } request_started_at_s = input_committed_at_s if input_committed_at_s is not None else response_created_at_s if request_started_at_s is not None: + default_measurement_origin = ( + { + "ttft": "input_audio_buffer.commit client send to first non-empty text delta", + "ttfp": "input_audio_buffer.commit client send to first audio packet", + } + if input_committed_at_s is not None + else { + "ttft": "response.created client receive to first non-empty text delta", + "ttfp": "response.created client receive to first audio packet", + } + ) + resolved_measurement_origin = { + **default_measurement_origin, + **(measurement_origin or {}), + } + if stage0_metrics is not None: + resolved_measurement_origin["tpot"] = "Stage-0 engine mean time per output token" + server_ttft_ms = _finite_number(response_request_metrics.get("ttft_ms"), nonnegative=True) + server_ttfp_ms = _finite_number(response_request_metrics.get("ttfp_ms"), nonnegative=True) + server_origins = response_request_metrics.get("measurement_origin") + if isinstance(server_origins, dict): + if server_ttft_ms is not None and isinstance(server_origins.get("ttft"), str): + resolved_measurement_origin["ttft"] = server_origins["ttft"] + if server_ttfp_ms is not None and isinstance(server_origins.get("ttfp"), str): + resolved_measurement_origin["ttfp"] = server_origins["ttfp"] audio_duration_ms = ( max(cumulative_audio_ms) if cumulative_audio_ms @@ -1495,21 +1641,49 @@ def timing_summary( # compute_audio_rtf — so metric definitions stay out of this # dependency-free client module. result["request_metrics"] = { - "source": "client_monotonic_receive", - "measurement_origin": measurement_origin - or { - "ttft": "input_audio_buffer.commit client send to first non-empty text delta", - "ttfp": "input_audio_buffer.commit client send to first audio packet", - }, + "source": ( + "server_request_start_and_client_receive" + if server_ttft_ms is not None or server_ttfp_ms is not None + else "client_monotonic_receive" + ), + "measurement_origin": resolved_measurement_origin, "ttft_ms": ( - _rounded_ms((first_text_received_at_s - request_started_at_s) * 1000.0) - if first_text_received_at_s is not None - else None + server_ttft_ms + if server_ttft_ms is not None + else ( + _rounded_ms((first_text_received_at_s - request_started_at_s) * 1000.0) + if first_text_received_at_s is not None + else None + ) + ), + "ttfp_ms": ( + server_ttfp_ms + if server_ttfp_ms is not None + else _rounded_ms((audio_received_at_s[0] - request_started_at_s) * 1000.0) ), - "ttfp_ms": _rounded_ms((audio_received_at_s[0] - request_started_at_s) * 1000.0), "audio_generation_ms": _rounded_ms(audio_generation_ms), "audio_duration_ms": _rounded_ms(audio_duration_ms), } + if server_ttft_ms is not None or server_ttfp_ms is not None: + request_metrics = result["request_metrics"] + assert isinstance(request_metrics, dict) + request_metrics["response_created_to_first_text_ms"] = ( + _rounded_ms((first_text_received_at_s - response_created_at_s) * 1000.0) + if first_text_received_at_s is not None and response_created_at_s is not None + else None + ) + request_metrics["response_created_to_first_audio_ms"] = ( + _rounded_ms((audio_received_at_s[0] - response_created_at_s) * 1000.0) + if response_created_at_s is not None + else None + ) + if stage0_metrics is not None: + request_metrics = result["request_metrics"] + assert isinstance(request_metrics, dict) + request_metrics["tpot_ms"] = _finite_number( + stage0_metrics.get("vllm_tpot_ms"), + nonnegative=True, + ) return result def _format_for_bytes(self) -> AudioFormat: @@ -1642,30 +1816,37 @@ def summarize_session_request_metrics( *, session_id: str | None, ) -> dict[str, object]: - """Average client-observed metrics across turns that emitted audio. + """Average client- and engine-observed metrics across audio responses. ``request_metrics`` entries are caller-assembled dicts; keys that are absent or non-numeric in an entry are simply skipped. ``rtf`` is not produced by :meth:`EventCollector.timing_summary` (which reports raw data only) — callers that want ``mean_rtf`` add an ``rtf`` value per turn, - e.g. via ``vllm_omni.metrics.definitions.compute_audio_rtf``. + e.g. via ``vllm_omni.metrics.definitions.compute_audio_rtf``. Zero or + missing ``tpot_ms`` values are omitted from ``mean_tpot_ms``. """ - def mean(metric: str, *, digits: int = 3) -> float | None: + def mean(metric: str, *, digits: int = 3, positive: bool = False) -> float | None: values = [ float(request[metric]) for request in request_metrics - if isinstance(request.get(metric), int | float) and math.isfinite(float(request[metric])) + if isinstance(request.get(metric), int | float) + and not isinstance(request.get(metric), bool) + and math.isfinite(float(request[metric])) + and (not positive or float(request[metric]) > 0) ] return round(sum(values) / len(values), digits) if values else None - return { + summary: dict[str, object] = { "session_id": session_id, "audio_turn_count": len(request_metrics), "mean_ttft_ms": mean("ttft_ms"), "mean_ttfp_ms": mean("ttfp_ms"), "mean_rtf": mean("rtf", digits=6), } + if (mean_tpot_ms := mean("tpot_ms", positive=True)) is not None: + summary["mean_tpot_ms"] = mean_tpot_ms + return summary async def acknowledge_collected_playback(client: DuplexClient, collector: EventCollector) -> None: diff --git a/vllm_omni/entrypoints/duplex/protocol.py b/vllm_omni/entrypoints/duplex/protocol.py index ff91e0db58e..1f5a2bab494 100644 --- a/vllm_omni/entrypoints/duplex/protocol.py +++ b/vllm_omni/entrypoints/duplex/protocol.py @@ -411,6 +411,10 @@ class ResponseState: active_request_id: str | None = None active_response_id: str | None = None active_response_turn_id: int | None = None + request_started_at_s_by_turn: dict[int, float] = field(default_factory=dict) + active_response_request_started_at_s: float | None = None + active_response_ttft_ms: float | None = None + active_response_ttfp_ms: float | None = None active_response_input_commit_seq: int | None = None active_response_awaits_input_commit: bool = False last_response_id: str | None = None @@ -654,6 +658,45 @@ def clear_request(self, expected_request_id: str | None = None) -> bool: def bind_response_turn(self, turn_id: int | None) -> None: self._response.active_response_turn_id = turn_id + def mark_model_turn_request_started(self, turn_id: int, started_at_s: float) -> None: + """Record the latest native request start that can own one model turn.""" + turn_id = int(turn_id) + started_at_s = float(started_at_s) + if self.active_response_turn_id == turn_id: + if self._response.active_response_request_started_at_s is None: + self._response.active_response_request_started_at_s = started_at_s + return + self._response.request_started_at_s_by_turn[turn_id] = started_at_s + + def mark_response_first_outputs( + self, + *, + observed_at_s: float, + has_text: bool, + has_audio: bool, + ) -> dict[str, object]: + """Return server-monotonic TTF metrics observed for the active response.""" + started_at_s = self._response.active_response_request_started_at_s + if started_at_s is None: + return {} + elapsed_ms = max(0.0, (float(observed_at_s) - started_at_s) * 1000.0) + if has_text and self._response.active_response_ttft_ms is None: + self._response.active_response_ttft_ms = elapsed_ms + if has_audio and self._response.active_response_ttfp_ms is None: + self._response.active_response_ttfp_ms = elapsed_ms + metrics: dict[str, object] = { + "source": "server_monotonic_request_start", + "measurement_origin": { + "ttft": "native model-turn request execution start to first non-empty text output", + "ttfp": "native model-turn request execution start to first audio output", + }, + } + if self._response.active_response_ttft_ms is not None: + metrics["ttft_ms"] = self._response.active_response_ttft_ms + if self._response.active_response_ttfp_ms is not None: + metrics["ttfp_ms"] = self._response.active_response_ttfp_ms + return metrics + def active_response_accepts_model_turn(self, turn_id: int | None) -> bool: if self._response.active_response_id is None: return False @@ -938,6 +981,11 @@ def commit_native_audio_input( def complete_model_turn(self, turn_id: int) -> None: """Advance the model-owned output identity after its terminal signal.""" completed_turn_id = int(turn_id) + self._response.request_started_at_s_by_turn = { + pending_turn_id: started_at_s + for pending_turn_id, started_at_s in self._response.request_started_at_s_by_turn.items() + if pending_turn_id > completed_turn_id + } if completed_turn_id >= self.turn_id: self.turn_id = completed_turn_id + 1 @@ -974,6 +1022,12 @@ def begin_response(self, *, turn_id: int | None = None) -> str: response_id = f"resp-{self.session_id}-{self.epoch}-{uuid4().hex[:8]}" self._response.active_response_id = response_id self._response.active_response_turn_id = self.turn_id if turn_id is None else int(turn_id) + self._response.active_response_request_started_at_s = self._response.request_started_at_s_by_turn.pop( + self._response.active_response_turn_id, + None, + ) + self._response.active_response_ttft_ms = None + self._response.active_response_ttfp_ms = None self._response.active_response_input_commit_seq = self.input_commit_seq self._response.active_response_awaits_input_commit = self.turn_state == DuplexTurnState.USER_SPEAKING self._response.last_response_id = response_id @@ -1055,17 +1109,25 @@ def accumulate_response_stage_metrics( tpot_ms = raw_values.get("vllm_tpot_ms") token_count = raw_values.get("num_tokens_out") - if isinstance(tpot_ms, int | float) and tpot_ms > 0: - weight = max(int(token_count) - 1, 1) if isinstance(token_count, int | float) else 1 - self._response.stage_metric_tpot_weighted_ms[stage_id] = ( - self._response.stage_metric_tpot_weighted_ms.get(stage_id, 0.0) + float(tpot_ms) * weight - ) - self._response.stage_metric_tpot_weight[stage_id] = ( - self._response.stage_metric_tpot_weight.get(stage_id, 0) + weight - ) - current["vllm_tpot_ms"] = self._response.stage_metric_tpot_weighted_ms[stage_id] / float( - self._response.stage_metric_tpot_weight[stage_id] + response_itls = current.get("vllm_itls_ms") + if isinstance(response_itls, list) and response_itls: + current["vllm_tpot_ms"] = current["vllm_itl_ms"] + elif isinstance(tpot_ms, int | float) and not isinstance(tpot_ms, bool) and tpot_ms > 0: + weight = ( + max(int(token_count) - 1, 0) + if isinstance(token_count, int | float) and not isinstance(token_count, bool) + else 1 ) + if weight > 0: + self._response.stage_metric_tpot_weighted_ms[stage_id] = ( + self._response.stage_metric_tpot_weighted_ms.get(stage_id, 0.0) + float(tpot_ms) * weight + ) + self._response.stage_metric_tpot_weight[stage_id] = ( + self._response.stage_metric_tpot_weight.get(stage_id, 0) + weight + ) + current["vllm_tpot_ms"] = self._response.stage_metric_tpot_weighted_ms[stage_id] / float( + self._response.stage_metric_tpot_weight[stage_id] + ) for name, value in raw_values.items(): if name not in handled_fields: @@ -1240,6 +1302,9 @@ def end_response( self._response.active_request_id = None self._response.active_response_id = None self._response.active_response_turn_id = None + self._response.active_response_request_started_at_s = None + self._response.active_response_ttft_ms = None + self._response.active_response_ttfp_ms = None self._response.active_response_input_commit_seq = None self._response.active_response_awaits_input_commit = False self._clear_response_metrics() @@ -1557,6 +1622,10 @@ def barge_in(self) -> int: self._response.active_request_id = None self._response.active_response_id = None self._response.active_response_turn_id = None + self._response.request_started_at_s_by_turn.clear() + self._response.active_response_request_started_at_s = None + self._response.active_response_ttft_ms = None + self._response.active_response_ttfp_ms = None self._response.active_response_input_commit_seq = None self._response.active_response_awaits_input_commit = False self._clear_response_metrics() @@ -1572,6 +1641,10 @@ def close(self) -> None: self.state = DuplexSessionState.CLOSED self.turn_state = DuplexTurnState.IDLE self._response.active_response_turn_id = None + self._response.request_started_at_s_by_turn.clear() + self._response.active_response_request_started_at_s = None + self._response.active_response_ttft_ms = None + self._response.active_response_ttfp_ms = None self._response.active_response_input_commit_seq = None self._response.active_response_awaits_input_commit = False self.clear_server_vad_audio() diff --git a/vllm_omni/entrypoints/duplex/runtime_bridge.py b/vllm_omni/entrypoints/duplex/runtime_bridge.py index bf839523ce5..96a82c06191 100644 --- a/vllm_omni/entrypoints/duplex/runtime_bridge.py +++ b/vllm_omni/entrypoints/duplex/runtime_bridge.py @@ -6,6 +6,7 @@ import asyncio import base64 import inspect +import time from collections.abc import Mapping from dataclasses import asdict, is_dataclass from enum import Enum @@ -1113,13 +1114,20 @@ async def _send_one_native_duplex_event( if response_id is None: response_id = session.begin_response(turn_id=model_turn_id) response_created = True - await send_json( - self._response_created_payload( - session, - response_id, - epoch=session.epoch, - ) + response_request_metrics = session.mark_response_first_outputs( + observed_at_s=time.monotonic(), + has_text=has_text, + has_audio=has_audio, + ) + if response_created: + response_created_payload = self._response_created_payload( + session, + response_id, + epoch=session.epoch, ) + if response_request_metrics: + response_created_payload["response_request_metrics"] = response_request_metrics + await send_json(response_created_payload) response_stage_metrics = session.accumulate_response_stage_metrics( native_result.get("stage_metrics") if isinstance(native_result.get("stage_metrics"), Mapping) else None ) @@ -1137,6 +1145,7 @@ async def _send_one_native_duplex_event( speak_payload, native_result, stage_metrics=response_stage_metrics, + response_request_metrics=response_request_metrics, ) await send_json(speak_payload) previous_sent_ms = session.playback.sent_ms @@ -1205,6 +1214,7 @@ async def _send_one_native_duplex_event( payload, native_result, stage_metrics=response_stage_metrics, + response_request_metrics=response_request_metrics, ) await send_json(payload) if ( @@ -1365,6 +1375,7 @@ def _attach_native_runtime_metadata( native_result: dict[str, object], *, stage_metrics: Mapping[str, object] | None = None, + response_request_metrics: Mapping[str, object] | None = None, ) -> None: metadata: dict[str, object] = {} runtime_impl = native_result.get("runtime_impl") @@ -1390,6 +1401,8 @@ def _attach_native_runtime_metadata( for stage_id, values in effective_stage_metrics.items() if isinstance(values, Mapping) } + if response_request_metrics: + metadata["response_request_metrics"] = dict(response_request_metrics) if metadata: payload["vllm_omni"] = metadata diff --git a/vllm_omni/entrypoints/duplex/session_runner.py b/vllm_omni/entrypoints/duplex/session_runner.py index ed18d859105..d870e4cfe18 100644 --- a/vllm_omni/entrypoints/duplex/session_runner.py +++ b/vllm_omni/entrypoints/duplex/session_runner.py @@ -6,6 +6,7 @@ import asyncio import binascii import json +import time import uuid from contextlib import suppress from copy import deepcopy @@ -483,6 +484,7 @@ def _discard_retained_committed_audio() -> None: async def _run() -> bool: nonlocal runtime_closed try: + session.mark_model_turn_request_started(append_turn_id, time.monotonic()) append_ok, emitted_response = await self._append_runtime_input( session, payload, diff --git a/vllm_omni/outputs/output_processor.py b/vllm_omni/outputs/output_processor.py index 13844518e7b..cd65a900f48 100644 --- a/vllm_omni/outputs/output_processor.py +++ b/vllm_omni/outputs/output_processor.py @@ -808,7 +808,9 @@ def _update_stats_from_output( itls_ms = record.setdefault("vllm_itls_ms", []) itls_ms.append(itl_ms) record["vllm_itl_ms"] = sum(itls_ms) / float(len(itls_ms)) - record["vllm_tpot_ms"] = _mean_time_per_output_token_ms(native_stats) + record["vllm_tpot_ms"] = ( + record["vllm_itl_ms"] if record["vllm_itls_ms"] else _mean_time_per_output_token_ms(native_stats) + ) def _update_stats_from_finished( self, @@ -836,6 +838,7 @@ def _update_stats_from_finished( finished_request = iteration_stats.finished_requests[-1] if finished_request.request_id != req_state.external_req_id: return - self._native_text_metric_record(req_state.external_req_id)["vllm_tpot_ms"] = ( - float(finished_request.mean_time_per_output_token) * 1000.0 - ) + finished_tpot_ms = float(finished_request.mean_time_per_output_token) * 1000.0 + record = self._native_text_metric_record(req_state.external_req_id) + if finished_tpot_ms > 0 and not record.get("vllm_itls_ms"): + record["vllm_tpot_ms"] = finished_tpot_ms From 8b3d34d3e3db88d3d7eac3656957c9e2a7919c91 Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Thu, 10 Sep 2026 06:41:20 +0000 Subject: [PATCH 2/7] [Frontend][Benchmark] duplex performance metrics for Omni-DuplexEval Signed-off-by: ZacheryAU --- .../fullduplex/test_omni_duplex_eval_cli.py | 81 ++++++++++++++++++- .../duplex/omni_duplex_eval_runner.py | 66 +++++++++++++-- vllm_omni/benchmarks/omniinteract.py | 77 +++--------------- .../cli/benchmark/omni_duplex_eval.py | 43 ++++++++-- 4 files changed, 182 insertions(+), 85 deletions(-) diff --git a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py index 192859de5e5..6c68ed363cc 100644 --- a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py +++ b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py @@ -15,6 +15,13 @@ from vllm_omni.benchmarks.duplex import omni_duplex_eval_runner as runner from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge +from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult +from vllm_omni.benchmarks.duplex_session_metrics import ( + DUPLEX_METRICS_FILENAME, + build_duplex_metrics_report, + collect_duplex_session_metrics, +) +from vllm_omni.clients.duplex import EventCollector from vllm_omni.entrypoints.cli.benchmark import omni_duplex_eval as cli from vllm_omni.entrypoints.cli.benchmark.omni_duplex_eval import OmniDuplexEvalSubcommand @@ -44,7 +51,25 @@ async def fake_generate(sample, *, output_root, **kwargs): output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps([{"sentence": "It moved.", "start": 0, "end": 1}]), encoding="utf-8") output.with_name(output.stem + ".meta.json").write_text(json.dumps({"clock": "media"}), encoding="utf-8") - return output + return GenerateSampleResult( + output=output, + request_metrics=[ + { + "sample_id": sample.id, + "split": sample.split, + "ttft_ms": 10.0, + "ttfp_ms": 20.0, + "rtf": 0.5, + } + ], + session_metrics={ + "sample_id": sample.id, + "split": sample.split, + "global_ttft_ms": 10.0, + "global_ttfp_ms": 20.0, + "global_rtf": 0.5, + }, + ) class FakeJudge: def __init__(self, *args, **kwargs): @@ -75,6 +100,19 @@ def chat(self, *args, **kwargs): ] ) OmniDuplexEvalSubcommand.cmd(generate) + duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8")) + assert duplex_metrics["duplex_request_metrics"] == [ + { + "sample_id": "sample-1", + "split": "PR_correction", + "ttft_ms": 10.0, + "ttfp_ms": 20.0, + "rtf": 0.5, + } + ] + assert duplex_metrics["mean_duplex_global_ttft_ms"] == 10.0 + assert duplex_metrics["mean_duplex_global_ttfp_ms"] == 20.0 + assert duplex_metrics["mean_duplex_global_rtf"] == 0.5 evaluate = parser.parse_args( [ "evaluate", @@ -144,7 +182,7 @@ async def handler(websocket): async with websockets.serve(handler, "127.0.0.1", 0) as server: port = server.sockets[0].getsockname()[1] - output = await runner.generate_sample( + result = await runner.generate_sample( sample, url=f"ws://127.0.0.1:{port}/v1/realtime?duplex=1", model="mock", @@ -152,13 +190,48 @@ async def handler(websocket): output_root=tmp_path / "responses", ) - assert json.loads(output.read_text(encoding="utf-8")) == [{"sentence": "Done.", "start": 0.8, "end": 0.8}] - meta = json.loads(output.with_name("sample.meta.json").read_text(encoding="utf-8")) + assert json.loads(result.output.read_text(encoding="utf-8")) == [{"sentence": "Done.", "start": 0.8, "end": 0.8}] + meta = json.loads(result.output.with_name("sample.meta.json").read_text(encoding="utf-8")) assert meta["response_done"] is True assert meta["drain_timeout"] is None + assert "duplex_request_metrics" not in meta + assert result.request_metrics + assert result.request_metrics[0]["sample_id"] == "sample" + assert result.session_metrics["global_ttfp_ms"] is not None assert any(event["type"] == "playback.ack" for event in received) +def test_collect_duplex_session_metrics_matches_omniinteract_window(): + collector = EventCollector() + audio = { + "type": "response.audio.delta", + "response_id": "r1", + "format": "pcm16", + "delta": base64.b64encode(bytes((1, 0)) * 2400).decode(), + "sample_rate_hz": 24_000, + "metadata": {"audio_duration_ms": 100}, + } + collector.add({"type": "response.created", "response": {"id": "r1"}}, received_at_s=10.1) + collector.add( + {"type": "response.output_text.delta", "response_id": "r1", "delta": "Done."}, + received_at_s=10.2, + ) + collector.add(audio, received_at_s=10.3) + collector.add({"type": "response.done", "response": {"id": "r1"}}, received_at_s=10.4) + bundle = collect_duplex_session_metrics(collector, stream_start=10.0, session_id="sample") + assert bundle.request_metrics[0]["ttft_ms"] == 100.0 + assert bundle.request_metrics[0]["ttfp_ms"] == 200.0 + assert bundle.request_metrics[0]["rtf"] == 2.0 + assert bundle.session_metrics["global_ttft_ms"] == 200.0 + assert bundle.session_metrics["global_ttfp_ms"] == 300.0 + report = build_duplex_metrics_report( + request_metrics=bundle.request_metrics, + session_metrics=[bundle.session_metrics], + ) + assert report["mean_duplex_global_ttft_ms"] == 200.0 + assert report["mean_duplex_global_ttfp_ms"] == 300.0 + + def test_judge_exercises_openai_http_schema(): requests = [] diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py index d5a28aa063f..db500ebf063 100644 --- a/vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py +++ b/vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py @@ -8,24 +8,58 @@ import asyncio import hashlib import json +import time +from dataclasses import dataclass, field from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING import pybase64 as base64 -from vllm_omni.experimental.fullduplex.client import RealtimeDuplexClient, build_realtime_url, wait_for +from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample +from vllm_omni.experimental.fullduplex.client import ( + RealtimeDuplexClient, + RealtimeEventCollector, + build_realtime_url, + wait_for, +) + +if TYPE_CHECKING: + from vllm_omni.clients.duplex import EventCollector from .omni_duplex_eval_clock import extract_timed_sentences from .omni_duplex_eval_media import iter_av_units, iter_jpegs, materialize_media, read_audio_pcm16, video_duration +@dataclass(frozen=True) +class GenerateSampleResult: + """Timed-sentence path plus duplex metrics for one generate sample.""" + + output: Path + request_metrics: list[dict[str, object]] = field(default_factory=list) + session_metrics: dict[str, object] = field(default_factory=dict) + + def _ref_audio(path: str | Path) -> str: value = Path(path).expanduser().read_bytes() return "data:audio/wav;base64," + base64.b64encode(value).decode("ascii") +def _event_collector_from_realtime(source: RealtimeEventCollector) -> EventCollector: + """Replay experimental collector events onto the public EventCollector.""" + from vllm_omni.clients.duplex import EventCollector + + collector = EventCollector() + for event, received_at_s in zip(source.events, source.event_received_at_s, strict=True): + collector.add(event, received_at_s=received_at_s) + return collector + + +def _with_sample_identity(payload: dict[str, object], sample: DuplexSample) -> dict[str, object]: + return {"sample_id": sample.id, "split": sample.split, **payload} + + async def generate_sample( - sample: Any, + sample: DuplexSample, *, url: str, model: str, @@ -37,11 +71,17 @@ async def generate_sample( clock: str = "media", overwrite: bool = False, unit_ms: int = 1000, -) -> Path: +) -> GenerateSampleResult: + """Generate one sample's timed sentences and collect duplex session metrics. + + Timed sentences and ``*.meta.json`` stay the quality-eval artifacts. Duplex + request/session metrics are returned for the CLI to write + ``duplex_metrics.json``; they are not written into meta. + """ output = Path(output_root) / sample.split / f"{sample.id}.json" meta_path = output.with_name(output.stem + ".meta.json") if output.exists() and not overwrite: - return output + return GenerateSampleResult(output=output) if mix != "question": raise NotImplementedError("v1 supports mix=question; soundtrack mixing is reserved for P1") media_dir = output.parent / ".media" @@ -57,10 +97,12 @@ async def generate_sample( response_done = False drain_timeout = None close_timeout = None + stream_start: float | None = None async with client: await client.configure(model, ref_audio=_ref_audio(ref_audio), instructions="Streaming Omni Conversation.") ack_task = asyncio.create_task(_ack_playback(client)) try: + stream_start = time.monotonic() await client.stream_av_units(iter_av_units(pcm, frames, unit_ms=unit_ms), realtime=realtime) await client.commit() try: @@ -84,6 +126,7 @@ async def generate_sample( except TimeoutError as exc: close_timeout = str(exc) events = list(client.events.events) + collector = _event_collector_from_realtime(client.events) timed = [sentence.as_dict() for sentence in extract_timed_sentences(events, clock=clock)] output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(timed, ensure_ascii=False, indent=2), encoding="utf-8") @@ -102,7 +145,18 @@ async def generate_sample( "ref_audio_sha256": hashlib.sha256(Path(ref_audio).read_bytes()).hexdigest(), } meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8") - return output + from vllm_omni.benchmarks.duplex_session_metrics import collect_duplex_session_metrics + + bundle = collect_duplex_session_metrics( + collector, + stream_start=stream_start if stream_start is not None else 0.0, + session_id=sample.id, + ) + return GenerateSampleResult( + output=output, + request_metrics=[_with_sample_identity(metric, sample) for metric in bundle.request_metrics], + session_metrics=_with_sample_identity(bundle.session_metrics, sample), + ) async def _ack_playback(client: RealtimeDuplexClient) -> None: diff --git a/vllm_omni/benchmarks/omniinteract.py b/vllm_omni/benchmarks/omniinteract.py index 792a61e2bf2..60790abb976 100644 --- a/vllm_omni/benchmarks/omniinteract.py +++ b/vllm_omni/benchmarks/omniinteract.py @@ -946,13 +946,9 @@ async def close_session(self, *, timeout_s: float = 20.0) -> None: def _audio_rtf_from_raw_metric(raw_metric: dict[str, object]) -> float | None: """Derive the audio RTF from a client-reported raw request metric.""" - from vllm_omni.metrics.definitions import compute_audio_rtf + from vllm_omni.benchmarks.duplex_session_metrics import audio_rtf_from_raw_metric - generation_ms = raw_metric.get("audio_generation_ms") - duration_ms = raw_metric.get("audio_duration_ms") - if not isinstance(generation_ms, int | float) or not isinstance(duration_ms, int | float) or duration_ms <= 0: - return None - return round(compute_audio_rtf(float(generation_ms) / 1000.0, float(duration_ms) / 1000.0), 6) + return audio_rtf_from_raw_metric(raw_metric) def _populate_response_metrics( @@ -961,69 +957,16 @@ def _populate_response_metrics( *, stream_start: float, ) -> None: - measurement_origin = { - "tpot": "Stage-0 engine mean time per output token", - "rtf": "response.created client receive to last audio packet divided by emitted audio duration", - } - request_metrics: list[dict[str, object]] = [] - output_tokens = 0 - for request_index, response_id in enumerate(collector.response_ids): - timing = collector.timing_summary( - after_s=stream_start, - input_committed_at_s=None, - response_id=response_id, - measurement_origin=measurement_origin, - ) - raw_metric = timing.get("request_metrics") - stage0 = timing.get("stage0_tokens") - metric = { - "session_id": result.session_id, - "request_index": request_index, - "response_id": response_id, - **(raw_metric if isinstance(raw_metric, dict) else {}), - } - if isinstance(raw_metric, dict): - # The client reports raw data only; the RTF is derived here with - # the canonical server-side metric definition. - metric["rtf"] = _audio_rtf_from_raw_metric(raw_metric) - if isinstance(stage0, dict): - metric["stage0_tokens"] = dict(stage0) - output_tokens += int(stage0.get("output_token_count") or 0) - if isinstance(raw_metric, dict) or isinstance(stage0, dict): - request_metrics.append(metric) - result.output_tokens = output_tokens - result.duplex_request_metrics = request_metrics - from vllm_omni.clients.duplex import summarize_session_request_metrics - - session_metrics = summarize_session_request_metrics( - request_metrics, + from vllm_omni.benchmarks.duplex_session_metrics import collect_duplex_session_metrics + + bundle = collect_duplex_session_metrics( + collector, + stream_start=stream_start, session_id=result.session_id, ) - global_metrics = collector.global_timing_summary( - after_s=stream_start, - window_started_at_s=stream_start, - response_ids=list(collector.response_ids), - measurement_origin={ - "ttft": "input stream start to first non-empty text delta", - "ttfp": "input stream start to first audio packet", - "rtf": ( - "input stream start-to-last-audio receive time divided by total emitted audio duration; " - "includes concurrent realtime input" - ), - }, - ) - if global_metrics: - session_metrics.update( - { - "global_ttft_ms": global_metrics.get("ttft_ms"), - "global_ttfp_ms": global_metrics.get("ttfp_ms"), - "global_rtf": _audio_rtf_from_raw_metric(global_metrics), - "global_audio_generation_ms": global_metrics.get("audio_generation_ms"), - "global_audio_duration_ms": global_metrics.get("audio_duration_ms"), - "global_measurement_origin": global_metrics.get("measurement_origin"), - } - ) - result.duplex_session_metrics = session_metrics + result.output_tokens = bundle.output_tokens + result.duplex_request_metrics = bundle.request_metrics + result.duplex_session_metrics = bundle.session_metrics async def run_omniinteract_case( diff --git a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py index a43a5ec71e9..761b9c04f42 100644 --- a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py +++ b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py @@ -1,16 +1,24 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""CLI for Omni-DuplexEval generate / evaluate / summarize. + +``generate`` still writes per-sample timed sentences and ``*.meta.json``. +Duplex performance metrics for the run are written once to +``/duplex_metrics.json``. +""" + import argparse import asyncio import concurrent.futures import json from pathlib import Path -from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DEFAULT_DATASET, load_samples +from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DEFAULT_DATASET, DuplexSample, load_samples from vllm_omni.benchmarks.duplex.omni_duplex_eval_eval import evaluate_sample, summarize_scores from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge -from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import generate_sample +from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult, generate_sample +from vllm_omni.benchmarks.duplex_session_metrics import DUPLEX_METRICS_FILENAME, build_duplex_metrics_report from vllm_omni.entrypoints.cli.benchmark.base import OmniBenchmarkSubcommandBase @@ -54,6 +62,25 @@ def add_cli_args(parser: argparse.ArgumentParser) -> None: summarize.add_argument("--score-root", required=True) +def _write_duplex_metrics(response_root: str | Path, results: list[GenerateSampleResult]) -> Path: + request_metrics = [metric for result in results for metric in result.request_metrics] + session_metrics = [result.session_metrics for result in results if result.session_metrics] + path = Path(response_root) / DUPLEX_METRICS_FILENAME + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + build_duplex_metrics_report( + request_metrics=request_metrics, + session_metrics=session_metrics, + ), + ensure_ascii=False, + indent=2, + ), + encoding="utf-8", + ) + return path + + def run(args: argparse.Namespace) -> int: if args.action == "summarize": print(json.dumps(summarize_scores(args.score_root), ensure_ascii=False, indent=2)) @@ -71,12 +98,12 @@ def run(args: argparse.Namespace) -> int: if args.concurrency < 1: raise ValueError("--concurrency must be at least 1") - async def generate() -> None: + async def generate() -> list[GenerateSampleResult]: semaphore = asyncio.Semaphore(args.concurrency) - async def generate_one(sample) -> None: + async def generate_one(sample: DuplexSample) -> GenerateSampleResult: async with semaphore: - await generate_sample( + return await generate_sample( sample, url=args.url, model=args.model, @@ -89,16 +116,16 @@ async def generate_one(sample) -> None: overwrite=args.overwrite, ) - await asyncio.gather(*(generate_one(sample) for sample in samples)) + return list(await asyncio.gather(*(generate_one(sample) for sample in samples))) - asyncio.run(generate()) + _write_duplex_metrics(args.response_root, asyncio.run(generate())) return 0 if args.eval_workers < 1: raise ValueError("--eval-workers must be at least 1") judge = DuplexJudge(args.judge_base_url, args.judge_model, api_key=args.judge_api_key) - def evaluate_one(sample) -> None: + def evaluate_one(sample: DuplexSample) -> None: response_path = Path(args.response_root) / sample.split / f"{sample.id}.json" score_path = Path(args.score_root) / sample.split / f"{sample.id}.json" if score_path.exists() and not args.overwrite: From 50d61e3de5ec45e940f88e8018d9d3117cc888d6 Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Thu, 10 Sep 2026 07:15:32 +0000 Subject: [PATCH 3/7] [Frontend][Benchmark] new ruff Signed-off-by: ZacheryAU --- .../features/fullduplex/test_omni_duplex_eval_cli.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py index 6c68ed363cc..6e5d9f066af 100644 --- a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py +++ b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py @@ -11,16 +11,16 @@ from pathlib import Path import pytest - -from vllm_omni.benchmarks.duplex import omni_duplex_eval_runner as runner -from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample -from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge -from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult from vllm_omni.benchmarks.duplex_session_metrics import ( DUPLEX_METRICS_FILENAME, build_duplex_metrics_report, collect_duplex_session_metrics, ) + +from vllm_omni.benchmarks.duplex import omni_duplex_eval_runner as runner +from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample +from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge +from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult from vllm_omni.clients.duplex import EventCollector from vllm_omni.entrypoints.cli.benchmark import omni_duplex_eval as cli from vllm_omni.entrypoints.cli.benchmark.omni_duplex_eval import OmniDuplexEvalSubcommand From b4c0b425daf2eee9efd86a35c205308dbbc1eb67 Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Thu, 10 Sep 2026 08:19:19 +0000 Subject: [PATCH 4/7] [Frontend][Benchmark] duplex_session_metrics.py Signed-off-by: ZacheryAU --- .../benchmarks/duplex_session_metrics.py | 156 ++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 vllm_omni/benchmarks/duplex_session_metrics.py diff --git a/vllm_omni/benchmarks/duplex_session_metrics.py b/vllm_omni/benchmarks/duplex_session_metrics.py new file mode 100644 index 00000000000..3d6752b22a0 --- /dev/null +++ b/vllm_omni/benchmarks/duplex_session_metrics.py @@ -0,0 +1,156 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Assemble duplex serve metrics from a public EventCollector. + +This module does not instrument the server. It reads the same +``EventCollector.timing_summary`` / ``global_timing_summary`` surfaces that +OmniInteract already uses, derives RTF with ``compute_audio_rtf``, and folds +per-session ``global_*`` values into a run-level report. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from vllm_omni.metrics.definitions import compute_audio_rtf + +if TYPE_CHECKING: + from vllm_omni.clients.duplex import EventCollector + +DUPLEX_METRICS_FILENAME = "duplex_metrics.json" + +_REQUEST_MEASUREMENT_ORIGIN = { + "tpot": "Stage-0 engine mean time per output token", + "rtf": "response.created client receive to last audio packet divided by emitted audio duration", +} +_GLOBAL_MEASUREMENT_ORIGIN = { + "ttft": "input stream start to first non-empty text delta", + "ttfp": "input stream start to first audio packet", + "rtf": ( + "input stream start-to-last-audio receive time divided by total emitted audio duration; " + "includes concurrent realtime input" + ), +} + + +@dataclass(frozen=True) +class DuplexSessionMetricBundle: + """One session's request rows plus the session/global summary.""" + + request_metrics: list[dict[str, object]] + session_metrics: dict[str, object] + output_tokens: int + + +def audio_rtf_from_raw_metric(raw_metric: Mapping[str, object]) -> float | None: + """Derive audio RTF from client-reported generation and duration milliseconds.""" + generation_ms = raw_metric.get("audio_generation_ms") + duration_ms = raw_metric.get("audio_duration_ms") + if not isinstance(generation_ms, int | float) or not isinstance(duration_ms, int | float) or duration_ms <= 0: + return None + return round(compute_audio_rtf(float(generation_ms) / 1000.0, float(duration_ms) / 1000.0), 6) + + +def collect_duplex_session_metrics( + collector: EventCollector, + *, + stream_start: float, + session_id: str | None, +) -> DuplexSessionMetricBundle: + """Build per-response and session/global metrics for one duplex session. + + ``stream_start`` is the client monotonic time just before input media is + pushed. Per-response TTFT/TTFP prefer server ``response_request_metrics`` + when present; RTF and the session ``global_*`` window stay client-receive. + """ + from vllm_omni.clients.duplex import summarize_session_request_metrics + + request_metrics: list[dict[str, object]] = [] + output_tokens = 0 + for request_index, response_id in enumerate(collector.response_ids): + timing = collector.timing_summary( + after_s=stream_start, + input_committed_at_s=None, + response_id=response_id, + measurement_origin=_REQUEST_MEASUREMENT_ORIGIN, + ) + raw_metric = timing.get("request_metrics") + stage0 = timing.get("stage0_tokens") + metric: dict[str, object] = { + "session_id": session_id, + "request_index": request_index, + "response_id": response_id, + } + if isinstance(raw_metric, dict): + metric.update(raw_metric) + metric["rtf"] = audio_rtf_from_raw_metric(raw_metric) + if isinstance(stage0, dict): + metric["stage0_tokens"] = dict(stage0) + output_tokens += int(stage0.get("output_token_count") or 0) + if isinstance(raw_metric, dict) or isinstance(stage0, dict): + request_metrics.append(metric) + session_metrics = summarize_session_request_metrics( + request_metrics, + session_id=session_id, + ) + global_metrics = collector.global_timing_summary( + after_s=stream_start, + window_started_at_s=stream_start, + response_ids=list(collector.response_ids), + measurement_origin=_GLOBAL_MEASUREMENT_ORIGIN, + ) + if global_metrics: + session_metrics.update( + { + "global_ttft_ms": global_metrics.get("ttft_ms"), + "global_ttfp_ms": global_metrics.get("ttfp_ms"), + "global_rtf": audio_rtf_from_raw_metric(global_metrics), + "global_audio_generation_ms": global_metrics.get("audio_generation_ms"), + "global_audio_duration_ms": global_metrics.get("audio_duration_ms"), + "global_measurement_origin": global_metrics.get("measurement_origin"), + } + ) + return DuplexSessionMetricBundle( + request_metrics=request_metrics, + session_metrics=session_metrics, + output_tokens=output_tokens, + ) + + +def mean_duplex_global_metrics(session_metrics: Sequence[Mapping[str, object]]) -> dict[str, float]: + """Average finite, non-negative per-session ``global_*`` values.""" + result: dict[str, float] = {} + for session_key, result_key in ( + ("global_ttft_ms", "mean_duplex_global_ttft_ms"), + ("global_ttfp_ms", "mean_duplex_global_ttfp_ms"), + ("global_rtf", "mean_duplex_global_rtf"), + ): + values = [ + float(value) + for metric in session_metrics + if isinstance((value := metric.get(session_key)), int | float) + and not isinstance(value, bool) + and math.isfinite(value) + and value >= 0 + ] + if values: + result[result_key] = sum(values) / len(values) + return result + + +def build_duplex_metrics_report( + *, + request_metrics: Sequence[Mapping[str, object]], + session_metrics: Sequence[Mapping[str, object]], +) -> dict[str, object]: + """Flatten this generate run into the ``duplex_metrics.json`` payload.""" + report: dict[str, object] = { + "duplex_request_metrics": [dict(metric) for metric in request_metrics], + "duplex_session_metrics": [dict(metric) for metric in session_metrics], + } + report.update(mean_duplex_global_metrics(session_metrics)) + return report From 67e8db993b297c24f5823e745ae5d29443e8197c Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Thu, 10 Sep 2026 14:26:08 +0000 Subject: [PATCH 5/7] [Frontend][Benchmark] new ruff Signed-off-by: ZacheryAU --- .../features/fullduplex/test_omni_duplex_eval_cli.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py index 6e5d9f066af..6c68ed363cc 100644 --- a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py +++ b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py @@ -11,16 +11,16 @@ from pathlib import Path import pytest -from vllm_omni.benchmarks.duplex_session_metrics import ( - DUPLEX_METRICS_FILENAME, - build_duplex_metrics_report, - collect_duplex_session_metrics, -) from vllm_omni.benchmarks.duplex import omni_duplex_eval_runner as runner from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult +from vllm_omni.benchmarks.duplex_session_metrics import ( + DUPLEX_METRICS_FILENAME, + build_duplex_metrics_report, + collect_duplex_session_metrics, +) from vllm_omni.clients.duplex import EventCollector from vllm_omni.entrypoints.cli.benchmark import omni_duplex_eval as cli from vllm_omni.entrypoints.cli.benchmark.omni_duplex_eval import OmniDuplexEvalSubcommand From 890efbe6671e4b06cc83e9b00f6870bf9cc79654 Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Fri, 11 Sep 2026 14:33:41 +0000 Subject: [PATCH 6/7] [Frontend][Benchmark] fix review Signed-off-by: ZacheryAU --- .../fullduplex/test_omni_duplex_eval_cli.py | 185 ++++++++++++++++++ tests/engine/test_output_processor.py | 29 +++ .../openai_api/test_duplex_handler.py | 30 ++- .../benchmarks/duplex_session_metrics.py | 70 +++++++ .../cli/benchmark/omni_duplex_eval.py | 14 +- vllm_omni/entrypoints/duplex/protocol.py | 5 +- vllm_omni/outputs/output_processor.py | 31 ++- 7 files changed, 347 insertions(+), 17 deletions(-) diff --git a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py index 6c68ed363cc..16b6db736cb 100644 --- a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py +++ b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py @@ -20,6 +20,8 @@ DUPLEX_METRICS_FILENAME, build_duplex_metrics_report, collect_duplex_session_metrics, + merge_duplex_metrics_report, + read_duplex_metrics_report, ) from vllm_omni.clients.duplex import EventCollector from vllm_omni.entrypoints.cli.benchmark import omni_duplex_eval as cli @@ -201,6 +203,189 @@ async def handler(websocket): assert any(event["type"] == "playback.ack" for event in received) +def _pr_manifest(tmp_path: Path, sample_ids: tuple[str, ...]) -> Path: + manifest = tmp_path / "manifest.json" + manifest.write_text( + json.dumps( + [ + { + "id": sample_id, + "split": "PR_correction", + "question_text": "What changed?", + "answer1": "The object moved.", + } + for sample_id in sample_ids + ] + ), + encoding="utf-8", + ) + return manifest + + +def _run_generate_cli(manifest: Path, response_root: Path) -> None: + parser = argparse.ArgumentParser() + OmniDuplexEvalSubcommand.add_cli_args(parser) + OmniDuplexEvalSubcommand.cmd( + parser.parse_args( + [ + "generate", + "--dataset", + str(manifest), + "--family", + "pr", + "--model", + "mock", + "--ref-audio", + str(manifest), + "--response-root", + str(response_root), + ] + ) + ) + + +def _metrics_for(sample_id: str, *, ttft_ms: float, ttfp_ms: float, rtf: float) -> GenerateSampleResult: + split = "PR_correction" + return GenerateSampleResult( + output=Path(split) / f"{sample_id}.json", + request_metrics=[{"sample_id": sample_id, "split": split, "ttft_ms": ttft_ms, "ttfp_ms": ttfp_ms, "rtf": rtf}], + session_metrics={ + "sample_id": sample_id, + "split": split, + "global_ttft_ms": ttft_ms, + "global_ttfp_ms": ttfp_ms, + "global_rtf": rtf, + }, + ) + + +def _fake_generate(results: dict[str, GenerateSampleResult | None]): + async def generate( + sample: DuplexSample, + *, + output_root: str | Path, + url: str, + model: str, + ref_audio: str | Path, + fps: float = 1.0, + mix: str = "question", + pace: str = "realtime", + clock: str = "media", + overwrite: bool = False, + unit_ms: int = 1000, + ) -> GenerateSampleResult: + _ = (url, model, ref_audio, fps, mix, pace, clock, overwrite, unit_ms) + result = results[sample.id] + if result is None: + return GenerateSampleResult(output=Path(output_root) / sample.split / f"{sample.id}.json") + return result + + return generate + + +def test_merge_duplex_metrics_report_replaces_only_incoming_sample_keys(): + existing = build_duplex_metrics_report( + request_metrics=[ + {"sample_id": "kept", "split": "PR_correction", "ttft_ms": 11.0}, + {"sample_id": "replaced", "split": "PR_correction", "ttft_ms": 12.0}, + ], + session_metrics=[ + {"sample_id": "kept", "split": "PR_correction", "global_ttft_ms": 11.0}, + {"sample_id": "replaced", "split": "PR_correction", "global_ttft_ms": 12.0}, + ], + ) + merged = merge_duplex_metrics_report( + existing, + request_metrics=[{"sample_id": "replaced", "split": "PR_correction", "ttft_ms": 99.0}], + session_metrics=[{"sample_id": "replaced", "split": "PR_correction", "global_ttft_ms": 99.0}], + ) + assert [row["sample_id"] for row in merged["duplex_request_metrics"]] == ["kept", "replaced"] + assert merged["duplex_request_metrics"][1]["ttft_ms"] == 99.0 + assert merged["mean_duplex_global_ttft_ms"] == 55.0 + + +def test_read_duplex_metrics_report_returns_none_when_missing(tmp_path: Path): + assert read_duplex_metrics_report(tmp_path / DUPLEX_METRICS_FILENAME) is None + + +def test_read_duplex_metrics_report_rejects_invalid_json(tmp_path: Path): + path = tmp_path / DUPLEX_METRICS_FILENAME + path.write_text("{not-json", encoding="utf-8") + with pytest.raises(json.JSONDecodeError): + read_duplex_metrics_report(path) + + +def test_merge_duplex_metrics_report_without_existing_is_passthrough(): + merged = merge_duplex_metrics_report( + None, + request_metrics=[{"sample_id": "a", "split": "PR_correction", "ttft_ms": 1.0}], + session_metrics=[{"sample_id": "a", "split": "PR_correction", "global_ttft_ms": 1.0}], + ) + assert merged["duplex_request_metrics"][0]["ttft_ms"] == 1.0 + assert merged["mean_duplex_global_ttft_ms"] == 1.0 + + +def test_merge_duplex_metrics_report_rejects_non_object_rows(): + with pytest.raises(ValueError, match="duplex_request_metrics entries must be JSON objects"): + merge_duplex_metrics_report( + {"duplex_request_metrics": ["bad"]}, + request_metrics=[], + session_metrics=[], + ) + + +def test_cli_generate_all_skipped_keeps_existing_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + manifest = _pr_manifest(tmp_path, ("sample-1",)) + response_root = tmp_path / "responses" + monkeypatch.setattr( + cli, + "generate_sample", + _fake_generate({"sample-1": _metrics_for("sample-1", ttft_ms=10.0, ttfp_ms=20.0, rtf=0.5)}), + ) + _run_generate_cli(manifest, response_root) + monkeypatch.setattr(cli, "generate_sample", _fake_generate({"sample-1": None})) + _run_generate_cli(manifest, response_root) + duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8")) + assert duplex_metrics["duplex_request_metrics"][0]["ttft_ms"] == 10.0 + assert duplex_metrics["mean_duplex_global_ttft_ms"] == 10.0 + assert duplex_metrics["mean_duplex_global_ttfp_ms"] == 20.0 + assert duplex_metrics["mean_duplex_global_rtf"] == 0.5 + + +def test_cli_generate_partial_resume_merges_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + manifest = _pr_manifest(tmp_path, ("sample-1", "sample-2")) + response_root = tmp_path / "responses" + monkeypatch.setattr( + cli, + "generate_sample", + _fake_generate( + { + "sample-1": _metrics_for("sample-1", ttft_ms=10.0, ttfp_ms=20.0, rtf=0.5), + "sample-2": _metrics_for("sample-2", ttft_ms=30.0, ttfp_ms=40.0, rtf=0.25), + } + ), + ) + _run_generate_cli(manifest, response_root) + monkeypatch.setattr( + cli, + "generate_sample", + _fake_generate( + { + "sample-1": None, + "sample-2": _metrics_for("sample-2", ttft_ms=90.0, ttfp_ms=80.0, rtf=1.5), + } + ), + ) + _run_generate_cli(manifest, response_root) + duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8")) + by_id = {row["sample_id"]: row for row in duplex_metrics["duplex_request_metrics"]} + assert by_id["sample-1"]["ttft_ms"] == 10.0 + assert by_id["sample-2"]["ttft_ms"] == 90.0 + assert duplex_metrics["mean_duplex_global_ttft_ms"] == 50.0 + assert duplex_metrics["mean_duplex_global_ttfp_ms"] == 50.0 + assert duplex_metrics["mean_duplex_global_rtf"] == 1.0 + + def test_collect_duplex_session_metrics_matches_omniinteract_window(): collector = EventCollector() audio = { diff --git a/tests/engine/test_output_processor.py b/tests/engine/test_output_processor.py index 351beb297a4..8618061d8ca 100644 --- a/tests/engine/test_output_processor.py +++ b/tests/engine/test_output_processor.py @@ -200,6 +200,35 @@ def update_segment(_output, timestamp, _was_prefilling, native_stats, *_args): assert second_segment["vllm_tpot_ms"] == pytest.approx(10.0) +def test_native_text_tpot_weights_multi_token_engine_output(monkeypatch): + monkeypatch.setattr(VLLMOutputProcessor, "_update_stats_from_output", lambda *args, **kwargs: None) + processor = object.__new__(MultimodalOutputProcessor) + processor._native_text_metrics_by_request = {} + processor.lora_states = {} + state = _make_state(RequestOutputKind.DELTA) + iteration_stats = MagicMock() + + def update_segment(_output, timestamp, _was_prefilling, native_stats, *_args): + native_stats.num_generation_tokens = 1 if timestamp == 10.0 else 4 + native_stats.first_token_ts = 10.0 + native_stats.last_token_ts = timestamp + native_stats.first_token_latency = 0.01 + + iteration_stats.update_from_output.side_effect = update_segment + state.is_prefilling = True + processor._update_stats_from_output(state, MagicMock(), 10.0, iteration_stats) + state.is_prefilling = False + processor._update_stats_from_output(state, MagicMock(), 10.03, iteration_stats) + record = processor.pop_native_text_metrics("r") + + assert record["num_generation_tokens"] == 4 + assert record["vllm_itls_ms"] == pytest.approx([30.0]) + assert record["vllm_itl_ms"] == pytest.approx(30.0) + assert record["vllm_tpot_ms"] == pytest.approx(10.0) + assert "_tpot_elapsed_ms" not in record + assert "_tpot_intervals" not in record + + @pytest.mark.parametrize( ("finished_tpot_s", "expected_tpot_ms"), [(0.0, 19.0), (0.023, 23.0)], diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index c0e2bb5376e..d3cfae586ba 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -8797,6 +8797,26 @@ def test_response_tpot_fallback_ignores_single_token_segment(): assert second_metrics["0"]["vllm_tpot_ms"] == 15.0 +def test_response_tpot_keeps_token_weighted_value_when_itls_exist(): + session = DuplexSession( + session_id="sid-response-tpot-multi-token", + config=DuplexSessionConfig(extra_body={"auto_response": True}), + ) + session.begin_response(turn_id=0) + metrics = session.accumulate_response_stage_metrics( + { + "0": { + "num_tokens_out": 4, + "vllm_tpot_ms": 10.0, + "vllm_itls_ms": [30.0], + } + } + ) + assert metrics["0"]["vllm_itls_ms"] == [30.0] + assert metrics["0"]["vllm_itl_ms"] == 30.0 + assert metrics["0"]["vllm_tpot_ms"] == 10.0 + + @pytest.mark.asyncio async def test_continuous_response_metrics_accumulate_only_owned_model_units(monkeypatch): handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient())) @@ -8814,7 +8834,7 @@ async def test_continuous_response_metrics_accumulate_only_owned_model_units(mon async def send_json(payload: dict[str, Any]) -> None: sent.append(payload) - async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list[float]) -> None: + async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list[float], tpot_ms: float) -> None: await handler._send_one_native_duplex_event( send_json, { @@ -8833,7 +8853,7 @@ async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list "num_tokens_out": tokens, "stage_gen_time_ms": 80.0, "vllm_ttft_ms": ttft_ms, - "vllm_tpot_ms": 9.0, + "vllm_tpot_ms": tpot_ms, "vllm_itls_ms": itls, } }, @@ -8841,8 +8861,8 @@ async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list session=session, ) - await emit(text="first", audio="audio-a", tokens=3, ttft_ms=50.0, itls=[10.0, 11.0]) - await emit(text="second", audio="audio-b", tokens=4, ttft_ms=70.0, itls=[12.0, 13.0, 14.0]) + await emit(text="first", audio="audio-a", tokens=3, ttft_ms=50.0, itls=[10.0, 11.0], tpot_ms=10.5) + await emit(text="second", audio="audio-b", tokens=4, ttft_ms=70.0, itls=[12.0, 13.0, 14.0], tpot_ms=13.0) deltas = [payload for payload in sent if payload.get("type") == "response.output_audio.delta"] assert len(deltas) == 2 @@ -8866,7 +8886,7 @@ async def emit(*, text: str, audio: str, tokens: int, ttft_ms: float, itls: list first_response_id = deltas[0]["response_id"] session.end_response() - await emit(text="third", audio="audio-c", tokens=3, ttft_ms=90.0, itls=[15.0, 16.0]) + await emit(text="third", audio="audio-c", tokens=3, ttft_ms=90.0, itls=[15.0, 16.0], tpot_ms=15.5) third_delta = [payload for payload in sent if payload.get("type") == "response.output_audio.delta"][-1] third_metrics = third_delta["vllm_omni"]["stage_metrics"]["0"] diff --git a/vllm_omni/benchmarks/duplex_session_metrics.py b/vllm_omni/benchmarks/duplex_session_metrics.py index 3d6752b22a0..388adadca34 100644 --- a/vllm_omni/benchmarks/duplex_session_metrics.py +++ b/vllm_omni/benchmarks/duplex_session_metrics.py @@ -11,9 +11,11 @@ from __future__ import annotations +import json import math from collections.abc import Mapping, Sequence from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING from vllm_omni.metrics.definitions import compute_audio_rtf @@ -154,3 +156,71 @@ def build_duplex_metrics_report( } report.update(mean_duplex_global_metrics(session_metrics)) return report + + +def sample_metric_key(metric: Mapping[str, object]) -> tuple[str, str] | None: + """Identity used to merge resume rows: ``(split, sample_id)``.""" + split = metric.get("split") + sample_id = metric.get("sample_id") + if isinstance(split, str) and split and isinstance(sample_id, str) and sample_id: + return (split, sample_id) + return None + + +def merge_metric_rows( + existing: Sequence[Mapping[str, object]], + incoming: Sequence[Mapping[str, object]], +) -> list[dict[str, object]]: + """Replace existing rows that share an incoming ``(split, sample_id)``. + + Incoming rows without a key are appended. Existing rows without a key are + kept because they cannot be matched. + """ + incoming_keys = {key for metric in incoming if (key := sample_metric_key(metric)) is not None} + merged = [dict(metric) for metric in existing if (key := sample_metric_key(metric)) not in incoming_keys] + merged.extend(dict(metric) for metric in incoming) + return merged + + +def read_duplex_metrics_report(path: Path) -> dict[str, object] | None: + """Load an on-disk report. Missing files return ``None``; corrupt JSON raises.""" + if not path.is_file(): + return None + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"{path} must contain a JSON object") + return payload + + +def _metric_rows(value: object, *, field_name: str) -> list[Mapping[str, object]]: + if value is None: + return [] + if not isinstance(value, list): + raise ValueError(f"{field_name} must be a JSON list") + rows: list[Mapping[str, object]] = [] + for item in value: + if not isinstance(item, dict): + raise ValueError(f"{field_name} entries must be JSON objects") + rows.append(item) + return rows + + +def merge_duplex_metrics_report( + existing: Mapping[str, object] | None, + *, + request_metrics: Sequence[Mapping[str, object]], + session_metrics: Sequence[Mapping[str, object]], +) -> dict[str, object]: + """Merge this run into a previous ``duplex_metrics.json`` payload.""" + previous_requests = _metric_rows( + None if existing is None else existing.get("duplex_request_metrics"), + field_name="duplex_request_metrics", + ) + previous_sessions = _metric_rows( + None if existing is None else existing.get("duplex_session_metrics"), + field_name="duplex_session_metrics", + ) + return build_duplex_metrics_report( + request_metrics=merge_metric_rows(previous_requests, request_metrics), + session_metrics=merge_metric_rows(previous_sessions, session_metrics), + ) diff --git a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py index 761b9c04f42..172108f2f77 100644 --- a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py +++ b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py @@ -4,8 +4,9 @@ """CLI for Omni-DuplexEval generate / evaluate / summarize. ``generate`` still writes per-sample timed sentences and ``*.meta.json``. -Duplex performance metrics for the run are written once to -``/duplex_metrics.json``. +Duplex performance metrics are written to ``/duplex_metrics.json``. +Skipped samples keep previously recorded rows; incoming ``(split, sample_id)`` +rows replace the matching ones, then ``mean_duplex_global_*`` is recomputed. """ import argparse @@ -18,7 +19,11 @@ from vllm_omni.benchmarks.duplex.omni_duplex_eval_eval import evaluate_sample, summarize_scores from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import GenerateSampleResult, generate_sample -from vllm_omni.benchmarks.duplex_session_metrics import DUPLEX_METRICS_FILENAME, build_duplex_metrics_report +from vllm_omni.benchmarks.duplex_session_metrics import ( + DUPLEX_METRICS_FILENAME, + merge_duplex_metrics_report, + read_duplex_metrics_report, +) from vllm_omni.entrypoints.cli.benchmark.base import OmniBenchmarkSubcommandBase @@ -69,7 +74,8 @@ def _write_duplex_metrics(response_root: str | Path, results: list[GenerateSampl path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( - build_duplex_metrics_report( + merge_duplex_metrics_report( + read_duplex_metrics_report(path), request_metrics=request_metrics, session_metrics=session_metrics, ), diff --git a/vllm_omni/entrypoints/duplex/protocol.py b/vllm_omni/entrypoints/duplex/protocol.py index 1f5a2bab494..df574a54f96 100644 --- a/vllm_omni/entrypoints/duplex/protocol.py +++ b/vllm_omni/entrypoints/duplex/protocol.py @@ -1109,10 +1109,7 @@ def accumulate_response_stage_metrics( tpot_ms = raw_values.get("vllm_tpot_ms") token_count = raw_values.get("num_tokens_out") - response_itls = current.get("vllm_itls_ms") - if isinstance(response_itls, list) and response_itls: - current["vllm_tpot_ms"] = current["vllm_itl_ms"] - elif isinstance(tpot_ms, int | float) and not isinstance(tpot_ms, bool) and tpot_ms > 0: + if isinstance(tpot_ms, int | float) and not isinstance(tpot_ms, bool) and tpot_ms > 0: weight = ( max(int(token_count) - 1, 0) if isinstance(token_count, int | float) and not isinstance(token_count, bool) diff --git a/vllm_omni/outputs/output_processor.py b/vllm_omni/outputs/output_processor.py index cd65a900f48..bd60e479397 100644 --- a/vllm_omni/outputs/output_processor.py +++ b/vllm_omni/outputs/output_processor.py @@ -59,6 +59,10 @@ def _modality_to_type_string(modality: OutputModality) -> str: return "text" +_TPOT_ELAPSED_MS = "_tpot_elapsed_ms" +_TPOT_INTERVALS = "_tpot_intervals" + + def _mean_time_per_output_token_ms(stats: RequestStateStats) -> float: output_intervals = stats.num_generation_tokens - 1 if output_intervals <= 0: @@ -67,6 +71,17 @@ def _mean_time_per_output_token_ms(stats: RequestStateStats) -> float: return decode_time_s * 1000.0 / float(output_intervals) +def _accumulate_segment_tpot(record: dict[str, object], *, elapsed_ms: float, new_tokens: int) -> None: + """Weight one decode step by its token count. ITL stays one sample per step.""" + if new_tokens <= 0: + return + total_elapsed_ms = float(record.get(_TPOT_ELAPSED_MS) or 0.0) + elapsed_ms + total_intervals = int(record.get(_TPOT_INTERVALS) or 0) + new_tokens + record[_TPOT_ELAPSED_MS] = total_elapsed_ms + record[_TPOT_INTERVALS] = total_intervals + record["vllm_tpot_ms"] = total_elapsed_ms / float(total_intervals) + + class OmniRequestState(RequestState): """Request state for omni models, tracking multimodal outputs. @@ -430,7 +445,10 @@ def _native_text_metric_record(self, request_id: str) -> dict[str, Any]: ) def pop_native_text_metrics(self, request_id: str) -> dict[str, Any]: - return self._native_text_metrics_by_request.pop(request_id, {}) + record = self._native_text_metrics_by_request.pop(request_id, {}) + record.pop(_TPOT_ELAPSED_MS, None) + record.pop(_TPOT_INTERVALS, None) + return record def abort_requests(self, request_ids, internal: bool) -> list[str]: aborted_ids, _outputs = self.abort_requests_collecting_outputs(request_ids, internal=internal) @@ -770,6 +788,7 @@ def _update_stats_from_output( was_prefilling = req_state.is_prefilling native_stats = req_state.native_text_stats if isinstance(req_state, OmniRequestState) else None previous_last_token_ts = native_stats.last_token_ts if native_stats is not None else 0.0 + previous_num_generation_tokens = int(native_stats.num_generation_tokens) if native_stats is not None else 0 # NOTE: We pass ``None`` for *iteration_stats* to the parent so that # the upstream's ``_update_stats_from_output`` logs stats via its own @@ -808,9 +827,13 @@ def _update_stats_from_output( itls_ms = record.setdefault("vllm_itls_ms", []) itls_ms.append(itl_ms) record["vllm_itl_ms"] = sum(itls_ms) / float(len(itls_ms)) - record["vllm_tpot_ms"] = ( - record["vllm_itl_ms"] if record["vllm_itls_ms"] else _mean_time_per_output_token_ms(native_stats) - ) + _accumulate_segment_tpot( + record, + elapsed_ms=itl_ms, + new_tokens=max(int(native_stats.num_generation_tokens) - previous_num_generation_tokens, 0), + ) + elif not record["vllm_itls_ms"]: + record["vllm_tpot_ms"] = _mean_time_per_output_token_ms(native_stats) def _update_stats_from_finished( self, From 8beca59e3c1d38256bcbeb40a8d9185321b94df6 Mon Sep 17 00:00:00 2001 From: ZacheryAU Date: Sun, 13 Sep 2026 15:40:09 +0000 Subject: [PATCH 7/7] [Frontend][Benchmark] fix review, add median / P99, change global_* to stream_* Signed-off-by: ZacheryAU --- tests/benchmarks/metrics/test_metrics.py | 34 +++++------ tests/benchmarks/patch/test_patch.py | 18 +++--- tests/benchmarks/test_omniinteract.py | 26 ++++----- tests/clients/test_duplex_client.py | 15 +++-- .../fullduplex/test_omni_duplex_eval_cli.py | 52 ++++++++--------- .../test_minicpmo_4_5_duplex.py | 11 ++-- .../test_minicpmo_realtime_duplex_drivers.py | 6 +- .../benchmarks/duplex_session_metrics.py | 51 +++++++++-------- vllm_omni/benchmarks/metrics/metrics.py | 8 +-- vllm_omni/benchmarks/patch/patch.py | 33 ++++++----- vllm_omni/clients/duplex.py | 57 +++++++++++++++---- .../cli/benchmark/omni_duplex_eval.py | 2 +- 12 files changed, 182 insertions(+), 131 deletions(-) diff --git a/tests/benchmarks/metrics/test_metrics.py b/tests/benchmarks/metrics/test_metrics.py index d7dd6c14a6b..0aae841afa4 100644 --- a/tests/benchmarks/metrics/test_metrics.py +++ b/tests/benchmarks/metrics/test_metrics.py @@ -150,16 +150,16 @@ def test_unmeasured_duplex_latency_does_not_add_zero_samples(): measured.audio_ttfp = 0.2 measured.audio_rtf = 0.5 measured.duplex_session_metrics = { - "mean_ttft_ms": 100.0, - "mean_ttfp_ms": 200.0, - "mean_rtf": 0.5, + "ttft_ms": 100.0, + "ttfp_ms": 200.0, + "rtf": 0.5, } listen_only = _make_output(100, output_tokens=0) listen_only.ttft = listen_only.audio_ttfp = listen_only.audio_rtf = 0.0 listen_only.duplex_session_metrics = { - "mean_ttft_ms": None, - "mean_ttfp_ms": None, - "mean_rtf": None, + "ttft_ms": None, + "ttfp_ms": None, + "rtf": None, } metrics = _calculate_test_metrics([measured, listen_only], {"ttft": 150.0}) @@ -174,9 +174,9 @@ def test_unmeasured_duplex_latency_does_not_add_zero_samples(): def test_all_unmeasured_duplex_latency_is_not_reported_as_zero(): listen_only = _make_output(100, output_tokens=0) listen_only.duplex_session_metrics = { - "mean_ttft_ms": None, - "mean_ttfp_ms": None, - "mean_rtf": None, + "ttft_ms": None, + "ttfp_ms": None, + "rtf": None, } metrics = _calculate_test_metrics( @@ -196,13 +196,13 @@ def test_unmeasured_duplex_tpot_does_not_add_zero_or_misalign_goodput(): missing_tpot.itl = [] missing_tpot.text_latency = missing_tpot.ttft = 0.1 missing_tpot.tpot_measured = False - missing_tpot.duplex_session_metrics = {"mean_ttft_ms": 100.0} + missing_tpot.duplex_session_metrics = {"ttft_ms": 100.0} slow_ttft = _make_output(100, output_tokens=5) slow_ttft.ttft = 1.0 slow_ttft.text_latency = 1.4 slow_ttft.itl = [0.1] * 4 - slow_ttft.duplex_session_metrics = {"mean_ttft_ms": 1000.0} + slow_ttft.duplex_session_metrics = {"ttft_ms": 1000.0} metrics = _calculate_test_metrics( [missing_tpot, slow_ttft], @@ -219,7 +219,7 @@ def test_all_unmeasured_duplex_token_timing_is_not_reported_as_zero(): output.itl = [] output.text_latency = output.ttft output.tpot_measured = False - output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + output.duplex_session_metrics = {"ttft_ms": 100.0} metrics = _calculate_test_metrics([output], {"tpot": float("inf")}) @@ -236,7 +236,7 @@ def test_duplex_response_timings_do_not_build_a_session_token_timeline(): {"response_id": "r1", "stage0_tokens": {"itls_ms": [100.0, 100.0]}}, {"response_id": "r2", "stage0_tokens": {"itls_ms": [100.0, 100.0]}}, ] - output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + output.duplex_session_metrics = {"ttft_ms": 100.0} metrics = _calculate_test_metrics([output]) @@ -252,7 +252,7 @@ def test_unmeasured_tpot_stays_missing_after_tokenizer_fallback(): output.itl = [] output.text_latency = output.ttft output.tpot_measured = False - output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + output.duplex_session_metrics = {"ttft_ms": 100.0} def tokenizer(text, *, add_special_tokens): assert text == output.generated_text @@ -282,7 +282,7 @@ def tokenizer(text, *, add_special_tokens): def test_zero_itl_does_not_create_zero_tpot(): output = _make_output(100, output_tokens=3) output.itl = [0.0, 0.0] - output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + output.duplex_session_metrics = {"ttft_ms": 100.0} metrics = _calculate_test_metrics([output], {"tpot": 1.0}) @@ -357,9 +357,9 @@ def test_single_token_responses_do_not_report_zero_tpot(capsys): def test_duplex_goodput_does_not_pair_measurements_from_different_requests(): text_only, audio_only = _make_output(100), _make_output(100) text_only.ttft, text_only.audio_ttfp = 0.1, 0.0 - text_only.duplex_session_metrics = {"mean_ttft_ms": 100.0, "mean_ttfp_ms": None, "mean_rtf": None} + text_only.duplex_session_metrics = {"ttft_ms": 100.0, "ttfp_ms": None, "rtf": None} audio_only.ttft, audio_only.audio_ttfp = 0.0, 0.2 - audio_only.duplex_session_metrics = {"mean_ttft_ms": None, "mean_ttfp_ms": 200.0, "mean_rtf": None} + audio_only.duplex_session_metrics = {"ttft_ms": None, "ttfp_ms": 200.0, "rtf": None} metrics = _calculate_test_metrics([text_only, audio_only], {"ttft": 500.0, "audio_ttft": 500.0}) diff --git a/tests/benchmarks/patch/test_patch.py b/tests/benchmarks/patch/test_patch.py index 664842d275c..2bece6fdc90 100644 --- a/tests/benchmarks/patch/test_patch.py +++ b/tests/benchmarks/patch/test_patch.py @@ -296,14 +296,16 @@ async def close_session(self, **_kwargs): "audio_generation_ms": pytest.approx(30.0, abs=2.0), "audio_duration_ms": 100.0, } - assert output.duplex_session_metrics == { - "session_id": session_id, - "audio_turn_count": 4, - "mean_ttft_ms": pytest.approx(20.0, abs=2.0), - "mean_tpot_ms": 10.0, - "mean_ttfp_ms": pytest.approx(30.0, abs=2.0), - "mean_rtf": pytest.approx(0.3, abs=0.03), - } + session = output.duplex_session_metrics + assert session["session_id"] == session_id + assert session["audio_turn_count"] == 4 + assert session["ttft_ms"]["count"] == 4 + assert session["ttft_ms"]["mean"] == pytest.approx(20.0, abs=2.0) + assert session["tpot_ms"] == {"count": 4, "mean": 10.0, "p50": 10.0, "p99": 10.0} + assert session["ttfp_ms"]["count"] == 4 + assert session["ttfp_ms"]["mean"] == pytest.approx(30.0, abs=2.0) + assert session["rtf"]["count"] == 4 + assert session["rtf"]["mean"] == pytest.approx(0.3, abs=0.03) def test_stage0_token_timings_use_weighted_tpot_when_itls_are_incomplete(): diff --git a/tests/benchmarks/test_omniinteract.py b/tests/benchmarks/test_omniinteract.py index b93c0ed0b6e..8b767c8ab39 100644 --- a/tests/benchmarks/test_omniinteract.py +++ b/tests/benchmarks/test_omniinteract.py @@ -148,7 +148,7 @@ def _successful_output( return case, collector, result -def test_response_metrics_include_engine_tpot_and_global_stream_window(): +def test_response_metrics_include_engine_tpot_and_stream_window(): first_audio = _audio("r1") first_audio["metadata"] = { "audio_duration_ms": 100, @@ -194,16 +194,16 @@ def test_response_metrics_include_engine_tpot_and_global_stream_window(): assert result.duplex_session_metrics == { "session_id": "session", "audio_turn_count": 2, - "mean_ttft_ms": 100.0, - "mean_tpot_ms": 15.0, - "mean_ttfp_ms": 300.0, - "mean_rtf": 2.0, - "global_ttft_ms": 200.0, - "global_ttfp_ms": 300.0, - "global_rtf": 8.333333, - "global_audio_generation_ms": 2500.0, - "global_audio_duration_ms": 300.0, - "global_measurement_origin": { + "ttft_ms": {"count": 2, "mean": 100.0, "p50": 100.0, "p99": 100.0}, + "tpot_ms": {"count": 2, "mean": 15.0, "p50": 10.0, "p99": 20.0}, + "ttfp_ms": {"count": 2, "mean": 300.0, "p50": 200.0, "p99": 400.0}, + "rtf": {"count": 2, "mean": 2.0, "p50": 2.0, "p99": 2.0}, + "stream_ttft_ms": 200.0, + "stream_ttfp_ms": 300.0, + "stream_rtf": 8.333333, + "stream_audio_generation_ms": 2500.0, + "stream_audio_duration_ms": 300.0, + "stream_measurement_origin": { "ttft": "input stream start to first non-empty text delta", "ttfp": "input stream start to first audio packet", "rtf": ( @@ -711,7 +711,7 @@ async def run(_case_arg, config, **kwargs): transcript="timing metadata missing", output_tokens=0, duplex_request_metrics=[{"request_metrics": {"ttft_ms": 1.0}}], - duplex_session_metrics={"mean_ttft_ms": 10.0}, + duplex_session_metrics={"ttft_ms": 10.0}, ) monkeypatch.setattr(benchmark_patch, "run_omniinteract_case", run) @@ -779,7 +779,7 @@ async def run(*args, **kwargs): success=True, output_tokens=5, duplex_request_metrics=request_metrics, - duplex_session_metrics={"mean_ttft_ms": 100.0}, + duplex_session_metrics={"ttft_ms": 100.0}, ) monkeypatch.setattr(benchmark_patch, "run_omniinteract_case", run) diff --git a/tests/clients/test_duplex_client.py b/tests/clients/test_duplex_client.py index 26beccceb6c..e4712fe6121 100644 --- a/tests/clients/test_duplex_client.py +++ b/tests/clients/test_duplex_client.py @@ -755,8 +755,13 @@ def test_session_summary_omits_unmeasured_zero_tpot(): session_id="session", ) - assert summary["mean_ttft_ms"] == 150.0 - assert "mean_tpot_ms" not in summary + assert summary["ttft_ms"] == { + "count": 2, + "mean": 150.0, + "p50": 100.0, + "p99": 200.0, + } + assert "tpot_ms" not in summary def test_event_collector_reports_global_metrics_across_responses(): @@ -952,9 +957,9 @@ def test_summarize_session_request_metrics_averages_audio_turns(): assert summary == { "session_id": "sess-1", "audio_turn_count": 2, - "mean_ttft_ms": 200.0, - "mean_ttfp_ms": 300.0, - "mean_rtf": 0.6, + "ttft_ms": {"count": 2, "mean": 200.0, "p50": 100.0, "p99": 300.0}, + "ttfp_ms": {"count": 2, "mean": 300.0, "p50": 200.0, "p99": 400.0}, + "rtf": {"count": 2, "mean": 0.6, "p50": 0.5, "p99": 0.7}, } diff --git a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py index 16b6db736cb..4212410eb28 100644 --- a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py +++ b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py @@ -67,9 +67,9 @@ async def fake_generate(sample, *, output_root, **kwargs): session_metrics={ "sample_id": sample.id, "split": sample.split, - "global_ttft_ms": 10.0, - "global_ttfp_ms": 20.0, - "global_rtf": 0.5, + "stream_ttft_ms": 10.0, + "stream_ttfp_ms": 20.0, + "stream_rtf": 0.5, }, ) @@ -112,9 +112,9 @@ def chat(self, *args, **kwargs): "rtf": 0.5, } ] - assert duplex_metrics["mean_duplex_global_ttft_ms"] == 10.0 - assert duplex_metrics["mean_duplex_global_ttfp_ms"] == 20.0 - assert duplex_metrics["mean_duplex_global_rtf"] == 0.5 + assert duplex_metrics["duplex_stream_ttft_ms"]["mean"] == 10.0 + assert duplex_metrics["duplex_stream_ttfp_ms"]["mean"] == 20.0 + assert duplex_metrics["duplex_stream_rtf"]["mean"] == 0.5 evaluate = parser.parse_args( [ "evaluate", @@ -199,7 +199,7 @@ async def handler(websocket): assert "duplex_request_metrics" not in meta assert result.request_metrics assert result.request_metrics[0]["sample_id"] == "sample" - assert result.session_metrics["global_ttfp_ms"] is not None + assert result.session_metrics["stream_ttfp_ms"] is not None assert any(event["type"] == "playback.ack" for event in received) @@ -252,9 +252,9 @@ def _metrics_for(sample_id: str, *, ttft_ms: float, ttfp_ms: float, rtf: float) session_metrics={ "sample_id": sample_id, "split": split, - "global_ttft_ms": ttft_ms, - "global_ttfp_ms": ttfp_ms, - "global_rtf": rtf, + "stream_ttft_ms": ttft_ms, + "stream_ttfp_ms": ttfp_ms, + "stream_rtf": rtf, }, ) @@ -290,18 +290,18 @@ def test_merge_duplex_metrics_report_replaces_only_incoming_sample_keys(): {"sample_id": "replaced", "split": "PR_correction", "ttft_ms": 12.0}, ], session_metrics=[ - {"sample_id": "kept", "split": "PR_correction", "global_ttft_ms": 11.0}, - {"sample_id": "replaced", "split": "PR_correction", "global_ttft_ms": 12.0}, + {"sample_id": "kept", "split": "PR_correction", "stream_ttft_ms": 11.0}, + {"sample_id": "replaced", "split": "PR_correction", "stream_ttft_ms": 12.0}, ], ) merged = merge_duplex_metrics_report( existing, request_metrics=[{"sample_id": "replaced", "split": "PR_correction", "ttft_ms": 99.0}], - session_metrics=[{"sample_id": "replaced", "split": "PR_correction", "global_ttft_ms": 99.0}], + session_metrics=[{"sample_id": "replaced", "split": "PR_correction", "stream_ttft_ms": 99.0}], ) assert [row["sample_id"] for row in merged["duplex_request_metrics"]] == ["kept", "replaced"] assert merged["duplex_request_metrics"][1]["ttft_ms"] == 99.0 - assert merged["mean_duplex_global_ttft_ms"] == 55.0 + assert merged["duplex_stream_ttft_ms"]["mean"] == 55.0 def test_read_duplex_metrics_report_returns_none_when_missing(tmp_path: Path): @@ -319,10 +319,10 @@ def test_merge_duplex_metrics_report_without_existing_is_passthrough(): merged = merge_duplex_metrics_report( None, request_metrics=[{"sample_id": "a", "split": "PR_correction", "ttft_ms": 1.0}], - session_metrics=[{"sample_id": "a", "split": "PR_correction", "global_ttft_ms": 1.0}], + session_metrics=[{"sample_id": "a", "split": "PR_correction", "stream_ttft_ms": 1.0}], ) assert merged["duplex_request_metrics"][0]["ttft_ms"] == 1.0 - assert merged["mean_duplex_global_ttft_ms"] == 1.0 + assert merged["duplex_stream_ttft_ms"]["mean"] == 1.0 def test_merge_duplex_metrics_report_rejects_non_object_rows(): @@ -347,9 +347,9 @@ def test_cli_generate_all_skipped_keeps_existing_metrics(tmp_path: Path, monkeyp _run_generate_cli(manifest, response_root) duplex_metrics = json.loads((response_root / DUPLEX_METRICS_FILENAME).read_text(encoding="utf-8")) assert duplex_metrics["duplex_request_metrics"][0]["ttft_ms"] == 10.0 - assert duplex_metrics["mean_duplex_global_ttft_ms"] == 10.0 - assert duplex_metrics["mean_duplex_global_ttfp_ms"] == 20.0 - assert duplex_metrics["mean_duplex_global_rtf"] == 0.5 + assert duplex_metrics["duplex_stream_ttft_ms"]["mean"] == 10.0 + assert duplex_metrics["duplex_stream_ttfp_ms"]["mean"] == 20.0 + assert duplex_metrics["duplex_stream_rtf"]["mean"] == 0.5 def test_cli_generate_partial_resume_merges_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): @@ -381,9 +381,9 @@ def test_cli_generate_partial_resume_merges_metrics(tmp_path: Path, monkeypatch: by_id = {row["sample_id"]: row for row in duplex_metrics["duplex_request_metrics"]} assert by_id["sample-1"]["ttft_ms"] == 10.0 assert by_id["sample-2"]["ttft_ms"] == 90.0 - assert duplex_metrics["mean_duplex_global_ttft_ms"] == 50.0 - assert duplex_metrics["mean_duplex_global_ttfp_ms"] == 50.0 - assert duplex_metrics["mean_duplex_global_rtf"] == 1.0 + assert duplex_metrics["duplex_stream_ttft_ms"]["mean"] == 50.0 + assert duplex_metrics["duplex_stream_ttfp_ms"]["mean"] == 50.0 + assert duplex_metrics["duplex_stream_rtf"]["mean"] == 1.0 def test_collect_duplex_session_metrics_matches_omniinteract_window(): @@ -407,14 +407,14 @@ def test_collect_duplex_session_metrics_matches_omniinteract_window(): assert bundle.request_metrics[0]["ttft_ms"] == 100.0 assert bundle.request_metrics[0]["ttfp_ms"] == 200.0 assert bundle.request_metrics[0]["rtf"] == 2.0 - assert bundle.session_metrics["global_ttft_ms"] == 200.0 - assert bundle.session_metrics["global_ttfp_ms"] == 300.0 + assert bundle.session_metrics["stream_ttft_ms"] == 200.0 + assert bundle.session_metrics["stream_ttfp_ms"] == 300.0 report = build_duplex_metrics_report( request_metrics=bundle.request_metrics, session_metrics=[bundle.session_metrics], ) - assert report["mean_duplex_global_ttft_ms"] == 200.0 - assert report["mean_duplex_global_ttfp_ms"] == 300.0 + assert report["duplex_stream_ttft_ms"]["mean"] == 200.0 + assert report["duplex_stream_ttfp_ms"]["mean"] == 300.0 def test_judge_exercises_openai_http_schema(): diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py index a8ed72511bd..c04d3acec4a 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py +++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py @@ -30,7 +30,7 @@ run_multi_session, ) from tests.helpers.mark import hardware_test -from vllm_omni.clients.duplex import build_realtime_url +from vllm_omni.clients.duplex import build_realtime_url, metric_mean from vllm_omni.experimental.fullduplex.video_stacking import concat_frames_b64 pytestmark = pytest.mark.omni @@ -59,9 +59,12 @@ def _assert_session_metrics(metrics: object, *, expected_count: int) -> None: assert isinstance(metrics, dict) assert isinstance(metrics["session_id"], str) assert metrics["audio_turn_count"] == expected_count - assert metrics["mean_ttft_ms"] is not None and metrics["mean_ttft_ms"] >= 0 - assert metrics["mean_ttfp_ms"] is not None and metrics["mean_ttfp_ms"] >= 0 - assert metrics["mean_rtf"] is not None and metrics["mean_rtf"] >= 0 + ttft_ms = metric_mean(metrics["ttft_ms"]) + ttfp_ms = metric_mean(metrics["ttfp_ms"]) + rtf = metric_mean(metrics["rtf"]) + assert ttft_ms is not None and ttft_ms >= 0 + assert ttfp_ms is not None and ttfp_ms >= 0 + assert rtf is not None and rtf >= 0 async def _receive_protocol_events(ws, required_types: set[str], *, timeout_s: float) -> list[dict[str, object]]: diff --git a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py index 821e505741c..393510d9c5b 100644 --- a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py +++ b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py @@ -1200,9 +1200,9 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity(): assert session_metrics == { "session_id": "seed-tts-session", "audio_turn_count": 2, - "mean_ttft_ms": 150.0, - "mean_ttfp_ms": 250.0, - "mean_rtf": 3.125, + "ttft_ms": {"count": 2, "mean": 150.0, "p50": 150.0, "p99": 150.0}, + "ttfp_ms": {"count": 2, "mean": 250.0, "p50": 200.0, "p99": 300.0}, + "rtf": {"count": 2, "mean": 3.125, "p50": 2.5, "p99": 3.75}, } diff --git a/vllm_omni/benchmarks/duplex_session_metrics.py b/vllm_omni/benchmarks/duplex_session_metrics.py index 388adadca34..ba3bab5ac33 100644 --- a/vllm_omni/benchmarks/duplex_session_metrics.py +++ b/vllm_omni/benchmarks/duplex_session_metrics.py @@ -6,7 +6,7 @@ This module does not instrument the server. It reads the same ``EventCollector.timing_summary`` / ``global_timing_summary`` surfaces that OmniInteract already uses, derives RTF with ``compute_audio_rtf``, and folds -per-session ``global_*`` values into a run-level report. +per-session ``stream_*`` values into a run-level report. """ from __future__ import annotations @@ -29,7 +29,7 @@ "tpot": "Stage-0 engine mean time per output token", "rtf": "response.created client receive to last audio packet divided by emitted audio duration", } -_GLOBAL_MEASUREMENT_ORIGIN = { +_STREAM_MEASUREMENT_ORIGIN = { "ttft": "input stream start to first non-empty text delta", "ttfp": "input stream start to first audio packet", "rtf": ( @@ -41,7 +41,7 @@ @dataclass(frozen=True) class DuplexSessionMetricBundle: - """One session's request rows plus the session/global summary.""" + """One session's request rows plus the session/stream summary.""" request_metrics: list[dict[str, object]] session_metrics: dict[str, object] @@ -63,11 +63,11 @@ def collect_duplex_session_metrics( stream_start: float, session_id: str | None, ) -> DuplexSessionMetricBundle: - """Build per-response and session/global metrics for one duplex session. + """Build per-response and session/stream metrics for one duplex session. ``stream_start`` is the client monotonic time just before input media is pushed. Per-response TTFT/TTFP prefer server ``response_request_metrics`` - when present; RTF and the session ``global_*`` window stay client-receive. + when present; RTF and the session ``stream_*`` window stay client-receive. """ from vllm_omni.clients.duplex import summarize_session_request_metrics @@ -99,21 +99,21 @@ def collect_duplex_session_metrics( request_metrics, session_id=session_id, ) - global_metrics = collector.global_timing_summary( + stream_metrics = collector.global_timing_summary( after_s=stream_start, window_started_at_s=stream_start, response_ids=list(collector.response_ids), - measurement_origin=_GLOBAL_MEASUREMENT_ORIGIN, + measurement_origin=_STREAM_MEASUREMENT_ORIGIN, ) - if global_metrics: + if stream_metrics: session_metrics.update( { - "global_ttft_ms": global_metrics.get("ttft_ms"), - "global_ttfp_ms": global_metrics.get("ttfp_ms"), - "global_rtf": audio_rtf_from_raw_metric(global_metrics), - "global_audio_generation_ms": global_metrics.get("audio_generation_ms"), - "global_audio_duration_ms": global_metrics.get("audio_duration_ms"), - "global_measurement_origin": global_metrics.get("measurement_origin"), + "stream_ttft_ms": stream_metrics.get("ttft_ms"), + "stream_ttfp_ms": stream_metrics.get("ttfp_ms"), + "stream_rtf": audio_rtf_from_raw_metric(stream_metrics), + "stream_audio_generation_ms": stream_metrics.get("audio_generation_ms"), + "stream_audio_duration_ms": stream_metrics.get("audio_duration_ms"), + "stream_measurement_origin": stream_metrics.get("measurement_origin"), } ) return DuplexSessionMetricBundle( @@ -123,13 +123,15 @@ def collect_duplex_session_metrics( ) -def mean_duplex_global_metrics(session_metrics: Sequence[Mapping[str, object]]) -> dict[str, float]: - """Average finite, non-negative per-session ``global_*`` values.""" - result: dict[str, float] = {} - for session_key, result_key in ( - ("global_ttft_ms", "mean_duplex_global_ttft_ms"), - ("global_ttfp_ms", "mean_duplex_global_ttfp_ms"), - ("global_rtf", "mean_duplex_global_rtf"), +def duplex_stream_metrics(session_metrics: Sequence[Mapping[str, object]]) -> dict[str, object]: + """Summarize finite, non-negative per-session ``stream_*`` values.""" + from vllm_omni.clients.duplex import distribution_summary + + result: dict[str, object] = {} + for session_key, result_key, digits in ( + ("stream_ttft_ms", "duplex_stream_ttft_ms", 3), + ("stream_ttfp_ms", "duplex_stream_ttfp_ms", 3), + ("stream_rtf", "duplex_stream_rtf", 6), ): values = [ float(value) @@ -139,8 +141,9 @@ def mean_duplex_global_metrics(session_metrics: Sequence[Mapping[str, object]]) and math.isfinite(value) and value >= 0 ] - if values: - result[result_key] = sum(values) / len(values) + summary = distribution_summary(values, digits=digits) + if summary is not None: + result[result_key] = summary return result @@ -154,7 +157,7 @@ def build_duplex_metrics_report( "duplex_request_metrics": [dict(metric) for metric in request_metrics], "duplex_session_metrics": [dict(metric) for metric in session_metrics], } - report.update(mean_duplex_global_metrics(session_metrics)) + report.update(duplex_stream_metrics(session_metrics)) return report diff --git a/vllm_omni/benchmarks/metrics/metrics.py b/vllm_omni/benchmarks/metrics/metrics.py index 52eec86b9c7..759e606671f 100644 --- a/vllm_omni/benchmarks/metrics/metrics.py +++ b/vllm_omni/benchmarks/metrics/metrics.py @@ -916,11 +916,9 @@ def calculate_metrics( all_tpots.append(tpot) itls += outputs[i].itl session_metrics = getattr(outputs[i], "duplex_session_metrics", None) - ttft_measured = not isinstance(session_metrics, dict) or session_metrics.get("mean_ttft_ms") is not None - audio_ttfp_measured = ( - not isinstance(session_metrics, dict) or session_metrics.get("mean_ttfp_ms") is not None - ) - audio_rtf_measured = not isinstance(session_metrics, dict) or session_metrics.get("mean_rtf") is not None + ttft_measured = not isinstance(session_metrics, dict) or session_metrics.get("ttft_ms") is not None + audio_ttfp_measured = not isinstance(session_metrics, dict) or session_metrics.get("ttfp_ms") is not None + audio_rtf_measured = not isinstance(session_metrics, dict) or session_metrics.get("rtf") is not None if ttft_measured: ttfts.append(outputs[i].ttft) if audio_ttfp_measured: diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index c0fafa3da57..a230b9c2e12 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -2272,9 +2272,11 @@ async def _async_request_omniinteract( output.audio_duration = case_result.audio_bytes / (24_000 * 2) output.audio_frames = case_result.audio_bytes // 2 session_metrics = case_result.duplex_session_metrics - output.ttft = float(session_metrics.get("mean_ttft_ms") or 0.0) / 1000.0 - output.audio_ttfp = float(session_metrics.get("mean_ttfp_ms") or 0.0) / 1000.0 - output.audio_rtf = float(session_metrics.get("mean_rtf") or 0.0) + from vllm_omni.clients.duplex import metric_mean + + output.ttft = (metric_mean(session_metrics.get("ttft_ms")) or 0.0) / 1000.0 + output.audio_ttfp = (metric_mean(session_metrics.get("ttfp_ms")) or 0.0) / 1000.0 + output.audio_rtf = metric_mean(session_metrics.get("rtf")) or 0.0 token_timing_measured = _apply_stage0_token_timings( output, [request_metric.get("stage0_tokens") for request_metric in case_result.duplex_request_metrics], @@ -2524,9 +2526,11 @@ async def async_request_openai_realtime_duplex( await client.close_session(timeout_s=30.0) output.generated_text = " ".join(filter(None, turn_transcripts)) - output.ttft = float(session_metrics.get("mean_ttft_ms") or 0.0) / 1000.0 - output.audio_ttfp = float(session_metrics.get("mean_ttfp_ms") or 0.0) / 1000.0 - output.audio_rtf = float(session_metrics.get("mean_rtf") or 0.0) + from vllm_omni.clients.duplex import metric_mean + + output.ttft = (metric_mean(session_metrics.get("ttft_ms")) or 0.0) / 1000.0 + output.audio_ttfp = (metric_mean(session_metrics.get("ttfp_ms")) or 0.0) / 1000.0 + output.audio_rtf = metric_mean(session_metrics.get("rtf")) or 0.0 output.audio_duration = ( sum(float(metric.get("audio_duration_ms") or 0.0) for metric in turn_metrics) / 1000.0 ) @@ -2926,7 +2930,7 @@ async def probe_loop(): def measured_ttft(output: RequestFuncOutput) -> float | None: session_metrics = getattr(output, "duplex_session_metrics", None) - if isinstance(session_metrics, dict) and session_metrics.get("mean_ttft_ms") is None: + if isinstance(session_metrics, dict) and session_metrics.get("ttft_ms") is None: return None return output.ttft @@ -2998,10 +3002,12 @@ def measured_ttft(output: RequestFuncOutput) -> float | None: ] if duplex_session_metrics: result["duplex_session_metrics"] = duplex_session_metrics - for session_key, result_key in ( - ("global_ttft_ms", "mean_duplex_global_ttft_ms"), - ("global_ttfp_ms", "mean_duplex_global_ttfp_ms"), - ("global_rtf", "mean_duplex_global_rtf"), + from vllm_omni.clients.duplex import distribution_summary + + for session_key, result_key, digits in ( + ("stream_ttft_ms", "duplex_stream_ttft_ms", 3), + ("stream_ttfp_ms", "duplex_stream_ttfp_ms", 3), + ("stream_rtf", "duplex_stream_rtf", 6), ): values = [ float(value) @@ -3011,8 +3017,9 @@ def measured_ttft(output: RequestFuncOutput) -> float | None: and np.isfinite(value) and value >= 0 ] - if values: - result[result_key] = sum(values) / len(values) + summary = distribution_summary(values, digits=digits) + if summary is not None: + result[result_key] = summary if omniinteract_summary is not None: result["omniinteract"] = omniinteract_summary diff --git a/vllm_omni/clients/duplex.py b/vllm_omni/clients/duplex.py index d6917b85f30..282a934be8e 100644 --- a/vllm_omni/clients/duplex.py +++ b/vllm_omni/clients/duplex.py @@ -35,7 +35,7 @@ import random import time import wave -from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Sequence +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Protocol @@ -80,6 +80,8 @@ "image_data_url", "read_pcm16_wav", "reference_audio_data_url", + "distribution_summary", + "metric_mean", "summarize_session_request_metrics", "wait_for_condition", "write_pcm16_wav", @@ -1277,6 +1279,36 @@ def nearest_rank(percentile: float) -> float: } +def distribution_summary(values: Sequence[float], *, digits: int = 3) -> dict[str, float | int] | None: + """Summarize values as ``{count, mean, p50, p99}`` for duplex report fields.""" + clean = sorted(float(value) for value in values if math.isfinite(float(value))) + if not clean: + return None + + def nearest_rank(percentile: float) -> float: + index = max(0, math.ceil(percentile * len(clean)) - 1) + return clean[min(index, len(clean) - 1)] + + return { + "count": len(clean), + "mean": round(sum(clean) / len(clean), digits), + "p50": round(nearest_rank(0.50), digits), + "p99": round(nearest_rank(0.99), digits), + } + + +def metric_mean(value: object) -> float | None: + """Read a scalar mean, or the ``mean`` field of a distribution summary.""" + if isinstance(value, Mapping): + nested = value.get("mean") + if isinstance(nested, int | float) and not isinstance(nested, bool) and math.isfinite(float(nested)): + return float(nested) + return None + if isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(float(value)): + return float(value) + return None + + def _event_stage_metrics(event: dict[str, object]) -> dict[str, object] | None: candidates: list[object] = [event.get("vllm_omni")] metadata = event.get("metadata") @@ -1816,18 +1848,20 @@ def summarize_session_request_metrics( *, session_id: str | None, ) -> dict[str, object]: - """Average client- and engine-observed metrics across audio responses. + """Summarize client- and engine-observed metrics across audio responses. ``request_metrics`` entries are caller-assembled dicts; keys that are absent or non-numeric in an entry are simply skipped. ``rtf`` is not produced by :meth:`EventCollector.timing_summary` (which reports raw data - only) — callers that want ``mean_rtf`` add an ``rtf`` value per turn, + only) — callers that want session ``rtf`` add an ``rtf`` value per turn, e.g. via ``vllm_omni.metrics.definitions.compute_audio_rtf``. Zero or - missing ``tpot_ms`` values are omitted from ``mean_tpot_ms``. + missing ``tpot_ms`` values are omitted from session ``tpot_ms``. + + Aggregatable fields are nested as ``{count, mean, p50, p99}``. """ - def mean(metric: str, *, digits: int = 3, positive: bool = False) -> float | None: - values = [ + def values(metric: str, *, positive: bool = False) -> list[float]: + return [ float(request[metric]) for request in request_metrics if isinstance(request.get(metric), int | float) @@ -1835,17 +1869,16 @@ def mean(metric: str, *, digits: int = 3, positive: bool = False) -> float | Non and math.isfinite(float(request[metric])) and (not positive or float(request[metric]) > 0) ] - return round(sum(values) / len(values), digits) if values else None summary: dict[str, object] = { "session_id": session_id, "audio_turn_count": len(request_metrics), - "mean_ttft_ms": mean("ttft_ms"), - "mean_ttfp_ms": mean("ttfp_ms"), - "mean_rtf": mean("rtf", digits=6), + "ttft_ms": distribution_summary(values("ttft_ms")), + "ttfp_ms": distribution_summary(values("ttfp_ms")), + "rtf": distribution_summary(values("rtf"), digits=6), } - if (mean_tpot_ms := mean("tpot_ms", positive=True)) is not None: - summary["mean_tpot_ms"] = mean_tpot_ms + if (tpot := distribution_summary(values("tpot_ms", positive=True))) is not None: + summary["tpot_ms"] = tpot return summary diff --git a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py index 172108f2f77..748100c5d24 100644 --- a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py +++ b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py @@ -6,7 +6,7 @@ ``generate`` still writes per-sample timed sentences and ``*.meta.json``. Duplex performance metrics are written to ``/duplex_metrics.json``. Skipped samples keep previously recorded rows; incoming ``(split, sample_id)`` -rows replace the matching ones, then ``mean_duplex_global_*`` is recomputed. +rows replace the matching ones, then ``duplex_stream_*`` is recomputed. """ import argparse