diff --git a/tests/benchmarks/metrics/test_metrics.py b/tests/benchmarks/metrics/test_metrics.py index e7d782f9e..c57dac9b7 100644 --- a/tests/benchmarks/metrics/test_metrics.py +++ b/tests/benchmarks/metrics/test_metrics.py @@ -17,10 +17,24 @@ from vllm_omni.benchmarks.metrics.metrics import ( print_stage_durations_metrics, ) from vllm_omni.benchmarks.patch.patch import MixRequestFuncOutput +from vllm_omni.metrics.definitions import stage_modality_flags pytestmark = [pytest.mark.core_model, pytest.mark.benchmark, pytest.mark.cpu] +def test_stage_modality_flags_match_print_stage(): + text = stage_modality_flags("text", "") + audio = stage_modality_flags("", "audio") + stream = stage_modality_flags("latent", "stream") + video_frames = stage_modality_flags("video", "image") + + assert text.is_text_stage and not text.is_audio_stage + assert audio.is_audio_stage and not audio.is_internal_stream_stage + assert stream.is_internal_stream_stage and not stream.is_text_stage + assert video_frames.is_video_stage and not video_frames.is_image_stage + assert not stage_modality_flags(None, None).is_text_stage + + def test_tpot_matches_mean_itl_per_request(): """TPOT and pooled ITL should agree when output_len tracks ITL samples.""" output = MixRequestFuncOutput() diff --git a/tests/benchmarks/test_omniinteract.py b/tests/benchmarks/test_omniinteract.py index 1a6acaca5..d8c129929 100644 --- a/tests/benchmarks/test_omniinteract.py +++ b/tests/benchmarks/test_omniinteract.py @@ -198,6 +198,12 @@ def test_response_metrics_include_engine_tpot_and_stream_window(): "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}, + "stages": { + "0": { + "ttft_ms": {"count": 2, "mean": 0.0, "p50": 0.0, "p99": 0.0}, + "tpot_ms": {"count": 2, "mean": 15.0, "p50": 10.0, "p99": 20.0}, + } + }, "stream_ttft_ms": 200.0, "stream_ttfp_ms": 300.0, "stream_rtf": 8.333333, diff --git a/tests/clients/test_duplex_client.py b/tests/clients/test_duplex_client.py index 7313828ce..63d4c282a 100644 --- a/tests/clients/test_duplex_client.py +++ b/tests/clients/test_duplex_client.py @@ -42,6 +42,7 @@ from vllm_omni.clients.duplex import ( read_pcm16_wav, reference_audio_data_url, summarize_session_request_metrics, + summarize_stage_metrics, write_pcm16_wav, ) from vllm_omni.clients.inline_duplex import InlineDuplexClient @@ -1066,6 +1067,7 @@ def test_event_collector_reports_engine_token_and_audio_intervals(): "max": 18.0, }, } + assert timing["stages"] == {"0": timing["stage0_tokens"]} assert timing["audio_output"] == { "source": "client_monotonic_receive", "chunk_count": 3, @@ -1321,6 +1323,152 @@ def test_summarize_session_request_metrics_averages_audio_turns(): } +def test_event_collector_reports_all_engine_stages(): + collector = EventCollector() + collector.add({"type": "response.created", "response": {"id": "resp-a"}}, received_at_s=10.0) + collector.add( + { + "type": "response.output_audio.delta", + "response_id": "resp-a", + "delta": base64.b64encode(b"audio").decode("ascii"), + "sample_rate_hz": 16_000, + "metadata": { + "audio_duration_ms": 80, + "vllm_omni": { + "stage_metrics": { + "0": { + "final_output_type": "text", + "num_tokens_out": 4, + "vllm_ttft_ms": 120.0, + "vllm_tpot_ms": 15.0, + "vllm_itls_ms": [10.0, 20.0], + }, + "1": { + "output_unit_type": "stream", + "output_unit_count": 8, + "serving_time_to_first_output_ms": 40.0, + "time_per_output_unit_ms": 5.0, + "inter_output_latencies_ms": [5.0], + }, + } + }, + }, + }, + received_at_s=10.2, + ) + collector.add( + { + "type": "response.output_audio.delta", + "response_id": "resp-a", + "delta": base64.b64encode(b"audio").decode("ascii"), + "sample_rate_hz": 16_000, + "metadata": { + "audio_duration_ms": 160, + "vllm_omni": { + "stage_metrics": { + "2": { + "final_output_type": "audio", + "output_unit_type": "audio", + "serving_time_to_first_output_ms": 80.0, + "output_unit_count": 2, + "audio_generated_frames": 2, + "audio_duration_s": 0.08, + } + } + }, + }, + }, + received_at_s=10.3, + ) + + timing = collector.timing_summary(after_s=10.0, response_id="resp-a") + + assert list(timing["stages"]) == ["0", "1", "2"] + assert timing["stages"]["0"] is timing["stage0_tokens"] + assert timing["stages"]["0"]["ttft_ms"] == 120.0 + assert "ttft_ms" not in timing["stages"]["1"] + assert timing["stages"]["1"]["ttfc_ms"] == 40.0 + assert timing["stages"]["1"]["tpop_ms"] == 5.0 + assert "ttft_ms" not in timing["stages"]["2"] + assert timing["stages"]["2"]["ttfp_ms"] == 80.0 + + +def test_audio_stage_omits_missing_or_zero_ttfp(): + def _audio_stage_timing(serving_time: object) -> dict[str, object]: + collector = EventCollector() + collector.add({"type": "response.created", "response": {"id": "resp-a"}}, received_at_s=10.0) + stage_2: dict[str, object] = { + "final_output_type": "audio", + "output_unit_type": "audio", + "output_unit_count": 1, + "audio_generated_frames": 1, + "audio_duration_s": 0.08, + } + if serving_time is not None: + stage_2["serving_time_to_first_output_ms"] = serving_time + collector.add( + { + "type": "response.output_audio.delta", + "response_id": "resp-a", + "delta": base64.b64encode(b"audio").decode("ascii"), + "sample_rate_hz": 16_000, + "metadata": { + "audio_duration_ms": 80, + "vllm_omni": {"stage_metrics": {"2": stage_2}}, + }, + }, + received_at_s=10.2, + ) + return collector.timing_summary(after_s=10.0, response_id="resp-a")["stages"]["2"] + + assert "ttfp_ms" not in _audio_stage_timing(None) + assert "ttfp_ms" not in _audio_stage_timing(0.0) + assert _audio_stage_timing(80.0)["ttfp_ms"] == 80.0 + + +def test_summarize_session_request_metrics_groups_stages(): + summary = summarize_session_request_metrics( + [ + { + "ttft_ms": 100.0, + "stages": { + "0": {"ttft_ms": 10.0, "tpot_ms": 2.0}, + "1": {"ttfc_ms": 20.0, "tpop_ms": 0.0}, + }, + }, + { + "ttft_ms": 300.0, + "stages": { + "0": {"ttft_ms": 30.0, "tpot_ms": 6.0}, + "1": {"ttfc_ms": 40.0, "tpop_ms": 8.0}, + "2": {"ttfp_ms": 50.0}, + }, + }, + ], + session_id="sess-1", + ) + + assert summary["stages"] == { + "0": { + "ttft_ms": {"count": 2, "mean": 20.0, "p50": 10.0, "p99": 30.0}, + "tpot_ms": {"count": 2, "mean": 4.0, "p50": 2.0, "p99": 6.0}, + }, + "1": { + "ttfc_ms": {"count": 2, "mean": 30.0, "p50": 20.0, "p99": 40.0}, + "tpop_ms": {"count": 1, "mean": 8.0, "p50": 8.0, "p99": 8.0}, + }, + "2": { + "ttfp_ms": {"count": 1, "mean": 50.0, "p50": 50.0, "p99": 50.0}, + }, + } + assert summarize_stage_metrics([{"stage0_tokens": {"ttft_ms": 12.0, "tpot_ms": 3.0}}]) == { + "0": { + "ttft_ms": {"count": 1, "mean": 12.0, "p50": 12.0, "p99": 12.0}, + "tpot_ms": {"count": 1, "mean": 3.0, "p50": 3.0, "p99": 3.0}, + } + } + + def test_pcm16_wav_round_trip(tmp_path): path = tmp_path / "audio.wav" pcm16 = b"\x01\x00\x02\x00" 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 b1b02ef65..096fb8789 100644 --- a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py +++ b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py @@ -417,6 +417,63 @@ def test_collect_duplex_session_metrics_matches_omniinteract_window(): assert report["duplex_stream_ttfp_ms"]["mean"] == 300.0 +def test_collect_duplex_session_metrics_keeps_all_engine_stages(): + collector = EventCollector() + audio = { + "type": "response.output_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, + "vllm_omni": { + "stage_metrics": { + "0": { + "final_output_type": "text", + "num_tokens_out": 4, + "vllm_ttft_ms": 12.0, + "vllm_tpot_ms": 3.0, + "vllm_itls_ms": [3.0], + }, + "1": { + "output_unit_type": "stream", + "output_unit_count": 8, + "serving_time_to_first_output_ms": 40.0, + "time_per_output_unit_ms": 5.0, + "inter_output_latencies_ms": [5.0], + }, + "2": { + "final_output_type": "audio", + "serving_time_to_first_output_ms": 80.0, + }, + } + }, + }, + } + 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") + stages = bundle.request_metrics[0]["stages"] + assert list(stages) == ["0", "1", "2"] + assert stages["0"] == bundle.request_metrics[0]["stage0_tokens"] + assert stages["1"]["ttfc_ms"] == 40.0 + assert stages["1"]["tpop_ms"] == 5.0 + assert stages["2"]["ttfp_ms"] == 80.0 + assert bundle.session_metrics["stages"]["0"]["tpot_ms"]["mean"] == 3.0 + assert bundle.session_metrics["stages"]["2"]["ttfp_ms"]["mean"] == 80.0 + report = build_duplex_metrics_report( + request_metrics=bundle.request_metrics, + session_metrics=[bundle.session_metrics], + ) + assert "duplex_stream_stages" not in report + + def test_judge_exercises_openai_http_schema(): requests = [] diff --git a/tests/engine/duplex/test_duplex_events.py b/tests/engine/duplex/test_duplex_events.py index 6330059a1..f369d9921 100644 --- a/tests/engine/duplex/test_duplex_events.py +++ b/tests/engine/duplex/test_duplex_events.py @@ -33,7 +33,9 @@ from vllm_omni.engine.duplex.events import ( SessionReplaced, SessionResumed, SessionResyncRequired, + Speak, TextDelta, + TranscriptDelta, TranscriptDone, TurnEvent, error_event, @@ -119,6 +121,18 @@ def test_optional_wire_fields_are_omitted_when_none(): assert delta["output_index"] == 0 assert delta["content_index"] == 0 + transcript = TranscriptDelta(response_id="resp_1", item_id="item_resp_1", delta="hi").to_realtime() + assert "metadata" not in transcript + assert ( + TranscriptDelta( + response_id="resp_1", + item_id="item_resp_1", + delta="hi", + metadata={"vllm_omni": {"response_request_metrics": {"ttft_ms": 200.0}}}, + ).to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"]["ttft_ms"] + == 200.0 + ) + anonymous_listen = Listen(session_id="sid", epoch=0, details={"reason": "silence"}).to_realtime() assert "response_id" not in anonymous_listen assert "id" not in anonymous_listen["response"] @@ -478,3 +492,87 @@ def test_projection_leaves_error_to_typed_emit_sites(): assert isinstance(projected[0], DuplexRawEvent) assert projected[0].type == "duplex.error" assert projected[0].to_realtime()["event"]["code"] == "bad_event" + + +def test_response_speak_projection_keeps_vllm_omni_request_metrics(): + state = RealtimeProjectionState(session_id="duplex-proj", model="test-model") + project_internal_event(state, {"type": "response.created", "response_id": "resp_1"}) + metrics = { + "source": "server_monotonic_request_start", + "ttft_ms": 200.0, + "measurement_origin": {"ttft": "text", "ttfp": "audio"}, + } + + events = project_internal_event( + state, + { + "type": "response.speak", + "response_id": "resp_1", + "session_id": "duplex-proj", + "epoch": 0, + "model_speak": True, + "vllm_omni": {"response_request_metrics": metrics}, + }, + ) + + assert _types(events) == ["response.speak"] + assert isinstance(events[0], Speak) + assert events[0].to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"]["ttft_ms"] == 200.0 + + +def test_text_then_audio_projects_ttft_on_transcript_and_ttfp_on_audio_delta(): + state = RealtimeProjectionState(session_id="duplex-proj", model="test-model") + project_internal_event( + state, {"type": "response.created", "response_id": "resp_1", "modalities": ["audio", "text"]} + ) + ttft = { + "source": "server_monotonic_request_start", + "ttft_ms": 200.0, + "measurement_origin": {"ttft": "text", "ttfp": "audio"}, + } + + text_only = project_internal_event( + state, + { + "type": "response.output_audio.delta", + "response_id": "resp_1", + "audio": "", + "text": "hi", + "session_id": "duplex-proj", + "epoch": 0, + "model_speak": True, + "vllm_omni": {"response_request_metrics": ttft}, + }, + ) + + assert _types(text_only) == ["response.output_audio_transcript.delta"] + assert isinstance(text_only[0], TranscriptDelta) + text_metrics = text_only[0].to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"] + assert text_metrics["ttft_ms"] == 200.0 + assert "ttfp_ms" not in text_metrics + + pcm = base64.b64encode(b"\x00\x10" * 8).decode("ascii") + ttfp = { + "source": "server_monotonic_request_start", + "ttft_ms": 200.0, + "ttfp_ms": 400.0, + "measurement_origin": {"ttft": "text", "ttfp": "audio"}, + } + audio = project_internal_event( + state, + { + "type": "response.output_audio.delta", + "response_id": "resp_1", + "audio": pcm, + "text": "", + "format": "pcm16", + "vllm_omni": {"response_request_metrics": ttfp}, + }, + ) + + assert "response.output_audio.delta" in _types(audio) + audio_delta = next(event for event in audio if event.type == "response.output_audio.delta") + assert isinstance(audio_delta, AudioDelta) + audio_metrics = audio_delta.to_realtime()["metadata"]["vllm_omni"]["response_request_metrics"] + assert audio_metrics["ttft_ms"] == 200.0 + assert audio_metrics["ttfp_ms"] == 400.0 diff --git a/tests/engine/duplex/test_engine_session.py b/tests/engine/duplex/test_engine_session.py index 358ee2d9c..a4e4b1e9d 100644 --- a/tests/engine/duplex/test_engine_session.py +++ b/tests/engine/duplex/test_engine_session.py @@ -5,6 +5,8 @@ from __future__ import annotations +import time +from collections.abc import Callable from dataclasses import FrozenInstanceError import pytest @@ -20,8 +22,13 @@ from vllm_omni.engine.duplex.config import ( ) from vllm_omni.engine.duplex.contracts import DuplexFence from vllm_omni.engine.duplex.events import TurnEvent -from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession, DuplexFenceMismatchError +from vllm_omni.engine.duplex.session.engine_session import ( + RESPONSE_REQUEST_MEASUREMENT_ORIGIN, + DuplexEngineSession, + DuplexFenceMismatchError, +) from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack +from vllm_omni.metrics.stats import DUPLEX_STAGE_TABLE_EXCLUDE, OrchestratorAggregator, StageRequestStats, StageStats from vllm_omni.model_executor.models.minicpmo_4_5.duplex.capabilities import ( minicpmo45_native_capabilities, ) @@ -29,8 +36,51 @@ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.capabilities import ( pytestmark = [pytest.mark.core_model, pytest.mark.cpu] -def _session(session_id: str = "duplex-test", config: DuplexSessionConfig | None = None) -> DuplexEngineSession: - return DuplexEngineSession(session_id=session_id, config=config or DuplexSessionConfig(model="test-model")) +def _session( + session_id: str = "duplex-test", + config: DuplexSessionConfig | None = None, + *, + num_stages: int = 1, + log_stats: bool = False, + clock: Callable[[], float] | None = None, +) -> DuplexEngineSession: + session = DuplexEngineSession( + session_id=session_id, + config=config or DuplexSessionConfig(model="test-model"), + num_stages=num_stages, + log_stats=log_stats, + ) + if clock is not None: + session._clock = clock + return session + + +def _stage_stats( + *, + stage_id: int, + request_id: str = "stage-req", + num_tokens_out: int = 3, + vllm_ttft_ms: float = 12.0, + vllm_tpot_ms: float = 0.0, + serving_time_to_first_output_ms: float = 0.0, +) -> StageRequestStats: + return StageRequestStats( + batch_id=0, + batch_size=1, + num_tokens_in=7, + num_tokens_out=num_tokens_out, + stage_gen_time_ms=120.0, + rx_transfer_bytes=0, + rx_decode_time_ms=0.0, + rx_in_flight_time_ms=0.0, + stage_stats=StageStats(), + stage_id=stage_id, + request_id=request_id, + final_output_type="text", + vllm_ttft_ms=vllm_ttft_ms, + vllm_tpot_ms=vllm_tpot_ms, + serving_time_to_first_output_ms=serving_time_to_first_output_ms, + ) def test_commit_audio_input_does_not_advance_model_turn_identity(): @@ -733,6 +783,222 @@ def test_minicpmo_native_capabilities_do_not_overclaim_single_session_deployment assert caps["supports_multi_session_same_replica"] is False +def test_response_timing_binds_latest_request_start_for_model_turn(): + session = _session() + session.mark_model_turn_request_started(0, 10.0) + session.mark_model_turn_request_started(0, 11.0) + session.begin_response(turn_id=0) + + first = session.mark_response_first_outputs( + observed_at_s=11.2, + has_text=True, + has_audio=False, + ) + assert first["ttft_ms"] == pytest.approx(200.0) + assert first["measurement_origin"] == RESPONSE_REQUEST_MEASUREMENT_ORIGIN + assert "ttfp_ms" not in first + + # After begin_response the origin is frozen at the pending start (11.0). + # The 12.0 append must not rebind, so TTFP is 12.3-11.0 = 1300 ms, not 300. + session.mark_model_turn_request_started(0, 12.0) + audio = session.mark_response_first_outputs( + observed_at_s=12.3, + has_text=False, + has_audio=True, + ) + assert audio["ttft_ms"] == pytest.approx(200.0) + assert audio["ttfp_ms"] == pytest.approx(1300.0) + + +def test_response_timing_attaches_metrics_only_when_newly_observed(): + session = _session() + session.mark_model_turn_request_started(0, 10.0) + session.begin_response(turn_id=0) + + first = session.mark_response_first_outputs(observed_at_s=10.2, has_text=True, has_audio=False) + assert first["ttft_ms"] == pytest.approx(200.0) + assert session.mark_response_first_outputs(observed_at_s=10.3, has_text=True, has_audio=False) == {} + audio = session.mark_response_first_outputs(observed_at_s=10.4, has_text=False, has_audio=True) + assert audio["ttft_ms"] == pytest.approx(200.0) + assert audio["ttfp_ms"] == pytest.approx(400.0) + assert session.mark_response_first_outputs(observed_at_s=10.5, has_text=True, has_audio=True) == {} + + +def test_response_timing_is_cleared_on_end_barge_in_and_close(): + session = _session() + session.mark_model_turn_request_started(0, 10.0) + session.begin_response(turn_id=0) + assert session.mark_response_first_outputs(observed_at_s=10.2, has_text=True, has_audio=False) + session.end_response() + assert session.mark_response_first_outputs(observed_at_s=10.4, has_text=True, has_audio=False) == {} + + session.mark_model_turn_request_started(0, 11.0) + session.barge_in() + session.begin_response(turn_id=0) + assert session.mark_response_first_outputs(observed_at_s=11.2, has_text=True, has_audio=False) == {} + + session.mark_model_turn_request_started(0, 12.0) + session.close() + session.begin_response(turn_id=0) + assert session.mark_response_first_outputs(observed_at_s=12.2, has_text=True, has_audio=False) == {} + + +def test_complete_model_turn_drops_request_starts_for_finished_turns(): + session = _session() + session.mark_model_turn_request_started(0, 10.0) + session.mark_model_turn_request_started(1, 11.0) + session.complete_model_turn(0) + session.begin_response(turn_id=1) + assert session.mark_response_first_outputs( + observed_at_s=11.25, + has_text=True, + has_audio=False, + )["ttft_ms"] == pytest.approx(250.0) + session.end_response() + session.begin_response(turn_id=0) + assert session.mark_response_first_outputs(observed_at_s=11.5, has_text=True, has_audio=False) == {} + + +def test_log_stats_off_does_not_open_a_response_aggregator(): + session = _session() + session.begin_response() + session.observe_stage_request_stats(0, _stage_stats(stage_id=0)) + assert session._response_aggregator is None + session.end_response() + assert session._response_aggregator is None + + +def test_end_response_logs_pending_and_active_stage_request_stats(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: {}) + session = _session(num_stages=2, log_stats=True) + wire_stats = _stage_stats(stage_id=0, request_id="stage0-req", num_tokens_out=3) + session.observe_stage_request_stats(0, wire_stats) + assert session._response_aggregator is None + assert wire_stats.request_id == "stage0-req" + + response_id = session.begin_response() + aggregator = session._response_aggregator + assert aggregator is not None + assert aggregator.num_stages == 2 + recorded = aggregator.stage_events[response_id] + assert recorded[0].num_tokens_out == 3 + assert recorded[0].request_id == response_id + assert wire_stats.request_id == "stage0-req" + + session.observe_stage_request_stats(1, _stage_stats(stage_id=1, num_tokens_out=5, vllm_ttft_ms=0.0)) + assert [event.stage_id for event in aggregator.stage_events[response_id]] == [0, 1] + + session.end_response() + assert session._response_aggregator is None + assert response_id in aggregator.e2e_done + + +def test_end_response_prints_one_column_per_stage(monkeypatch: pytest.MonkeyPatch): + logged: list[OrchestratorAggregator] = [] + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: logged.append(self) or {}) + session = _session(num_stages=3, log_stats=True) + session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=3, vllm_ttft_ms=40.0)) + response_id = session.begin_response() + session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=4, vllm_ttft_ms=12.0)) + session.observe_stage_request_stats(1, _stage_stats(stage_id=1, num_tokens_out=5, vllm_ttft_ms=20.0)) + session.observe_stage_request_stats(1, _stage_stats(stage_id=1, num_tokens_out=6, vllm_ttft_ms=0.0)) + session.observe_stage_request_stats(2, _stage_stats(stage_id=2, num_tokens_out=0, vllm_ttft_ms=0.0)) + session.end_response() + + assert len(logged) == 1 + rows = logged[0].stage_events[response_id] + assert [event.stage_id for event in rows] == [0, 1, 2] + assert rows[0].num_tokens_out == 7 + assert rows[0].vllm_ttft_ms == pytest.approx(40.0) + assert rows[1].num_tokens_out == 11 + assert rows[1].vllm_ttft_ms == pytest.approx(20.0) + assert logged[0].stage_table_exclude == DUPLEX_STAGE_TABLE_EXCLUDE + + +def test_end_response_omits_serving_time_to_first_output_from_stage_table( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original = OrchestratorAggregator.build_and_log_summary + captured: list[tuple[OrchestratorAggregator, dict[str, object]]] = [] + + def _capture(self: OrchestratorAggregator) -> dict[str, object]: + summary = original(self) + captured.append((self, summary)) + return summary + + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", _capture) + session = _session(num_stages=3, log_stats=True) + response_id = session.begin_response() + session.observe_stage_request_stats(0, _stage_stats(stage_id=0, serving_time_to_first_output_ms=80.631)) + session.observe_stage_request_stats(1, _stage_stats(stage_id=1, serving_time_to_first_output_ms=418.829)) + session.observe_stage_request_stats( + 2, _stage_stats(stage_id=2, vllm_ttft_ms=0.0, serving_time_to_first_output_ms=487.109) + ) + session.end_response() + + assert len(captured) == 1 + aggregator, summary = captured[0] + stage_table = summary.get("stage_table") + assert isinstance(stage_table, list) and stage_table + rows = stage_table[0]["stages"] + assert isinstance(rows, list) + assert [row["stage_id"] for row in rows] == [0, 1, 2] + assert all("serving_time_to_first_output_ms" not in row for row in rows) + assert all("vllm_ttft_ms" in row for row in rows) + events = aggregator.stage_events[response_id] + assert [event.serving_time_to_first_output_ms for event in events] == pytest.approx([80.631, 418.829, 487.109]) + + +def test_http_stage_table_keeps_serving_time_to_first_output() -> None: + agg = OrchestratorAggregator(num_stages=2, log_stats=True, wall_start_ts=0.0, final_stage_id_for_e2e=1) + agg.on_stage_metrics(0, "r1", _stage_stats(stage_id=0, serving_time_to_first_output_ms=80.0)) + agg.on_stage_metrics(1, "r1", _stage_stats(stage_id=1, serving_time_to_first_output_ms=418.0)) + agg.on_finalize_request(1, "r1", req_start_ts=0.0) + + summary = agg.build_and_log_summary() + rows = summary["stage_table"][0]["stages"] + assert [row["serving_time_to_first_output_ms"] for row in rows] == [80.0, 418.0] + + +def test_logged_e2e_includes_wait_before_first_output(monkeypatch: pytest.MonkeyPatch) -> None: + mono = {"t": 100.0} + wall = {"t": 1_000.0} + monkeypatch.setattr(time, "time", lambda: wall["t"]) + logged: list[OrchestratorAggregator] = [] + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: logged.append(self) or {}) + session = _session(log_stats=True, clock=lambda: mono["t"]) + session.mark_model_turn_request_started(0, 100.0) + + mono["t"] = 102.0 + wall["t"] = 1_002.0 + response_id = session.begin_response(turn_id=0) + first = session.mark_response_first_outputs(observed_at_s=102.0, has_text=True, has_audio=True) + assert first["ttft_ms"] == pytest.approx(2000.0) + + mono["t"] = 102.1 + wall["t"] = 1_002.1 + session.end_response() + + assert logged[0].e2e_events[0].request_id == response_id + assert logged[0].e2e_events[0].e2e_total_ms == pytest.approx(2100.0) + + +def test_logged_tpot_matches_client_weighted_aggregation(monkeypatch: pytest.MonkeyPatch) -> None: + logged: list[OrchestratorAggregator] = [] + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", lambda self: logged.append(self) or {}) + session = _session(log_stats=True) + response_id = session.begin_response() + session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=11, vllm_tpot_ms=10.0)) + session.observe_stage_request_stats(0, _stage_stats(stage_id=0, num_tokens_out=3, vllm_tpot_ms=100.0)) + session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 11, "vllm_tpot_ms": 10.0}}) + client = session.accumulate_response_stage_metrics({"0": {"num_tokens_out": 3, "vllm_tpot_ms": 100.0}}) + expected = client["0"]["vllm_tpot_ms"] + session.end_response() + + assert expected == pytest.approx(25.0) + assert logged[0].stage_events[response_id][0].vllm_tpot_ms == pytest.approx(expected) + + def test_draining_response_stays_ack_admissible_after_next_begin_response() -> None: from vllm_omni.engine.duplex.events import ErrorEvent from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack diff --git a/tests/engine/duplex/test_session_runner.py b/tests/engine/duplex/test_session_runner.py index cf6a0243d..92fba2793 100644 --- a/tests/engine/duplex/test_session_runner.py +++ b/tests/engine/duplex/test_session_runner.py @@ -46,9 +46,10 @@ from vllm_omni.engine.duplex.messages import ( DuplexSessionEventMessage, OpenDuplexSessionMessage, ) +from vllm_omni.engine.duplex.session.engine_session import RESPONSE_REQUEST_MEASUREMENT_ORIGIN from vllm_omni.engine.duplex.session.manager import DuplexSessionManager from vllm_omni.engine.duplex.session.runner import DuplexSessionRunner -from vllm_omni.metrics.stats import StageRequestStats, StageStats +from vllm_omni.metrics.stats import OrchestratorAggregator, StageRequestStats, StageStats from vllm_omni.model_executor.models.minicpmo_4_5.duplex.plugin import MiniCPMO45DuplexPlugin pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -202,6 +203,7 @@ async def open_harness( runtime_config: DuplexSessionRuntimeConfig | None = None, stage_count: int = 2, clock: Any = None, + log_stats: bool = False, ) -> Harness: plugin = MiniCPMO45DuplexPlugin(_fake_encode_audio) port = RecordingStagePort(stage_count=stage_count) @@ -214,6 +216,7 @@ async def open_harness( result_sink=results, runtime_config=runtime_config or DuplexSessionRuntimeConfig(), model_config=None, + log_stats=log_stats, clock=clock, ) body: dict[str, object] = {"auto_response": auto_response, **(extra_body or {})} @@ -1398,6 +1401,144 @@ async def test_stage0_metrics_from_several_units_are_summed_into_one_response() await close_harness(h) +@pytest.mark.asyncio +async def test_response_done_logs_orchestrator_stage_table(monkeypatch: pytest.MonkeyPatch) -> None: + logged: list[OrchestratorAggregator] = [] + + def _capture(self: OrchestratorAggregator) -> dict[str, object]: + logged.append(self) + return {} + + monkeypatch.setattr(OrchestratorAggregator, "build_and_log_summary", _capture) + h = await open_harness(log_stats=True) + try: + assert h.session.log_stats is True + assert h.session.num_stages == 2 + await h.run(append_audio()) + request_id = h.stage0_request_id() + h.deliver( + SimpleNamespace( + request_id=request_id, + finished=False, + outputs=[SimpleNamespace(text="hi", token_ids=[11], multimodal_output={})], + multimodal_output={}, + ), + stage_id=0, + metrics=stage_stats(stage_id=0, request_id=request_id, num_tokens_out=3), + ) + assert await h.settle() == [] + + events = await h.deliver_and_settle( + tts_output(request_id, samples=24000, text="hi", turn_end=True), + metrics=stage_stats(stage_id=1, request_id=request_id, num_tokens_out=4), + ) + done = find(events, "response.done") + assert done.status == "completed" + assert len(logged) == 1 + aggregator = logged[0] + assert aggregator.num_stages == 2 + stage_ids = [event.stage_id for event in aggregator.stage_events[done.response_id]] + assert stage_ids == [0, 1] + assert done.response_id in aggregator.e2e_done + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_duplex_stage_request_stamps_wall_clock_request_timestamp() -> None: + """serving_time_to_first_output_ms is (first_output_ts - request_timestamp)*1000. + + Duplex used to leave request_timestamp at 0, so the table printed unix_ts*1000. + """ + from tests.engine.test_duplex_orchestrator import ( + SESSION_ID as ORCH_SESSION_ID, + ) + from tests.engine.test_duplex_orchestrator import ( + _build, + _close, + _open, + _stage0_request_id, + ) + from vllm_omni.engine.duplex_orchestrator import DuplexOrchestratorRequestState + + orchestrator, _, rpc_q, _ = _build() + try: + result = await _open(orchestrator, rpc_q) + assert result.ok + state = orchestrator.request_states[_stage0_request_id()] + assert isinstance(state, DuplexOrchestratorRequestState) + assert state.request_timestamp > 1_000_000_000.0 + finally: + if ORCH_SESSION_ID in orchestrator.session_manager.runners: + await _close(orchestrator, rpc_q) + await orchestrator.session_manager.shutdown() + + +def _response_request_metrics_of(event: object) -> dict[str, object]: + """Server request-start clocks as the client reads them off one wire event.""" + payload = event.to_realtime() + metadata = payload.get("metadata") + assert isinstance(metadata, dict), payload + vllm_omni = metadata.get("vllm_omni") + assert isinstance(vllm_omni, dict), metadata + metrics = vllm_omni.get("response_request_metrics") + assert isinstance(metrics, dict), vllm_omni + return metrics + + +@pytest.mark.asyncio +async def test_first_audio_delta_carries_server_request_start_metrics() -> None: + clock = {"now": 1000.0} + h = await open_harness(clock=lambda: clock["now"]) + try: + await h.run(append_audio()) + request_id = h.stage0_request_id() + clock["now"] = 1001.3 + events = await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hi")) + metrics = _response_request_metrics_of(find(events, "response.output_audio.delta")) + assert metrics["source"] == "server_monotonic_request_start" + assert metrics["measurement_origin"] == dict(RESPONSE_REQUEST_MEASUREMENT_ORIGIN) + assert metrics["ttft_ms"] == pytest.approx(1300.0) + assert metrics["ttfp_ms"] == pytest.approx(1300.0) + speak_metrics = _response_request_metrics_of(find(events, "response.speak")) + assert speak_metrics["ttft_ms"] == pytest.approx(1300.0) + assert speak_metrics["ttfp_ms"] == pytest.approx(1300.0) + + clock["now"] = 1002.0 + later = await h.deliver_and_settle(tts_output(request_id, samples=48000, text="hello")) + later_payload = find(later, "response.output_audio.delta").to_realtime() + later_metadata = later_payload.get("metadata") + later_vllm_omni = later_metadata.get("vllm_omni") if isinstance(later_metadata, dict) else None + later_metrics = later_vllm_omni.get("response_request_metrics") if isinstance(later_vllm_omni, dict) else None + assert later_metrics is None + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_stale_epoch_append_does_not_own_request_start() -> None: + clock = {"now": 1000.0} + h = await open_harness(clock=lambda: clock["now"]) + try: + await h.run(append_audio()) + request_id = h.stage0_request_id() + clock["now"] = 1000.5 + append_ok, emitted = await h.runner.model.append_runtime_input( + {"duplex_turn_id": 0}, + final=False, + expected_epoch=h.session.epoch + 1, + ) + assert append_ok is True + assert emitted is False + clock["now"] = 1001.3 + events = await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hi")) + metrics = _response_request_metrics_of(find(events, "response.output_audio.delta")) + assert metrics["ttft_ms"] == pytest.approx(1300.0) + assert metrics["ttfp_ms"] == pytest.approx(1300.0) + finally: + await close_harness(h) + + # --------------------------------------------------------------------------- # # Server VAD # # --------------------------------------------------------------------------- # diff --git a/vllm_omni/benchmarks/duplex_session_metrics.py b/vllm_omni/benchmarks/duplex_session_metrics.py index ba3bab5ac..bfa606e4d 100644 --- a/vllm_omni/benchmarks/duplex_session_metrics.py +++ b/vllm_omni/benchmarks/duplex_session_metrics.py @@ -81,6 +81,7 @@ def collect_duplex_session_metrics( measurement_origin=_REQUEST_MEASUREMENT_ORIGIN, ) raw_metric = timing.get("request_metrics") + stages = timing.get("stages") stage0 = timing.get("stage0_tokens") metric: dict[str, object] = { "session_id": session_id, @@ -90,10 +91,17 @@ def collect_duplex_session_metrics( if isinstance(raw_metric, dict): metric.update(raw_metric) metric["rtf"] = audio_rtf_from_raw_metric(raw_metric) + copied_stages: dict[str, dict[str, object]] = {} + if isinstance(stages, dict): + for stage_id, stage_snapshot in stages.items(): + if isinstance(stage_snapshot, dict): + copied_stages[str(stage_id)] = dict(stage_snapshot) + if copied_stages: + metric["stages"] = copied_stages 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): + if isinstance(raw_metric, dict) or copied_stages or isinstance(stage0, dict): request_metrics.append(metric) session_metrics = summarize_session_request_metrics( request_metrics, diff --git a/vllm_omni/benchmarks/metrics/metrics.py b/vllm_omni/benchmarks/metrics/metrics.py index 6cfe473d5..eb18050f7 100644 --- a/vllm_omni/benchmarks/metrics/metrics.py +++ b/vllm_omni/benchmarks/metrics/metrics.py @@ -131,14 +131,6 @@ def _p_label(p: float) -> str: return str(pf) -_STREAMING_OUTPUT_UNIT_TYPES = frozenset( - { - "text", - "stream", - "audio", - } -) - _AGGREGATE_PERCENTILE_FIELD_NAMES = { defs.AUDIO_TTFP: ( defs.MEAN_AUDIO_TTFP_MS, @@ -187,28 +179,6 @@ def _wants_stream_icl(metrics: list[str]) -> bool: return "icl" in metrics -def _stage_modality_flags( - final_output_type: str, - output_unit_type: str, -) -> tuple[bool, bool, bool, bool, bool]: - is_text_stage = final_output_type == "text" or output_unit_type == "text" - is_audio_stage = final_output_type == "audio" or output_unit_type == "audio" - is_video_stage = final_output_type in {"video", "videos"} or output_unit_type == "video" - # Video diffusion may still report output_unit_type="image" when frames are - # stored in ``images``; prefer video when final_output_type says so. - is_image_stage = (not is_video_stage) and (final_output_type in {"image", "images"} or output_unit_type == "image") - is_internal_stream_stage = ( - output_unit_type in _STREAMING_OUTPUT_UNIT_TYPES and not is_text_stage and not is_audio_stage - ) - return ( - is_text_stage, - is_audio_stage, - is_image_stage, - is_video_stage, - is_internal_stream_stage, - ) - - def has_metric_samples(metrics: object, metric_name: str) -> bool: sample_count_attr = { "ttft": "num_ttft_samples", @@ -736,7 +706,7 @@ def print_stage_metrics( is_image_stage, is_video_stage, is_internal_stream_stage, - ) = _stage_modality_flags(getattr(sm, "final_output_type"), getattr(sm, "output_unit_type")) + ) = defs.stage_modality_flags(getattr(sm, "final_output_type"), getattr(sm, "output_unit_type")) print("{s:{c}^{n}}".format(s=title, n=50, c="=")) if is_video_stage: @@ -786,7 +756,7 @@ def _build_stage_metrics_from_outputs( ] output_unit_type = output_unit_types[0] if output_unit_types else "other" - is_text_stage, is_audio_stage, _, _, is_internal_stream_stage = _stage_modality_flags( + is_text_stage, is_audio_stage, _, _, is_internal_stream_stage = defs.stage_modality_flags( final_output_type, output_unit_type ) total_output = 0 diff --git a/vllm_omni/clients/duplex.py b/vllm_omni/clients/duplex.py index a11be7b94..3b7688270 100644 --- a/vllm_omni/clients/duplex.py +++ b/vllm_omni/clients/duplex.py @@ -40,7 +40,7 @@ import random import time import wave from abc import ABC, abstractmethod -from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Sequence from dataclasses import dataclass, field from pathlib import Path from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -48,6 +48,18 @@ from uuid import uuid4 import pybase64 as base64 +from vllm_omni.clients.utils import ( + _finite_metric_values, + _finite_number, + _interval_summary, + _rounded_ms, + _stage_engine_metrics_block, + _stage_id_sort_key, + distribution_summary, + metric_mean, + summarize_stage_metrics, +) + __all__ = [ "DUPLEX_FIRST_UNIT_MS", "DUPLEX_UNIT_MS", @@ -89,6 +101,7 @@ __all__ = [ "distribution_summary", "metric_mean", "summarize_session_request_metrics", + "summarize_stage_metrics", "wait_for_condition", "write_pcm16_wav", ] @@ -1339,67 +1352,6 @@ class DuplexClient(DuplexClientBase): # Test/benchmark collector -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: - return {"count": 0, "mean": 0.0, "p50": 0.0, "p95": 0.0, "max": 0.0} - - 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": _rounded_ms(sum(clean) / len(clean)), - "p50": nearest_rank(0.50), - "p95": nearest_rank(0.95), - "max": clean[-1], - } - - -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") @@ -1629,7 +1581,7 @@ class EventCollector: measurement_origin: dict[str, str] | None = None, ) -> dict[str, object]: """Summarize engine token metrics and client-observed audio cadence.""" - stage0_metrics: dict[str, object] | None = None + observed_stage_metrics: dict[str, dict[str, object]] = {} response_request_metrics: dict[str, object] = {} response_created_at_s: float | None = None first_text_received_at_s: float | None = None @@ -1657,9 +1609,10 @@ class EventCollector: first_text_received_at_s = received_at_s stage_metrics = _event_stage_metrics(event) - stage0 = stage_metrics.get("0") if isinstance(stage_metrics, dict) else None - if isinstance(stage0, dict): - stage0_metrics = stage0 + if isinstance(stage_metrics, dict): + for stage_id, stage_snapshot in stage_metrics.items(): + if isinstance(stage_snapshot, dict): + observed_stage_metrics[str(stage_id)] = stage_snapshot event_request_metrics = _event_response_request_metrics(event) if event_request_metrics is not None: @@ -1677,21 +1630,18 @@ class EventCollector: cumulative_audio_ms.append(max(0.0, float(duration_ms))) result: dict[str, object] = {} - if stage0_metrics is not None: - raw_itls = stage0_metrics.get("vllm_itls_ms") - itls = ( - [float(value) for value in raw_itls if isinstance(value, int | float)] - if isinstance(raw_itls, list) - else [] - ) - result["stage0_tokens"] = { - "source": "engine_stage_metrics", - "output_token_count": int(stage0_metrics.get("num_tokens_out") or 0), - "ttft_ms": float(stage0_metrics.get("vllm_ttft_ms") or 0.0), - "tpot_ms": float(stage0_metrics.get("vllm_tpot_ms") or 0.0), - "itls_ms": itls, - "inter_token_interval_ms": _interval_summary(itls), + stage0_metrics = observed_stage_metrics.get("0") + if observed_stage_metrics: + stages = { + stage_id: _stage_engine_metrics_block(stage_snapshot) + for stage_id, stage_snapshot in sorted( + observed_stage_metrics.items(), + key=lambda item: _stage_id_sort_key(item[0]), + ) } + result["stages"] = stages + if "0" in stages: + result["stage0_tokens"] = stages["0"] if audio_received_at_s: intervals_ms = [ @@ -1943,26 +1893,17 @@ def summarize_session_request_metrics( Aggregatable fields are nested as ``{count, mean, p50, p99}``. """ - - 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) - and not isinstance(request.get(metric), bool) - and math.isfinite(float(request[metric])) - and (not positive or float(request[metric]) > 0) - ] - summary: dict[str, object] = { "session_id": session_id, "audio_turn_count": len(request_metrics), - "ttft_ms": distribution_summary(values("ttft_ms")), - "ttfp_ms": distribution_summary(values("ttfp_ms")), - "rtf": distribution_summary(values("rtf"), digits=6), + "ttft_ms": distribution_summary(_finite_metric_values(request_metrics, "ttft_ms")), + "ttfp_ms": distribution_summary(_finite_metric_values(request_metrics, "ttfp_ms")), + "rtf": distribution_summary(_finite_metric_values(request_metrics, "rtf"), digits=6), } - if (tpot := distribution_summary(values("tpot_ms", positive=True))) is not None: + if (tpot := distribution_summary(_finite_metric_values(request_metrics, "tpot_ms", positive=True))) is not None: summary["tpot_ms"] = tpot + if (stages := summarize_stage_metrics(request_metrics)) is not None: + summary["stages"] = stages return summary diff --git a/vllm_omni/clients/utils.py b/vllm_omni/clients/utils.py new file mode 100644 index 000000000..5cfcb15f5 --- /dev/null +++ b/vllm_omni/clients/utils.py @@ -0,0 +1,264 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Client-local helpers for duplex JSON metric blocks.""" + +from __future__ import annotations + +import math +from collections.abc import Mapping, Sequence +from typing import NamedTuple + +__all__ = [ + "STREAMING_OUTPUT_UNIT_TYPES", + "StageModalityFlags", + "distribution_summary", + "metric_mean", + "stage_modality_flags", + "summarize_stage_metrics", +] + +# Intermediate streaming units that are not the pipeline's text or audio +# final output (e.g. MiniCPM Talker codec tokens). +STREAMING_OUTPUT_UNIT_TYPES = frozenset({"text", "stream", "audio"}) + + +class StageModalityFlags(NamedTuple): + """How duplex JSON classifies one engine stage.""" + + is_text_stage: bool + is_audio_stage: bool + is_image_stage: bool + is_video_stage: bool + is_internal_stream_stage: bool + + +def stage_modality_flags( + final_output_type: object, + output_unit_type: object, +) -> StageModalityFlags: + """Classify a stage from wire ``final_output_type`` / ``output_unit_type``.""" + final_type = final_output_type if isinstance(final_output_type, str) else "" + unit_type = output_unit_type if isinstance(output_unit_type, str) else "" + is_text_stage = final_type == "text" or unit_type == "text" + is_audio_stage = final_type == "audio" or unit_type == "audio" + is_video_stage = final_type in {"video", "videos"} or unit_type == "video" + # Video diffusion may still report output_unit_type="image" when frames are + # stored in ``images``; prefer video when final_output_type says so. + is_image_stage = (not is_video_stage) and (final_type in {"image", "images"} or unit_type == "image") + is_internal_stream_stage = unit_type in STREAMING_OUTPUT_UNIT_TYPES and not is_text_stage and not is_audio_stage + return StageModalityFlags( + is_text_stage=is_text_stage, + is_audio_stage=is_audio_stage, + is_image_stage=is_image_stage, + is_video_stage=is_video_stage, + is_internal_stream_stage=is_internal_stream_stage, + ) + + +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: + return {"count": 0, "mean": 0.0, "p50": 0.0, "p95": 0.0, "max": 0.0} + + 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": _rounded_ms(sum(clean) / len(clean)), + "p50": nearest_rank(0.50), + "p95": nearest_rank(0.95), + "max": clean[-1], + } + + +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 _stage_id_sort_key(stage_id: str) -> tuple[int, int | str]: + try: + return (0, int(stage_id)) + except ValueError: + return (1, stage_id) + + +def _stage_type_labels(stage_metrics: Mapping[str, object]) -> dict[str, object]: + labels: dict[str, object] = {} + for key in ("final_output_type", "output_unit_type"): + value = stage_metrics.get(key) + if isinstance(value, str) and value: + labels[key] = value + return labels + + +def _stage_ms_list(stage_metrics: Mapping[str, object], field_name: str) -> list[float]: + raw = stage_metrics.get(field_name) + return [float(value) for value in raw if isinstance(value, int | float)] if isinstance(raw, list) else [] + + +def _has_legacy_vllm_token_metrics(stage_metrics: Mapping[str, object]) -> bool: + if stage_metrics.get("vllm_ttft_ms") or stage_metrics.get("vllm_tpot_ms"): + return True + raw_itls = stage_metrics.get("vllm_itls_ms") + return isinstance(raw_itls, list) and bool(raw_itls) + + +def _text_stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]: + itls = _stage_ms_list(stage_metrics, "vllm_itls_ms") + return { + "source": "engine_stage_metrics", + **_stage_type_labels(stage_metrics), + "output_token_count": int(stage_metrics.get("num_tokens_out") or 0), + "ttft_ms": float(stage_metrics.get("vllm_ttft_ms") or 0.0), + "tpot_ms": float(stage_metrics.get("vllm_tpot_ms") or 0.0), + "itls_ms": itls, + "inter_token_interval_ms": _interval_summary(itls), + } + + +def _audio_stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]: + block: dict[str, object] = { + "source": "engine_stage_metrics", + **_stage_type_labels(stage_metrics), + "output_unit_count": int(stage_metrics.get("output_unit_count") or 0), + "audio_generated_frames": int(stage_metrics.get("audio_generated_frames") or 0), + "audio_duration_s": float(stage_metrics.get("audio_duration_s") or 0.0), + } + serving_time = _finite_number( + stage_metrics.get("serving_time_to_first_output_ms"), + nonnegative=True, + ) + if serving_time: + block["ttfp_ms"] = serving_time + return block + + +def _stream_stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]: + icls = _stage_ms_list(stage_metrics, "inter_output_latencies_ms") + return { + "source": "engine_stage_metrics", + **_stage_type_labels(stage_metrics), + "output_unit_count": int(stage_metrics.get("output_unit_count") or 0), + "ttfc_ms": float(stage_metrics.get("serving_time_to_first_output_ms") or 0.0), + "tpop_ms": float(stage_metrics.get("time_per_output_unit_ms") or 0.0), + "icls_ms": icls, + "inter_chunk_interval_ms": _interval_summary(icls), + } + + +def _stage_engine_metrics_block(stage_metrics: Mapping[str, object]) -> dict[str, object]: + flags = stage_modality_flags( + stage_metrics.get("final_output_type"), + stage_metrics.get("output_unit_type"), + ) + if flags.is_text_stage or ( + not flags.is_audio_stage + and not flags.is_internal_stream_stage + and not flags.is_image_stage + and not flags.is_video_stage + and _has_legacy_vllm_token_metrics(stage_metrics) + ): + return _text_stage_engine_metrics_block(stage_metrics) + if flags.is_audio_stage: + return _audio_stage_engine_metrics_block(stage_metrics) + if flags.is_internal_stream_stage: + return _stream_stage_engine_metrics_block(stage_metrics) + block: dict[str, object] = {"source": "engine_stage_metrics", **_stage_type_labels(stage_metrics)} + gen_time_ms = stage_metrics.get("stage_gen_time_ms") + if isinstance(gen_time_ms, int | float) and not isinstance(gen_time_ms, bool): + block["stage_gen_time_ms"] = float(gen_time_ms) + return block + + +def _finite_metric_values( + rows: Sequence[Mapping[str, object]], + metric: str, + *, + positive: bool = False, +) -> list[float]: + return [ + float(row[metric]) + for row in rows + if isinstance(row.get(metric), int | float) + and not isinstance(row.get(metric), bool) + and math.isfinite(float(row[metric])) + and (not positive or float(row[metric]) > 0) + ] + + +def summarize_stage_metrics( + request_metrics: Sequence[Mapping[str, object]], +) -> dict[str, dict[str, object]] | None: + """Roll per-response engine stage blocks into ``{count, mean, p50, p99}``. + + Reads ``stages`` when present, otherwise ``stage0_tokens`` as stage ``"0"``. + Zero or missing ``tpot_ms`` / ``tpop_ms`` values are omitted, matching + session ``tpot_ms``. + """ + buckets: dict[str, list[Mapping[str, object]]] = {} + for request in request_metrics: + stages = request.get("stages") + if not isinstance(stages, dict): + stage0 = request.get("stage0_tokens") + stages = {"0": stage0} if isinstance(stage0, dict) else {} + for stage_id, stage_snapshot in stages.items(): + if isinstance(stage_snapshot, dict): + buckets.setdefault(str(stage_id), []).append(stage_snapshot) + summary: dict[str, dict[str, object]] = {} + for stage_id in sorted(buckets, key=_stage_id_sort_key): + rows = buckets[stage_id] + stage_summary: dict[str, object] = {} + for metric, positive in ( + ("ttft_ms", False), + ("tpot_ms", True), + ("ttfc_ms", False), + ("tpop_ms", True), + ("ttfp_ms", False), + ): + rolled = distribution_summary(_finite_metric_values(rows, metric, positive=positive)) + if rolled is not None: + stage_summary[metric] = rolled + if stage_summary: + summary[stage_id] = stage_summary + return summary or None diff --git a/vllm_omni/engine/duplex/events.py b/vllm_omni/engine/duplex/events.py index 197dec613..b8b4e84a0 100644 --- a/vllm_omni/engine/duplex/events.py +++ b/vllm_omni/engine/duplex/events.py @@ -4,8 +4,8 @@ """The event vocabulary one duplex session speaks, assembled from two layers. The classes themselves no longer live here. ``vllm_omni.protocol.duplex.events`` -carries the whole vocabulary --- 22 Tier 1 classes re-exported from -``vllm_omni.protocol.realtime.events``, 8 Tier 2 ones and 12 Tier 3 ones --- so +carries the whole vocabulary --- 21 Tier 1 classes re-exported from +``vllm_omni.protocol.realtime.events``, 9 Tier 2 ones and 12 Tier 3 ones --- so a non-duplex Realtime consumer can take the Tier 1 half and its wire rendering without the duplex session control plane (RFC #6592 P0a). The engine imports the duplex module and never the Tier 1 one directly. @@ -74,6 +74,7 @@ from vllm_omni.protocol.duplex.events import ( new_event_id, ) +#: Internal events that terminate a response/session and must never be dropped as stale. DOMAIN_TERMINAL_EVENTS = frozenset( { "response.done", diff --git a/vllm_omni/engine/duplex/realtime_events.py b/vllm_omni/engine/duplex/realtime_events.py index 210154d48..35eb0db5b 100644 --- a/vllm_omni/engine/duplex/realtime_events.py +++ b/vllm_omni/engine/duplex/realtime_events.py @@ -476,7 +476,7 @@ def _response_created_event(event: Mapping[str, object]) -> ResponseCreated: def _response_speak_metadata(event: Mapping[str, object]) -> dict[str, object]: - return {key: event[key] for key in ("session_id", "epoch", "model_speak") if key in event} + return {key: event[key] for key in ("session_id", "epoch", "model_speak", "vllm_omni") if key in event} def _realtime_audio_delta_events( @@ -839,17 +839,19 @@ def _project(state: RealtimeProjectionState, event: dict[str, object]) -> list[D _refresh_in_progress_response_item(state, response_id) text = event.get("text") has_text = isinstance(text, str) and bool(text) + has_audio_delta = isinstance(audio, str) and bool(audio) if has_text: _append_response_transcript(state, response_id, cast("str", text)) _refresh_in_progress_response_item(state, response_id) # Keep the audio.delta + transcript.delta pair invariant even for # text-less units so clients that treat the pair as unit-complete work. - if has_text or (isinstance(audio, str) and bool(audio)): + if has_text or has_audio_delta: events.append( TranscriptDelta( response_id=_str_or_none(response_id), item_id=_response_item_id(state, response_id), delta=cast("str", text) if has_text else "", + metadata=_response_speak_metadata(event) if has_text and not has_audio_delta else None, ) ) if event.get("end_of_turn") is True: diff --git a/vllm_omni/engine/duplex/session/engine_session.py b/vllm_omni/engine/duplex/session/engine_session.py index 6c4d61f69..37ab18d4b 100644 --- a/vllm_omni/engine/duplex/session/engine_session.py +++ b/vllm_omni/engine/duplex/session/engine_session.py @@ -43,6 +43,14 @@ from vllm_omni.engine.duplex.session.lease import ( DuplexLeaseConfig, DuplexLeaseState, ) +from vllm_omni.metrics.stats import ( + DUPLEX_STAGE_TABLE_EXCLUDE, + OrchestratorAggregator, + StageRequestStats, + _one_row_per_stage, + _tpot_interval_weight, +) +from vllm_omni.metrics.utils import _as_int, _as_optional_int if TYPE_CHECKING: from vllm_omni.engine.duplex.plugin import DuplexModelSessionState @@ -57,26 +65,6 @@ def _object_dict(**fields: object) -> dict[str, object]: return dict(fields) -def _as_int(value: object, default: int = 0) -> int: - if isinstance(value, bool): - return default - if isinstance(value, int): - return value - if isinstance(value, float): - return int(value) - return default - - -def _as_optional_int(value: object) -> int | None: - if value is None or isinstance(value, bool): - return None - if isinstance(value, int): - return value - if isinstance(value, float): - return int(value) - return None - - def _copy_mapping(value: object) -> dict[str, object] | None: if not isinstance(value, dict): return None @@ -116,6 +104,22 @@ class ResponseState: stage_metrics: dict[str, dict[str, object]] = field(default_factory=dict) stage_metric_tpot_weighted_ms: dict[str, float] = field(default_factory=dict) stage_metric_tpot_weight: dict[str, int] = field(default_factory=dict) + 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 + + +RESPONSE_REQUEST_MEASUREMENT_ORIGIN: dict[str, str] = { + "ttft": ( + "accepted native-append start to first non-empty text output; " + "pending turns keep the latest append before first output, so overlapping user speech shortens TTFT" + ), + "ttfp": ( + "accepted native-append start to first audio output; " + "pending turns keep the latest append before first output, so overlapping user speech shortens TTFP" + ), +} @dataclass @@ -217,6 +221,14 @@ class DuplexEngineSession: model_state: DuplexModelSessionState | None = field(default=None, repr=False) projector: RealtimeProjectionState | None = field(default=None, repr=False) created_monotonic: float = field(default_factory=time.monotonic) + #: Pipeline width for the per-response ``OrchestratorAggregator`` table. + num_stages: int = 1 + #: Same serve ``log_stats`` gate as HTTP ``build_and_log_summary``. + log_stats: bool = False + _response_aggregator: OrchestratorAggregator | None = field(default=None, repr=False) + #: ``StageRequestStats`` that arrived before ``begin_response`` (stage 0 + #: feeding TTS). Replayed onto the aggregator when the response opens. + _pending_stage_request_stats: list[tuple[int, StageRequestStats]] = field(default_factory=list, repr=False) def __post_init__(self) -> None: if self.accepted_fence is None: @@ -743,6 +755,11 @@ class DuplexEngineSession: 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 self.sync_fence() @@ -827,10 +844,17 @@ class DuplexEngineSession: return "".join(self._response.assistant_text_buffer) def begin_response(self, *, turn_id: int | None = None) -> str: + if self._response_aggregator is not None: + self._log_response_aggregator() self._activate_response_options() 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._clear_response_request_timing() + 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_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 @@ -842,6 +866,7 @@ class DuplexEngineSession: self._playback.current = DuplexPlaybackCursor() self._playback.by_response[response_id] = self._playback.current self.turn_state = DuplexTurnState.ASSISTANT_GENERATING + self._start_response_aggregator() return response_id def _clear_response_metrics(self) -> None: @@ -849,6 +874,120 @@ class DuplexEngineSession: self._response.stage_metric_tpot_weighted_ms.clear() self._response.stage_metric_tpot_weight.clear() + def _response_aggregator_wall_start_ts(self) -> float: + now_wall_s = time.time() + started_at_s = self._response.active_response_request_started_at_s + if started_at_s is None: + return now_wall_s + elapsed_s = max(0.0, self._clock() - started_at_s) + return now_wall_s - elapsed_s + + def _new_response_aggregator(self) -> OrchestratorAggregator: + num_stages = max(int(self.num_stages), 1) + return OrchestratorAggregator( + num_stages=num_stages, + log_stats=True, + wall_start_ts=self._response_aggregator_wall_start_ts(), + final_stage_id_for_e2e=num_stages - 1, + stage_table_exclude=DUPLEX_STAGE_TABLE_EXCLUDE, + ) + + def _start_response_aggregator(self) -> None: + if not self.log_stats: + self._pending_stage_request_stats.clear() + return + self._response_aggregator = self._new_response_aggregator() + pending, self._pending_stage_request_stats = self._pending_stage_request_stats, [] + for stage_id, stats in pending: + self._record_stage_request_stats(stage_id, stats) + + def _record_stage_request_stats(self, stage_id: int, metrics: StageRequestStats) -> None: + aggregator = self._response_aggregator + request_id = self.active_response_id + if aggregator is None or request_id is None: + return + if stage_id < 0 or stage_id >= aggregator.num_stages: + return + aggregator.on_stage_metrics(stage_id, request_id, metrics, metrics.final_output_type) + + def observe_stage_request_stats(self, stage_id: int, metrics: StageRequestStats) -> None: + """Feed one engine ``StageRequestStats`` into the response's logger table.""" + if not self.log_stats: + return + event = copy.copy(metrics) + if event.stage_id is None: + event.stage_id = stage_id + if self.active_response_id is None: + self._pending_stage_request_stats.append((int(stage_id), event)) + return + if self._response_aggregator is None: + self._start_response_aggregator() + self._record_stage_request_stats(int(stage_id), event) + + def _log_response_aggregator(self) -> None: + aggregator = self._response_aggregator + self._response_aggregator = None + self._pending_stage_request_stats.clear() + if aggregator is None: + return + response_id = self._response.active_response_id + if response_id is not None and str(response_id) not in aggregator.e2e_done: + final_stage = aggregator.num_stages - 1 if aggregator.num_stages > 0 else 0 + aggregator.on_finalize_request(final_stage, response_id, aggregator.wall_start_ts) + for rid, events in aggregator.stage_events.items(): + aggregator.stage_events[rid] = _one_row_per_stage(events) + aggregator.build_and_log_summary() + + def _clear_response_request_timing(self) -> None: + self._response.active_response_request_started_at_s = None + self._response.active_response_ttft_ms = None + self._response.active_response_ttfp_ms = None + + def mark_model_turn_request_started(self, turn_id: int, started_at_s: float) -> None: + """Record the native request start that can own one model turn. + + Pending turns keep the latest accepted append. After ``begin_response`` + the first accepted append wins; later appends must not rebind. + """ + 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 newly 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) + newly_observed = False + if has_text and self._response.active_response_ttft_ms is None: + self._response.active_response_ttft_ms = elapsed_ms + newly_observed = True + if has_audio and self._response.active_response_ttfp_ms is None: + self._response.active_response_ttfp_ms = elapsed_ms + newly_observed = True + if not newly_observed: + return {} + metrics: dict[str, object] = { + "source": "server_monotonic_request_start", + "measurement_origin": dict(RESPONSE_REQUEST_MEASUREMENT_ORIGIN), + } + 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 stash_stage_metrics(self, stage_metrics: Mapping[Any, Any] | None) -> None: """Hold a stage snapshot until a response exists to attribute it to. @@ -945,7 +1084,7 @@ class DuplexEngineSession: 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 + weight = _tpot_interval_weight(token_count) 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 ) @@ -1160,6 +1299,7 @@ class DuplexEngineSession: playback_commit_policy: str | None = None, preserve_request: bool = False, ) -> dict[str, object] | None: + self._log_response_aggregator() response_id = self._response.active_response_id response_input_commit_seq = self._response.active_response_input_commit_seq if response_input_commit_seq is None: @@ -1205,6 +1345,7 @@ class DuplexEngineSession: self._response.active_request_id = None self._response.active_response_id = None self._response.active_response_turn_id = None + self._clear_response_request_timing() self._response.active_response_input_commit_seq = None self._response.active_response_awaits_input_commit = False # Keep draining bindings for other responses (older TTS may still play). @@ -1534,6 +1675,7 @@ class DuplexEngineSession: return self._playback.by_response.get(item_id.removeprefix("item_")) def barge_in(self) -> int: + self._log_response_aggregator() self.epoch += 1 self.sync_fence() self._response.assistant_text_buffer.clear() @@ -1541,6 +1683,8 @@ class DuplexEngineSession: 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._clear_response_request_timing() self._response.active_response_input_commit_seq = None self._response.active_response_awaits_input_commit = False self._response.draining_response_by_request.clear() @@ -1554,9 +1698,12 @@ class DuplexEngineSession: self.state = DuplexSessionState.CLOSING def close(self) -> None: + self._log_response_aggregator() 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._clear_response_request_timing() self._response.active_response_input_commit_seq = None self._response.active_response_awaits_input_commit = False self._clear_response_metrics() diff --git a/vllm_omni/engine/duplex/session/manager.py b/vllm_omni/engine/duplex/session/manager.py index 76e2c7ee8..15e038806 100644 --- a/vllm_omni/engine/duplex/session/manager.py +++ b/vllm_omni/engine/duplex/session/manager.py @@ -99,6 +99,7 @@ class DuplexSessionManager: result_sink: janus.AsyncQueue[EngineQueueMessage], runtime_config: DuplexSessionRuntimeConfig, model_config: ModelConfig | None, + log_stats: bool = False, clock: Callable[[], float] | None = None, executor: concurrent.futures.ThreadPoolExecutor | None = None, ) -> None: @@ -106,6 +107,7 @@ class DuplexSessionManager: self.plugin = plugin self.stage_port = stage_port self.model_config = model_config + self.log_stats = bool(log_stats) self.runtime_config = runtime_config self._output_sink = output_sink self._result_sink = result_sink @@ -499,6 +501,8 @@ class DuplexSessionManager: lease=DuplexLeaseState(config=self._lease_config, generation=0, last_activity=self._clock()), _clock=self._clock, _runtime_config=dict(runtime_config), + num_stages=self.stage_port.stage_count, + log_stats=self.log_stats, ) # Validates the plugin's sampling policy for this runtime config before admission. self.sampling_params_for(session) diff --git a/vllm_omni/engine/duplex/session/model_channel.py b/vllm_omni/engine/duplex/session/model_channel.py index 74c6cbf4f..ef3655995 100644 --- a/vllm_omni/engine/duplex/session/model_channel.py +++ b/vllm_omni/engine/duplex/session/model_channel.py @@ -46,6 +46,7 @@ from vllm_omni.engine.duplex.contracts import ( from vllm_omni.engine.duplex.plugin import ( DuplexRuntimeConfigError, coerce_int, + payload_turn_id, ) from vllm_omni.engine.duplex.session import helpers from vllm_omni.engine.duplex.session.context import DuplexSessionContext, StageOutput @@ -192,6 +193,7 @@ class ModelChannel: return True, False if expected_epoch is not None and session.epoch != expected_epoch: return True, False + self._mark_accepted_append_request_start(payload) # Commit timing state before returned output events can clear the # silence-continuation chain (e.g. a terminal turn-end). if on_append_accepted is not None: @@ -208,6 +210,16 @@ class ModelChannel: return False, emitted_response return True, emitted_response + def _mark_accepted_append_request_start(self, payload: object) -> None: + """Stamp TTFT/TTFP origin only after the data plane accepted this append.""" + session = self._ctx.session + turn_id = payload_turn_id(payload) + if turn_id is None: + turn_id = ( + session.active_response_turn_id if session.active_response_turn_id is not None else session.turn_id + ) + session.mark_model_turn_request_started(turn_id, session._clock()) + async def _append_via_data_plane( self, payload: object, @@ -471,8 +483,9 @@ class ModelChannel: context=context, ) - @staticmethod - def stage_metrics_snapshot(stage_id: int, metrics: object, output: object) -> dict[str, dict[str, object]] | None: + def stage_metrics_snapshot( + self, stage_id: int, metrics: object, output: object + ) -> dict[str, dict[str, object]] | None: if not isinstance(metrics, StageRequestStats): return None event = metrics @@ -482,6 +495,7 @@ class ModelChannel: final_output_type = getattr(output, "final_output_type", None) if isinstance(final_output_type, str): event = replace(event, final_output_type=final_output_type) + self._ctx.session.observe_stage_request_stats(stage_id, event) try: merged = OrchestratorAggregator._merge_stage_metric_event(None, event) except Exception: @@ -927,7 +941,16 @@ class ModelChannel: if response_id is None: response_id = session.begin_response(turn_id=model_turn_id) response_created = True - self._out.emit(self.response_created_payload(response_id, epoch=session.epoch)) + response_request_metrics = session.mark_response_first_outputs( + observed_at_s=session._clock(), + has_text=has_text, + has_audio=has_audio, + ) + if response_created: + created_payload = self.response_created_payload(response_id, epoch=session.epoch) + if response_request_metrics: + created_payload["response_request_metrics"] = response_request_metrics + self._out.emit(created_payload) stage_metrics = model_result.get("stage_metrics") response_stage_metrics = session.accumulate_response_stage_metrics( stage_metrics if isinstance(stage_metrics, Mapping) else None @@ -942,7 +965,12 @@ class ModelChannel: "end_of_turn": end_of_turn, "model_speak": True, } - self._attach_runtime_metadata(speak_payload, model_result, stage_metrics=response_stage_metrics) + self._attach_runtime_metadata( + speak_payload, + model_result, + stage_metrics=response_stage_metrics, + response_request_metrics=response_request_metrics, + ) self._out.emit(speak_payload) target_id = draining_response_id if draining_response_id not in (None, session.active_response_id) else None previous_sent_ms = session.playback_for_response(target_id).sent_ms @@ -1013,7 +1041,12 @@ class ModelChannel: sample_rate_hz = model_result.get("sample_rate_hz") or model_result.get("audio_sample_rate_hz") if isinstance(sample_rate_hz, int | float) and int(sample_rate_hz) > 0: payload["sample_rate_hz"] = int(sample_rate_hz) - self._attach_runtime_metadata(payload, model_result, stage_metrics=response_stage_metrics) + self._attach_runtime_metadata( + payload, + model_result, + stage_metrics=response_stage_metrics, + response_request_metrics=response_request_metrics, + ) self._out.emit(payload) if ( not end_of_turn @@ -1138,6 +1171,7 @@ class ModelChannel: model_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 = model_result.get("runtime_impl") @@ -1160,6 +1194,8 @@ class ModelChannel: 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/engine/duplex_orchestrator.py b/vllm_omni/engine/duplex_orchestrator.py index e0b18061e..39719453a 100644 --- a/vllm_omni/engine/duplex_orchestrator.py +++ b/vllm_omni/engine/duplex_orchestrator.py @@ -90,6 +90,7 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): result_sink=self.rpc_async_queue, runtime_config=self.duplex_session_config, model_config=model_config, + log_stats=self.log_stats, ) # ------------------------------------------------------------------ # @@ -133,6 +134,8 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): logger.debug("[DuplexOrchestrator] stale stage binding ignored for %s stage-%s", request_id, stage_id) return req_state.stage_submit_ts[stage_id] = _time.time() + if req_state.request_timestamp <= 0.0: + req_state.request_timestamp = req_state.stage_submit_ts[stage_id] self._register_running_request(req_state) self.session_manager.register_request(request_id, req_state.session_id) @@ -320,6 +323,7 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): prompt=None, sampling_params_list=list(context.sampling_params), final_stage_id=context.final_stage_id, + request_timestamp=_time.time(), session_owned=True, session_id=context.session_id, fence=context.fence, @@ -332,6 +336,8 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): request_state.config_generation = context.config_generation request_state.session_id = context.session_id request_state.fence = context.fence + if request_state.request_timestamp <= 0.0: + request_state.request_timestamp = _time.time() else: raise RuntimeError(f"request {context.request_id} is not session-owned") self._sync_bridge_state(request_state, context) @@ -408,7 +414,9 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): f"async-chunk prewarm failed for duplex request {context.request_id}; the request was aborted" ) request_state.stage_fences[context.stage_id] = context.fence - request_state.stage_submit_ts[context.stage_id] = _time.time() + now = _time.time() + request_state.stage_submit_ts[context.stage_id] = now + request_state.request_timestamp = now self._register_running_request(request_state) return DuplexStageSubmissionResult( request_id=context.request_id, diff --git a/vllm_omni/metrics/definitions.py b/vllm_omni/metrics/definitions.py index 59b575e50..1a7db5431 100644 --- a/vllm_omni/metrics/definitions.py +++ b/vllm_omni/metrics/definitions.py @@ -17,6 +17,7 @@ time-bearing metrics use the ``_s`` suffix (values in seconds), counters use import logging import os from collections.abc import Mapping, Sequence +from typing import NamedTuple from vllm_omni.metrics.utils import resolve_int_by_sequential_keys @@ -341,6 +342,43 @@ AUDIO_CHANNELS_ENV = "VLLM_OMNI_BENCH_AUDIO_CHANNELS" # the commonly-cited "audible gap" threshold for streaming TTS. AUDIO_CONTINUITY_DEFAULT_THRESHOLD_S = 0.1 +# Intermediate streaming units that are not the pipeline's text or audio +# final output (e.g. MiniCPM Talker codec tokens). +STREAMING_OUTPUT_UNIT_TYPES = frozenset({"text", "stream", "audio"}) + + +class StageModalityFlags(NamedTuple): + """How ``--print-stage`` classifies one engine stage.""" + + is_text_stage: bool + is_audio_stage: bool + is_image_stage: bool + is_video_stage: bool + is_internal_stream_stage: bool + + +def stage_modality_flags( + final_output_type: object, + output_unit_type: object, +) -> StageModalityFlags: + """Classify a stage from wire ``final_output_type`` / ``output_unit_type``.""" + final_type = final_output_type if isinstance(final_output_type, str) else "" + unit_type = output_unit_type if isinstance(output_unit_type, str) else "" + is_text_stage = final_type == "text" or unit_type == "text" + is_audio_stage = final_type == "audio" or unit_type == "audio" + is_video_stage = final_type in {"video", "videos"} or unit_type == "video" + # Video diffusion may still report output_unit_type="image" when frames are + # stored in ``images``; prefer video when final_output_type says so. + is_image_stage = (not is_video_stage) and (final_type in {"image", "images"} or unit_type == "image") + is_internal_stream_stage = unit_type in STREAMING_OUTPUT_UNIT_TYPES and not is_text_stage and not is_audio_stage + return StageModalityFlags( + is_text_stage=is_text_stage, + is_audio_stage=is_audio_stage, + is_image_stage=is_image_stage, + is_video_stage=is_video_stage, + is_internal_stream_stage=is_internal_stream_stage, + ) + # ============================================================================ # Formula helpers (shared by server-side observe and bench-side calculation) diff --git a/vllm_omni/metrics/stats.py b/vllm_omni/metrics/stats.py index e38d83133..df5183f45 100644 --- a/vllm_omni/metrics/stats.py +++ b/vllm_omni/metrics/stats.py @@ -3,9 +3,10 @@ from __future__ import annotations +import copy import time from collections import defaultdict -from collections.abc import Callable +from collections.abc import Callable, Sequence from contextlib import contextmanager from dataclasses import dataclass from typing import TYPE_CHECKING, Any @@ -13,7 +14,7 @@ from typing import TYPE_CHECKING, Any from vllm.logger import init_logger from vllm_omni.metrics import definitions as defs -from vllm_omni.metrics.utils import _build_field_defs, _build_row, _format_table +from vllm_omni.metrics.utils import _as_float, _as_float_list, _as_int, _build_field_defs, _build_row, _format_table if TYPE_CHECKING: from vllm_omni.metrics.transfer import OmniTransferMetrics @@ -130,6 +131,11 @@ STAGE_EXCLUDE = { "finish_reason", "pipeline_timings", } +# Duplex logs one StageRequestStats table per response. Chunk submits refresh +# ``request_timestamp``, so serving_time_to_first_output_ms is often a clamped +# 0 on the audio column and is not a useful TTFP. HTTP ``--print-stage`` keeps +# the row. +DUPLEX_STAGE_TABLE_EXCLUDE = frozenset({defs.SERVING_TIME_TO_FIRST_OUTPUT_MS}) TRANSFER_EXCLUDE = {"from_stage", "to_stage", "request_id", "used_shm"} E2E_EXCLUDE = {"request_id"} @@ -140,6 +146,86 @@ TRANSFER_FIELDS = _build_field_defs(TransferEdgeStats, TRANSFER_EXCLUDE, FIELD_T E2E_FIELDS = _build_field_defs(RequestE2EStats, E2E_EXCLUDE, FIELD_TRANSFORMS) +def _tpot_interval_weight(token_count: object) -> int: + if isinstance(token_count, int | float) and not isinstance(token_count, bool): + return max(int(token_count) - 1, 1) + return 1 + + +def _weighted_tpot_ms(events: Sequence[StageRequestStats]) -> float | None: + weighted_ms = 0.0 + weight = 0 + for event in events: + tpot_ms = float(event.vllm_tpot_ms) + if tpot_ms <= 0: + continue + chunk_weight = _tpot_interval_weight(event.num_tokens_out) + weighted_ms += tpot_ms * chunk_weight + weight += chunk_weight + if weight <= 0: + return None + return weighted_ms / float(weight) + + +def _apply_merged_stage_stats(template: StageRequestStats, merged: dict[str, object]) -> StageRequestStats: + """Write one ``_merge_stage_metric_event`` snapshot back onto a stats row.""" + stats = copy.copy(template) + stats.stage_id = _as_int(merged.get("stage_id"), default=template.stage_id or 0) + stats.final_output_type = ( + str(merged["final_output_type"]) + if isinstance(merged.get("final_output_type"), str) + else template.final_output_type + ) + stats.finish_reason = str(merged["finish_reason"]) if isinstance(merged.get("finish_reason"), str) else None + stats.num_tokens_in = _as_int(merged.get(defs.NUM_TOKENS_IN)) + stats.num_tokens_out = _as_int(merged.get(defs.NUM_TOKENS_OUT)) + stats.stage_gen_time_ms = _as_float(merged.get(defs.STAGE_GEN_TIME_MS)) + stats.postprocess_time_ms = _as_float(merged.get(defs.POSTPROCESS_TIME_MS)) + stats.audio_generated_frames = _as_int(merged.get(defs.AUDIO_FRAMES)) + stats.audio_sample_rate = _as_int(merged.get(defs.AUDIO_SAMPLE_RATE)) + stats.audio_duration_s = _as_float(merged.get(f"{defs.AUDIO_DURATION}_s")) + stats.image_pixels = _as_int(merged.get(defs.IMAGE_PIXELS)) + stats.denoise_step_latency_ms = _as_float(merged.get(defs.DENOISE_STEP_LATENCY_MS)) + stats.output_unit_type = ( + str(merged["output_unit_type"]) + if isinstance(merged.get("output_unit_type"), str) + else template.output_unit_type + ) + stats.output_unit_count = _as_int(merged.get(defs.OUTPUT_UNIT_COUNT)) + stats.serving_time_to_first_output_ms = _as_float(merged.get(defs.SERVING_TIME_TO_FIRST_OUTPUT_MS)) + stats.image_time_to_first_output_ms = _as_float(merged.get(defs.IMAGE_TIME_TO_FIRST_OUTPUT_MS)) + stats.time_per_output_unit_ms = _as_float(merged.get(defs.TIME_PER_OUTPUT_UNIT_MS)) + stats.inter_output_latencies_ms = _as_float_list(merged.get(defs.INTER_OUTPUT_LATENCIES_MS)) + stats.inter_output_latency_ms = _as_float(merged.get(defs.INTER_OUTPUT_LATENCY_MS)) + stats.vllm_ttft_ms = _as_float(merged.get(defs.VLLM_TTFT_MS)) + stats.vllm_tpot_ms = _as_float(merged.get(defs.VLLM_TPOT_MS)) + stats.vllm_itls_ms = _as_float_list(merged.get(defs.VLLM_ITLS_MS)) + stats.vllm_itl_ms = _as_float(merged.get(defs.VLLM_ITL_MS)) + return stats + + +def _one_row_per_stage(events: list[StageRequestStats]) -> list[StageRequestStats]: + """Fold chunk snapshots so the logger table has one column per stage.""" + merged_by_stage: dict[int, dict[str, object]] = {} + templates: dict[int, StageRequestStats] = {} + chunks_by_stage: dict[int, list[StageRequestStats]] = {} + for evt in events: + if evt.stage_id is None: + continue + sid = int(evt.stage_id) + templates.setdefault(sid, evt) + chunks_by_stage.setdefault(sid, []).append(evt) + merged_by_stage[sid] = OrchestratorAggregator._merge_stage_metric_event(merged_by_stage.get(sid), evt) + rows: list[StageRequestStats] = [] + for sid in sorted(merged_by_stage): + merged = merged_by_stage[sid] + tpot_ms = _weighted_tpot_ms(chunks_by_stage[sid]) + if tpot_ms is not None: + merged[defs.VLLM_TPOT_MS] = tpot_ms + rows.append(_apply_merged_stage_stats(templates[sid], merged)) + return rows + + class OrchestratorAggregator: def __init__( self, @@ -150,10 +236,12 @@ class OrchestratorAggregator: *, transfer_emitter: OmniTransferMetrics | None = None, replica_resolver: Callable[[int, str], int | None] | None = None, + stage_table_exclude: frozenset[str] = frozenset(), ) -> None: self.num_stages = int(num_stages) self.log_stats = bool(log_stats) self.final_stage_id_for_e2e = final_stage_id_for_e2e + self.stage_table_exclude = frozenset(stage_table_exclude) self.init_run_state(wall_start_ts) self.stage_events: dict[str, list[StageRequestStats]] = {} self.transfer_events: dict[ @@ -880,7 +968,8 @@ class OrchestratorAggregator: # === Stage table (columns = stage_id) === # if any stage has diffusion_metrics, remove postprocess_time_ms field # because it is already included in diffusion_metrics - local_exclude = STAGE_EXCLUDE.copy() + local_exclude: set[str] = set(STAGE_EXCLUDE) + local_exclude.update(self.stage_table_exclude) has_diffusion_metrics = any(getattr(evt, "diffusion_metrics", None) for evt in stage_evts) if has_diffusion_metrics: local_exclude.add("postprocess_time_ms") diff --git a/vllm_omni/metrics/utils.py b/vllm_omni/metrics/utils.py index ad58e4cec..6c9de80ad 100644 --- a/vllm_omni/metrics/utils.py +++ b/vllm_omni/metrics/utils.py @@ -20,6 +20,40 @@ FAILURE_REASONS = frozenset({"client_abort", "client_disconnect", "stage_error", DIFFUSION_METRICS_ONLY_REQUEST_ID = "__vllm_omni_diffusion_metrics__" +def _as_int(value: object, default: int = 0) -> int: + if isinstance(value, bool): + return default + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + return default + + +def _as_optional_int(value: object) -> int | None: + if value is None or isinstance(value, bool): + return None + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + return None + + +def _as_float(value: object, default: float = 0.0) -> float: + if isinstance(value, bool): + return default + if isinstance(value, int | float): + return float(value) + return default + + +def _as_float_list(value: object) -> list[float]: + if not isinstance(value, list): + return [] + return [item for item in value if isinstance(item, int | float) and not isinstance(item, bool)] + + def extract_queue_wait_s(pipeline_timings: Mapping[str, float] | None) -> float | None: if pipeline_timings is None or "queue_wait_ms" not in pipeline_timings: return None diff --git a/vllm_omni/protocol/duplex/events.py b/vllm_omni/protocol/duplex/events.py index f2abe748e..cd1705487 100644 --- a/vllm_omni/protocol/duplex/events.py +++ b/vllm_omni/protocol/duplex/events.py @@ -55,7 +55,6 @@ from vllm_omni.protocol.realtime.events import ( SpeechStopped, TextDelta, TextDone, - TranscriptDelta, TranscriptDone, new_event_id, wire_value, @@ -116,6 +115,14 @@ class AudioDelta(realtime_events.AudioDelta): metadata: Mapping[str, object] | None = None +@dataclass(frozen=True, slots=True, kw_only=True) +class TranscriptDelta(realtime_events.TranscriptDelta): + """Tier 2: OpenAI's ``TranscriptDelta`` plus duplex request-start metrics on text-only units.""" + + optional_wire_fields = frozenset({"metadata"}) + metadata: Mapping[str, object] | None = None + + @dataclass(frozen=True, slots=True, kw_only=True) class InputCommitted(realtime_events.InputCommitted): """Tier 2: OpenAI's ``InputCommitted`` plus the native ``input.committed`` event echoed under ``event``.""" @@ -358,8 +365,8 @@ def error_event( ) -#: The complete event vocabulary a duplex client may receive: 22 Tier 1 -#: classes re-exported unchanged, the 8 Tier 2 ones defined above, and 12 +#: The complete event vocabulary a duplex client may receive: 21 Tier 1 +#: classes re-exported unchanged, the 9 Tier 2 ones defined above, and 12 #: Tier 3 ones. A duplex consumer imports from here and never reaches past #: this module into ``vllm_omni.protocol.realtime``. __all__ = [ @@ -384,7 +391,6 @@ __all__ = [ "SpeechStopped", "TextDelta", "TextDone", - "TranscriptDelta", "TranscriptDone", # Tier 2 --- an OpenAI event plus duplex-only fields. "AudioDelta", @@ -395,6 +401,7 @@ __all__ = [ "ResponseCreated", "ResponseDone", "SessionCreated", + "TranscriptDelta", # Tier 3 --- vLLM-Omni only. "DuplexRawEvent", "Listen",