diff --git a/docker/Dockerfile.npu b/docker/Dockerfile.npu index eaeef693c..9bb72ff65 100644 --- a/docker/Dockerfile.npu +++ b/docker/Dockerfile.npu @@ -1,5 +1,5 @@ ARG VLLM_ASCEND_IMAGE=quay.io/atlas-ci/vllm-ascend -ARG VLLM_ASCEND_TAG=v0.27.0 +ARG VLLM_ASCEND_TAG=v0.28.0 FROM ${VLLM_ASCEND_IMAGE}:${VLLM_ASCEND_TAG} # WORKDIR /vllm-workspace/vllm diff --git a/docker/Dockerfile.npu.a3 b/docker/Dockerfile.npu.a3 index 2dc6334d9..1b234d3fe 100644 --- a/docker/Dockerfile.npu.a3 +++ b/docker/Dockerfile.npu.a3 @@ -1,5 +1,5 @@ ARG VLLM_ASCEND_IMAGE=quay.io/atlas-ci/vllm-ascend -ARG VLLM_ASCEND_TAG=v0.27.0-a3 +ARG VLLM_ASCEND_TAG=v0.28.0-a3 FROM ${VLLM_ASCEND_IMAGE}:${VLLM_ASCEND_TAG} # WORKDIR /vllm-workspace/vllm diff --git a/docker/Dockerfile.npu.ci b/docker/Dockerfile.npu.ci index f5af5fb31..42968e467 100644 --- a/docker/Dockerfile.npu.ci +++ b/docker/Dockerfile.npu.ci @@ -1,5 +1,5 @@ ARG VLLM_ASCEND_IMAGE=quay.io/atlas-ci/vllm-ascend -ARG VLLM_ASCEND_TAG=v0.27.0 +ARG VLLM_ASCEND_TAG=v0.28.0 FROM ${VLLM_ASCEND_IMAGE}:${VLLM_ASCEND_TAG} ARG APP_DIR=/vllm-workspace/vllm-omni diff --git a/docker/Dockerfile.npu.ci.a3 b/docker/Dockerfile.npu.ci.a3 index 0529cb2da..48568275f 100644 --- a/docker/Dockerfile.npu.ci.a3 +++ b/docker/Dockerfile.npu.ci.a3 @@ -1,5 +1,5 @@ ARG VLLM_ASCEND_IMAGE=quay.io/atlas-ci/vllm-ascend -ARG VLLM_ASCEND_TAG=v0.27.0-a3 +ARG VLLM_ASCEND_TAG=v0.28.0-a3 FROM ${VLLM_ASCEND_IMAGE}:${VLLM_ASCEND_TAG} ARG APP_DIR=/vllm-workspace/vllm-omni diff --git a/tests/benchmarks/metrics/test_metrics.py b/tests/benchmarks/metrics/test_metrics.py index de746ad10..315685258 100644 --- a/tests/benchmarks/metrics/test_metrics.py +++ b/tests/benchmarks/metrics/test_metrics.py @@ -279,17 +279,79 @@ def test_unmeasured_tpot_stays_missing_after_tokenizer_fallback(): assert metrics.request_goodput == 0.0 -def test_measured_zero_itl_is_not_treated_as_missing(): +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} metrics = _calculate_test_metrics([output], {"tpot": 1.0}) - assert (metrics.num_tpot_samples, metrics.num_itl_samples) == (1, 2) - assert metrics.mean_tpot_ms == 0.0 + assert (metrics.num_tpot_samples, metrics.num_itl_samples) == (0, 2) + assert math.isnan(metrics.mean_tpot_ms) assert metrics.mean_itl_ms == 0.0 - assert metrics.request_goodput == 1.0 + assert metrics.request_goodput == 0.0 + + +def test_stage_output_tokens_without_client_timing_omit_tpot(capsys): + output = _make_output(100, output_tokens=5) + output.itl = [] + output.text_latency = 0.0 + output.ttft = 0.1 + + metrics, _ = calculate_metrics( + input_requests=[], + outputs=[output], + dur_s=1.0, + tokenizer=None, + selected_percentiles=[50.0, 99.0], + goodput_config_dict={}, + task_type=TaskType.GENERATION, + selected_percentile_metrics=["tpot"], + max_concurrency=None, + request_rate=float("inf"), + benchmark_duration=1.0, + ) + + printed = capsys.readouterr().out + assert metrics.num_tpot_samples == 0 + assert math.isnan(metrics.mean_tpot_ms) + assert "Time per Output Token" not in printed + assert "Mean TPOT" not in printed + + +def test_consistent_client_latency_can_supply_tpot_fallback(): + output = _make_output(100, output_tokens=10) + output.itl = [] + output.ttft = 0.1 + output.text_latency = 1.0 + + metrics = _calculate_test_metrics([output]) + + assert metrics.num_tpot_samples == 1 + assert metrics.mean_tpot_ms == pytest.approx(100.0) + + +def test_single_token_responses_do_not_report_zero_tpot(capsys): + output = _make_output(100, output_tokens=1) + + metrics, _ = calculate_metrics( + input_requests=[], + outputs=[output], + dur_s=1.0, + tokenizer=None, + selected_percentiles=[50.0], + goodput_config_dict={}, + task_type=TaskType.GENERATION, + selected_percentile_metrics=["tpot"], + max_concurrency=None, + request_rate=float("inf"), + benchmark_duration=1.0, + ) + + printed = capsys.readouterr().out + assert metrics.num_tpot_samples == 0 + assert math.isnan(metrics.mean_tpot_ms) + assert "Time per Output Token" not in printed def test_duplex_goodput_does_not_pair_measurements_from_different_requests(): diff --git a/tests/benchmarks/patch/test_patch.py b/tests/benchmarks/patch/test_patch.py index e0a9ff49b..e40e8f954 100644 --- a/tests/benchmarks/patch/test_patch.py +++ b/tests/benchmarks/patch/test_patch.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """ Unit tests for patch.py @@ -17,6 +17,7 @@ from vllm.benchmarks.lib.endpoint_request_func import RequestFuncInput from vllm_omni.benchmarks.patch.patch import ( MixRequestFuncOutput, + _apply_stage0_token_timings, async_request_openai_chat_omni_completions, async_request_openai_realtime_duplex, ) @@ -102,7 +103,24 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch) received_at_s=now + 0.03, ) self.events.add( - {"type": "response.done", "response": {"id": response_id}}, + { + "type": "response.done", + "response": { + "id": response_id, + "metadata": { + "vllm_omni": { + "stage_metrics": { + "0": { + "num_tokens_out": 3, + "vllm_ttft_ms": 20.0, + "vllm_tpot_ms": 10.0, + "vllm_itls_ms": [10.0, 10.0], + } + } + } + }, + }, + }, received_at_s=now + 0.04, ) @@ -143,6 +161,7 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch) assert client.configure_kwargs["native_duplex"] is False assert client.configure_kwargs["extra_body"] == { "ref_audio": "data:audio/wav;base64,AAAA", + "return_stage_metrics": True, } assert client.sent == [ event @@ -167,6 +186,10 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch) assert output.audio_ttfp > output.ttft assert output.audio_rtf > 0 assert output.latency > 0 + assert output.output_tokens == 12 + assert output.itl == [0.01, 0.01] * 4 + assert output.text_latency == pytest.approx(output.ttft + 0.08) + assert output.tpot_measured is True assert output.tts_turn_pcm_bytes == [b"\x00\x00" * 2400] * 4 assert output.tts_output_pcm_bytes == b"\x00\x00" * 9600 session_id = output.duplex_request_metrics[0]["session_id"] @@ -198,6 +221,54 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch) } +def test_stage0_token_timings_use_weighted_tpot_when_itls_are_incomplete(): + output = MixRequestFuncOutput(ttft=0.1) + + measured = _apply_stage0_token_timings( + output, + [ + {"output_token_count": 3, "itls_ms": [], "tpot_ms": 10.0}, + {"output_token_count": 2, "itls_ms": [], "tpot_ms": 40.0}, + ], + expected_output_tokens=5, + ) + + assert measured is True + assert output.itl == [] + assert output.tpot_measured is True + assert (output.text_latency - output.ttft) / 4 == pytest.approx(0.02) + + +def test_stage0_zero_itls_fall_back_to_positive_tpot(): + output = MixRequestFuncOutput(ttft=0.1) + + measured = _apply_stage0_token_timings( + output, + [{"output_token_count": 3, "itls_ms": [0.0, 0.0], "tpot_ms": 25.0}], + expected_output_tokens=3, + ) + + assert measured is True + assert output.itl == [] + assert output.tpot_measured is True + assert (output.text_latency - output.ttft) / 2 == pytest.approx(0.025) + + +def test_stage0_token_count_without_timing_is_not_measured(): + output = MixRequestFuncOutput(ttft=0.1) + + measured = _apply_stage0_token_timings( + output, + [{"output_token_count": 3, "itls_ms": [], "tpot_ms": None}], + expected_output_tokens=3, + ) + + assert measured is False + assert output.itl == [] + assert output.text_latency == output.ttft + assert output.tpot_measured is False + + def create_sse_chunk(data_dict): """Helper to create SSE formatted chunk""" return f"data: {json.dumps(data_dict)}\n\n".encode() diff --git a/tests/benchmarks/test_omniinteract.py b/tests/benchmarks/test_omniinteract.py index bb4fce251..c38e001bf 100644 --- a/tests/benchmarks/test_omniinteract.py +++ b/tests/benchmarks/test_omniinteract.py @@ -585,7 +585,7 @@ async def test_adapter_requires_and_forwards_exact_prepared_payload(tmp_path: Pa @pytest.mark.asyncio @pytest.mark.parametrize( - ("request_metrics", "expected_itl", "expected_text_latency"), + ("request_metrics", "expected_itl", "expected_text_latency", "expected_tpot_measured"), [ ( [ @@ -594,6 +594,7 @@ async def test_adapter_requires_and_forwards_exact_prepared_payload(tmp_path: Pa ], [0.01, 0.02, 0.03], 0.16, + True, ), ( [ @@ -602,6 +603,7 @@ async def test_adapter_requires_and_forwards_exact_prepared_payload(tmp_path: Pa ], [], 0.18, + True, ), ( [ @@ -610,6 +612,7 @@ async def test_adapter_requires_and_forwards_exact_prepared_payload(tmp_path: Pa ], [], 0.1, + False, ), ], ) @@ -619,6 +622,7 @@ async def test_adapter_reports_exact_or_weighted_token_timing( request_metrics: list[dict[str, object]], expected_itl: list[float], expected_text_latency: float, + expected_tpot_measured: bool, ): case, options = _case(tmp_path), _session_options(tmp_path) request = _request(case, options) @@ -643,7 +647,7 @@ async def test_adapter_reports_exact_or_weighted_token_timing( assert output.success assert output.itl == expected_itl assert output.text_latency == pytest.approx(expected_text_latency) - assert output.tpot_measured is True + assert output.tpot_measured is expected_tpot_measured def test_batch_finalization_publishes_only_measured_results(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): diff --git a/tests/dfx/perf/scripts/run_benchmark.py b/tests/dfx/perf/scripts/run_benchmark.py index 1eee88ac7..64285dca3 100644 --- a/tests/dfx/perf/scripts/run_benchmark.py +++ b/tests/dfx/perf/scripts/run_benchmark.py @@ -1,4 +1,8 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + import json +import math import os import threading from collections.abc import Callable @@ -159,6 +163,18 @@ def benchmark_params(request): def assert_result(result, params, num_prompt) -> None: assert result["completed"] == num_prompt, "Request failures exist" + baseline = params.get("baseline") + hardware = result.get("Hardware") + hardware_baseline = baseline.get(hardware) if isinstance(baseline, dict) and isinstance(hardware, str) else None + if isinstance(hardware_baseline, dict) and "mean_tpot_ms" in hardware_baseline: + num_tpot_samples = result.get("num_tpot_samples") + mean_tpot_ms = result.get("mean_tpot_ms") + assert isinstance(num_tpot_samples, int) and not isinstance(num_tpot_samples, bool) and num_tpot_samples > 0, ( + "TPOT baseline is configured, but no measurable TPOT samples were produced" + ) + assert ( + isinstance(mean_tpot_ms, int | float) and not isinstance(mean_tpot_ms, bool) and math.isfinite(mean_tpot_ms) + ), "TPOT baseline is configured, but mean_tpot_ms is not finite" expected_audio_turns = params.get("expected_duplex_audio_turns_per_session") if expected_audio_turns is not None: session_metrics = result.get("duplex_session_metrics") diff --git a/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json b/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json index 073c0c2da..801743134 100644 --- a/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json +++ b/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json @@ -43,7 +43,7 @@ "extra_body": { "save_duplex_request_metrics": true }, - "percentile-metrics": "ttft,e2el,audio_ttfp,audio_rtf,audio_duration", + "percentile-metrics": "ttft,tpot,itl,e2el,audio_ttfp,audio_rtf,audio_duration", "baseline": { "H100": { "request_throughput": [ diff --git a/tests/dfx/perf/tests/test_runner_metadata.py b/tests/dfx/perf/tests/test_runner_metadata.py index ac52a15cc..2e98dcaf7 100644 --- a/tests/dfx/perf/tests/test_runner_metadata.py +++ b/tests/dfx/perf/tests/test_runner_metadata.py @@ -515,3 +515,41 @@ def test_omni_duplex_expected_audio_turns_rejects_incomplete_session(): {"expected_duplex_audio_turns_per_session": 4}, 1, ) + + +def test_omni_tpot_baseline_accepts_measured_finite_sample(): + from tests.dfx.perf.scripts.run_benchmark import assert_result + + assert_result( + { + "completed": 1, + "Hardware": "H100", + "num_tpot_samples": 1, + "mean_tpot_ms": 10.0, + }, + {"baseline": {"H100": {"mean_tpot_ms": 20.0}}}, + 1, + ) + + +@pytest.mark.parametrize( + ("num_tpot_samples", "mean_tpot_ms", "match"), + [ + (0, float("nan"), "no measurable TPOT samples"), + (1, float("nan"), "mean_tpot_ms is not finite"), + ], +) +def test_omni_tpot_baseline_rejects_missing_or_nonfinite_sample(num_tpot_samples, mean_tpot_ms, match): + from tests.dfx.perf.scripts.run_benchmark import assert_result + + with pytest.raises(AssertionError, match=match): + assert_result( + { + "completed": 1, + "Hardware": "H100", + "num_tpot_samples": num_tpot_samples, + "mean_tpot_ms": mean_tpot_ms, + }, + {"baseline": {"H100": {"mean_tpot_ms": 20.0}}}, + 1, + ) diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index 467f928ce..475d58bd5 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -4021,6 +4021,55 @@ async def test_duplex_chat_audio_stream_uses_output_audio_delta_event(): ] +@pytest.mark.asyncio +async def test_duplex_chat_stage_metrics_use_latest_streaming_snapshot(): + class SnapshotChatService(FakeChatService): + async def create_chat_completion(self, request, raw_request=None): + del raw_request + self.seen_request_ids.append(request.request_id) + snapshots = [ + {"num_tokens_out": 1, "vllm_tpot_ms": 0.0, "vllm_itls_ms": []}, + {"num_tokens_out": 2, "vllm_tpot_ms": 10.0, "vllm_itls_ms": [10.0]}, + {"num_tokens_out": 3, "vllm_tpot_ms": 11.0, "vllm_itls_ms": [10.0, 12.0]}, + ] + + async def generate(): + for index, snapshot in enumerate(snapshots): + payload = { + "modality": "audio", + "choices": [{"delta": {"content": f"audio-{index}"}}], + "metrics": {"stage_metrics": {"0": snapshot}}, + } + yield f"data: {json.dumps(payload)}\n\n" + + return generate() + + handler = OmniDuplexSessionHandler( + chat_service=SnapshotChatService(FakeEngineClient()), + config_timeout_s=0.1, + idle_timeout_s=1, + ) + session = DuplexSession( + session_id="sid-chat-snapshots", + config=DuplexSessionConfig(instructions="speak"), + ) + sent: list[dict[str, Any]] = [] + + async def send_json(data: dict[str, Any]) -> None: + sent.append(data) + + await handler._run_response(session, send_json) + + audio_deltas = [event for event in sent if event.get("type") == "response.output_audio.delta"] + done = next(event for event in sent if event.get("type") == "response.done") + assert all("vllm_omni" not in event for event in audio_deltas) + assert done["vllm_omni"]["stage_metrics"]["0"] == { + "num_tokens_out": 3, + "vllm_tpot_ms": 11.0, + "vllm_itls_ms": [10.0, 12.0], + } + + @pytest.mark.asyncio async def test_duplex_chat_rejects_unknown_output_modality(): handler = OmniDuplexSessionHandler( diff --git a/vllm_omni/benchmarks/metrics/metrics.py b/vllm_omni/benchmarks/metrics/metrics.py index 32716c7af..3d7c983f3 100644 --- a/vllm_omni/benchmarks/metrics/metrics.py +++ b/vllm_omni/benchmarks/metrics/metrics.py @@ -773,25 +773,33 @@ def calculate_metrics( output_len = len(tokenizer(outputs[i].generated_text, add_special_tokens=False).input_ids) actual_output_lens.append(output_len) total_input += outputs[i].prompt_len - tpot: float | None = 0.0 + tpot: float | None = None if output_len > 1: if outputs[i].itl: # Use mean(ITL) directly so per-request TPOT == mean(ITL). # The ITL list records one entry per SSE chunk; server may # bundle multiple tokens per chunk, so len(itl)+1 != output_len. # Using mean(itl) keeps TPOT and ITL on the same footing. - tpot = sum(outputs[i].itl) / len(outputs[i].itl) + measured_tpot = sum(outputs[i].itl) / len(outputs[i].itl) + if np.isfinite(measured_tpot) and measured_tpot > 0: + tpot = measured_tpot elif getattr(outputs[i], "tpot_measured", True): - try: - latency_minus_ttft = outputs[i].text_latency - outputs[i].ttft - except Exception: - latency_minus_ttft = outputs[i].latency - outputs[i].ttft - tpot = latency_minus_ttft / (output_len - 1) - else: - tpot = None + text_latency = getattr(outputs[i], "text_latency", None) + ttft = getattr(outputs[i], "ttft", None) + if ( + isinstance(text_latency, (int, float)) + and isinstance(ttft, (int, float)) + and np.isfinite(text_latency) + and np.isfinite(ttft) + and text_latency > ttft > 0 + ): + tpot = (text_latency - ttft) / (output_len - 1) if tpot is not None: tpots.append(tpot) - # Note: if output_len <= 1, we regard tpot as 0 for goodput + # TPOT is not measurable for a single-token response or when the + # client did not observe a positive decode interval. In + # particular, stage-level output token counts alone must not turn + # an unset text_latency into a negative TPOT sample. all_tpots.append(tpot) itls += outputs[i].itl session_metrics = getattr(outputs[i], "duplex_session_metrics", None) @@ -949,10 +957,10 @@ def calculate_metrics( std_ttft_ms=np.std(ttfts or missing_duplex_value) * 1000, median_ttft_ms=np.median(ttfts or missing_duplex_value) * 1000, percentiles_ttft_ms=[(p, np.percentile(ttfts or missing_duplex_value, p) * 1000) for p in selected_percentiles], - mean_tpot_ms=np.mean(tpots or missing_duplex_value) * 1000, - std_tpot_ms=np.std(tpots or missing_duplex_value) * 1000, - median_tpot_ms=np.median(tpots or missing_duplex_value) * 1000, - percentiles_tpot_ms=[(p, np.percentile(tpots or missing_duplex_value, p) * 1000) for p in selected_percentiles], + mean_tpot_ms=np.mean(tpots or float("nan")) * 1000, + std_tpot_ms=np.std(tpots or float("nan")) * 1000, + median_tpot_ms=np.median(tpots or float("nan")) * 1000, + percentiles_tpot_ms=[(p, np.percentile(tpots or float("nan"), p) * 1000) for p in selected_percentiles], mean_itl_ms=np.mean(itls or missing_duplex_value) * 1000, std_itl_ms=np.std(itls or missing_duplex_value) * 1000, median_itl_ms=np.median(itls or missing_duplex_value) * 1000, diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index 942a9963b..217467a2c 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -1492,6 +1492,72 @@ def _nonnegative_number(value: object) -> bool: return isinstance(value, int | float) and not isinstance(value, bool) and np.isfinite(value) and value >= 0 +def _apply_stage0_token_timings( + output: MixRequestFuncOutput, + raw_stages: list[object], + *, + expected_output_tokens: int, +) -> bool: + """Populate aggregate token timing from complete per-response Stage 0 metrics.""" + output.itl = [] + output.text_latency = output.ttft + output.tpot_measured = False + + stages: list[tuple[int, dict[str, object]]] = [] + for stage in raw_stages: + if not isinstance(stage, dict): + return False + token_count = stage.get("output_token_count") + if not isinstance(token_count, int) or isinstance(token_count, bool) or token_count < 0: + return False + stages.append((token_count, stage)) + + if not stages or sum(count for count, _ in stages) != expected_output_tokens: + return False + + timed = [(count - 1, stage) for count, stage in stages if count > 1] + if not timed: + return False + + measured_itls: list[float] = [] + exact_itls = True + for intervals, stage in timed: + values = stage.get("itls_ms") + if ( + not isinstance(values, list) + or len(values) != intervals + or not all(_nonnegative_number(value) for value in values) + ): + exact_itls = False + break + measured_itls.extend(float(value) / 1000.0 for value in values if isinstance(value, int | float)) + if exact_itls: + if sum(measured_itls) > 0: + output.itl = measured_itls + output.text_latency = output.ttft + sum(output.itl) + output.tpot_measured = True + return True + + weighted_tpot_ms = 0.0 + for intervals, stage in timed: + tpot_ms = stage.get("tpot_ms") + if not _nonnegative_number(tpot_ms) or not isinstance(tpot_ms, int | float) or tpot_ms <= 0: + break + weighted_tpot_ms += float(tpot_ms) * intervals + else: + decode_intervals = sum(intervals for intervals, _ in timed) + weighted_tpot = weighted_tpot_ms / decode_intervals / 1000.0 + # ``calculate_metrics`` consumes a request-shaped latency. Preserve + # the weighted per-response TPOT when projecting it onto the combined + # output token count; gaps and TTFTs between turns are intentionally + # excluded. + output.text_latency = output.ttft + weighted_tpot * (expected_output_tokens - 1) + output.tpot_measured = True + return True + + return False + + async def _async_request_omniinteract( request_func_input: RequestFuncInput, *, @@ -1538,35 +1604,12 @@ async def _async_request_omniinteract( 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) - stages: list[tuple[int, dict[str, object]]] = [] - for request_metric in case_result.duplex_request_metrics: - stage0 = request_metric.get("stage0_tokens") - if not isinstance(stage0, dict): - continue - token_count = stage0.get("output_token_count") - if isinstance(token_count, int) and not isinstance(token_count, bool) and token_count >= 0: - stages.append((token_count, stage0)) - timed = [(count - 1, stage) for count, stage in stages if count > 1] - complete = ( - len(stages) == len(case_result.duplex_request_metrics) - and sum(count for count, _ in stages) == output.output_tokens + token_timing_measured = _apply_stage0_token_timings( + output, + [request_metric.get("stage0_tokens") for request_metric in case_result.duplex_request_metrics], + expected_output_tokens=output.output_tokens, ) - exact = complete and all( - isinstance(values := stage.get("itls_ms"), list) - and len(values) == intervals - and all(_nonnegative_number(value) for value in values) - for intervals, stage in timed - ) - output.itl = [float(value) / 1000.0 for _, stage in timed for value in stage["itls_ms"]] if exact else [] - if output.itl: - output.text_latency = output.ttft + sum(output.itl) - elif complete and timed and all(_nonnegative_number(stage.get("tpot_ms")) for _, stage in timed): - intervals = sum(count for count, _ in timed) - weighted_tpot = sum(float(stage["tpot_ms"]) * count for count, stage in timed) / intervals / 1000.0 - output.text_latency = output.ttft + weighted_tpot * (output.output_tokens - 1) - else: - output.text_latency = output.ttft - output.tpot_measured = False + if not token_timing_measured: if output.output_tokens > 1 or (output.output_tokens == 0 and output.generated_text): logger.warning( "OmniInteract session %s omitted complete engine token timing; standard TPOT/ITL are unavailable", @@ -1610,6 +1653,9 @@ async def async_request_openai_realtime_duplex( speech_extra = getattr(request_func_input, "seed_tts_speech_extra", None) if not isinstance(speech_extra, dict): speech_extra = {} + else: + speech_extra = dict(speech_extra) + speech_extra["return_stage_metrics"] = True configured_turns = getattr(request_func_input, "seed_tts_turns", ()) turn_prompts = [ ( @@ -1740,6 +1786,16 @@ async def async_request_openai_realtime_duplex( for timing in turn_timings if isinstance((stage0 := timing.get("stage0_tokens")), dict) ) + token_timing_measured = _apply_stage0_token_timings( + output, + [timing.get("stage0_tokens") for timing in turn_timings], + expected_output_tokens=output.output_tokens, + ) + if not token_timing_measured and output.output_tokens > 1: + logger.warning( + "Realtime TTS session %s omitted complete engine token timing; standard TPOT/ITL are unavailable", + session_id, + ) output.success = True except Exception: output.success = False diff --git a/vllm_omni/experimental/fullduplex/openai/chat_fallback.py b/vllm_omni/experimental/fullduplex/openai/chat_fallback.py index 76c45fc73..4edba1646 100644 --- a/vllm_omni/experimental/fullduplex/openai/chat_fallback.py +++ b/vllm_omni/experimental/fullduplex/openai/chat_fallback.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + from __future__ import annotations import asyncio @@ -43,20 +46,22 @@ class ChatFallbackProjectorMixin: else: await self._emit_full_response(session, result, epoch, response_id, send_json) if session.epoch == epoch: + final_stage_metrics = session.accumulate_response_stage_metrics(None) should_commit = self._should_commit_response_to_history(session, response_id) committed_message = session.end_response(commit_text=should_commit) if should_commit: session.register_history_item(f"item_{response_id}", committed_message) - await send_json( - { - "type": "response.done", - "session_id": session.session_id, - "response_id": response_id, - "epoch": epoch, - "committed": committed_message is not None, - "playback": session.playback.as_dict(), - } - ) + done_payload: dict[str, object] = { + "type": "response.done", + "session_id": session.session_id, + "response_id": response_id, + "epoch": epoch, + "committed": committed_message is not None, + "playback": session.playback.as_dict(), + } + if final_stage_metrics: + done_payload["vllm_omni"] = {"stage_metrics": final_stage_metrics} + await send_json(done_payload) except asyncio.CancelledError: raise except Exception as exc: @@ -177,6 +182,11 @@ class ChatFallbackProjectorMixin: response_id: str, send_json, ) -> None: + metrics = payload.get("metrics") + stage_metrics = metrics.get("stage_metrics") if isinstance(metrics, dict) else None + if isinstance(stage_metrics, dict): + session.replace_response_stage_metric_snapshots(stage_metrics) + modality = payload.get("modality") if modality not in {None, "text", "audio"}: await send_json( diff --git a/vllm_omni/experimental/fullduplex/openai/protocol.py b/vllm_omni/experimental/fullduplex/openai/protocol.py index d5a9cdcf8..bc259d59d 100644 --- a/vllm_omni/experimental/fullduplex/openai/protocol.py +++ b/vllm_omni/experimental/fullduplex/openai/protocol.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + from __future__ import annotations import copy @@ -818,6 +821,23 @@ class DuplexSession: return copy.deepcopy(self._response.stage_metrics) + def replace_response_stage_metric_snapshots( + self, + stage_metrics: Mapping[object, object] | None, + ) -> dict[str, dict[str, object]]: + """Merge cumulative chat snapshots by replacing each stage's latest value.""" + if self.active_response_id is None or not isinstance(stage_metrics, Mapping): + return copy.deepcopy(self._response.stage_metrics) + + for raw_stage_id, raw_values in stage_metrics.items(): + if not isinstance(raw_values, Mapping): + continue + stage_id = str(raw_stage_id) + self._response.stage_metrics[stage_id] = copy.deepcopy(dict(raw_values)) + self._response.stage_metric_tpot_weighted_ms.pop(stage_id, None) + self._response.stage_metric_tpot_weight.pop(stage_id, None) + return copy.deepcopy(self._response.stage_metrics) + def accumulate_overlap_speech(self, duration_ms: int) -> int: self._input.overlap_speech_ms += max(0, int(duration_ms)) return self._input.overlap_speech_ms