diff --git a/docs/cli/README.md b/docs/cli/README.md index 40858ca1a18..348c8f3ddde 100644 --- a/docs/cli/README.md +++ b/docs/cli/README.md @@ -39,4 +39,5 @@ vllm bench serve --omni \ --num-prompts 5 ``` -See [vllm bench serve](./bench/serve.md) for the full reference of all available arguments. +See [vllm bench serve](./bench/serve.md) for serving benchmark arguments and +dataset-specific examples, including OmniInteract native-duplex sessions. diff --git a/docs/cli/bench/serve.md b/docs/cli/bench/serve.md index 2a1acd1f055..3e16ae5542f 100644 --- a/docs/cli/bench/serve.md +++ b/docs/cli/bench/serve.md @@ -2,7 +2,7 @@ The vllm bench command launches the vLLM-Omni benchmark to evaluate the performance of multimodal models. ## Notes -vLLM-Omni registers the `openai-chat-omni`, `openai-audio-speech`, `openai-image-edits-omni`, and `daily-omni` serving benchmark backends. +vLLM-Omni registers the `openai-chat-omni`, `openai-audio-speech`, `openai-image-edits-omni`, `daily-omni`, and `openai-realtime-duplex` serving benchmark backends. It also adds the `omniinteract` dataset. ## Basic Parameter Description You can use `vllm bench serve --omni --help=all` to get descriptions of all parameters. The commonly used parameters are described below: @@ -10,7 +10,7 @@ You can use `vllm bench serve --omni --help=all` to get descriptions of all para Enable Omni (multimodal) mode, supporting multimodal inputs and outputs such as images, videos, and audio. - `--backend` - Specify the backend adapter. vLLM-Omni adds `openai-chat-omni`, `openai-audio-speech`, `openai-image-edits-omni`, and `daily-omni` to the upstream vLLM backend choices. + Specify the backend adapter. vLLM-Omni adds `openai-chat-omni`, `openai-audio-speech`, `openai-image-edits-omni`, `daily-omni`, and `openai-realtime-duplex` to the upstream vLLM backend choices. - `--model` The model identifier to load, filled according to the models supported by vLLM-Omni. @@ -19,7 +19,7 @@ You can use `vllm bench serve --omni --help=all` to get descriptions of all para The API endpoint exposed externally, to which clients send their requests. - `--dataset-name` - The name of the dataset used; random-mm indicates generating random multimodal inputs (images, videos, audio). + The name of the dataset used; random-mm indicates generating random multimodal inputs (images, videos, audio), while `omniinteract` replays official OmniInteract videos. - `--num-prompts` The total number of requests to send, an integer. @@ -269,6 +269,44 @@ We use audio generation time / audio duration to calculate RTF. +### OmniInteract Realtime Benchmark + +OmniInteract runs each video as one native-duplex WebSocket sample in the standard serving-benchmark lifecycle: + +```bash +vllm bench serve --omni \ + --backend openai-realtime-duplex \ + --dataset-name omniinteract \ + --dataset-path /path/to/OmniInteract \ + --model openbmb/MiniCPM-o-4_5 \ + --base-url http://127.0.0.1:8000 \ + --endpoint /v1/realtime \ + --omniinteract-ref-audio /path/to/reference.wav \ + --omniinteract-output-dir ./omniinteract-artifacts \ + --num-prompts 3 \ + --num-warmups 0 +``` + +`--dataset-path` accepts an extracted directory, `data.tar[.gz]`, or a Hugging Face dataset ID; omitting it uses +`lucky-lance/OmniInteract`. `--num-prompts` is the total across subsets and defaults to 3 for OmniInteract; explicit `0` +selects all and oversize values use all available cases. Reference audio is required, and OmniInteract uses the +`/v1/realtime` endpoint. + +Audio is replayed as 16 kHz PCM16 in 200 ms chunks and video at 1 FPS with real-time pacing. All selected media is decoded +before timing and remains in client memory for the run, so `--max-concurrency` does not limit media preparation memory; use +explicit `--num-prompts 0` only when the client has enough RAM for the full dataset. Media commands are bounded by +`--omniinteract-media-timeout-s`, and concurrency defaults to 1. Standard request-rate, warmup, result-saving, and summary +options apply. Use `--omniinteract-require-response` only for functional E2E cases; LISTEN is a valid benchmark result. + +Each completed case writes `output.wav`, `wav_transcript.json`, `events.json`, `result.json`, and a final `.done` marker under +`--omniinteract-output-dir`. The root also contains `batch_summary.json` and `official_eval_manifest.jsonl`; failed cases write +`.failed.json`. Runs sharing one output root are serialized. Completion validates transport, response lifecycle, and artifacts, +not answer accuracy. Transcript timestamps are serialized playback-queue times. Clipped or cancelled outputs are ineligible and +omitted from the official manifest; `audio_clipped_bytes` records output beyond the rounded video horizon. + +TTFT, TTFP, and RTF start at client receipt of `response.created`. TPOT/ITL use engine stage-0 timing; ITL is emitted only when +every token interval is present. + ### Multi-Modal Benchmark
diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py index b6c86174755..23d65eb35f8 100644 --- a/examples/online_serving/minicpmo/realtime_duplex_demo.py +++ b/examples/online_serving/minicpmo/realtime_duplex_demo.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + """Minimal single-input MiniCPM-o 4.5 Realtime duplex demo. Run this after starting the duplex server. Strict lifecycle, overlap, and @@ -19,8 +22,8 @@ sys.path.insert(0, str(REPO_ROOT)) from vllm_omni.experimental.fullduplex.client import ( # noqa: E402 - PCM16_BYTES_PER_SAMPLE, - PCM16_SAMPLE_RATE, + PCM16_BYTES_PER_SAMPLE, # noqa: F401 - compatibility export + PCM16_SAMPLE_RATE, # noqa: F401 - compatibility export RealtimeDuplexClient, RealtimeEventCollector, build_realtime_url, @@ -28,6 +31,13 @@ wait_for, write_pcm16_wav, ) +from vllm_omni.experimental.fullduplex.client import chunk_period_ms as _chunk_period_ms # noqa: E402 +from vllm_omni.experimental.fullduplex.client import ( # noqa: E402 + has_residual_model_unit as _has_residual_model_unit, +) +from vllm_omni.experimental.fullduplex.client import ( # noqa: E402 + reference_audio_data_url as _ref_audio_data_url, +) class _StreamingOutputWriter: @@ -133,25 +143,6 @@ def _latest_model_decision( return decision -def _chunk_period_ms(events: list[dict[str, object]]) -> int: - for event in reversed(events): - session = event.get("session") - if not isinstance(session, dict): - continue - capabilities = session.get("capabilities") - if not isinstance(capabilities, dict): - continue - chunk_period_ms = capabilities.get("chunk_period_ms") - if isinstance(chunk_period_ms, int) and chunk_period_ms > 0: - return chunk_period_ms - return 1000 - - -def _has_residual_model_unit(pcm16: bytes, *, chunk_period_ms: int) -> bool: - unit_bytes = PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * chunk_period_ms // 1000 - return bool(unit_bytes > 0 and len(pcm16) % unit_bytes) - - def _response_in_progress(events: list[dict[str, object]]) -> bool: return sum(event.get("type") == "response.created" for event in events) > sum( event.get("type") == "response.done" for event in events @@ -168,13 +159,6 @@ def _event_count_after( return sum(event.get("type") == event_type for event in events[index + 1 :]) -def _ref_audio_data_url(path: str | None) -> str | None: - if path is None: - return None - ref_path = Path(path).expanduser() - return "data:audio/wav;base64," + base64.b64encode(ref_path.read_bytes()).decode("ascii") - - async def run_demo(args: argparse.Namespace) -> dict[str, object]: input_pcm16 = read_pcm16_wav(Path(args.input_wav)) if not input_pcm16: diff --git a/tests/benchmarks/metrics/test_metrics.py b/tests/benchmarks/metrics/test_metrics.py index 32d9d7958f5..de746ad10ae 100644 --- a/tests/benchmarks/metrics/test_metrics.py +++ b/tests/benchmarks/metrics/test_metrics.py @@ -1,10 +1,12 @@ # 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 metrics.py """ +import math + import pytest from vllm.benchmarks.serve import TaskType @@ -66,6 +68,22 @@ def _make_output(prompt_len: int, output_tokens: int = 10) -> MixRequestFuncOutp return output +def _calculate_test_metrics(outputs, goodput=None): + return calculate_metrics( + input_requests=[], + outputs=outputs, + dur_s=1.0, + tokenizer=None, + selected_percentiles=[50.0], + goodput_config_dict=goodput or {}, + task_type=TaskType.GENERATION, + selected_percentile_metrics=[], + max_concurrency=None, + request_rate=float("inf"), + benchmark_duration=1.0, + )[0] + + # ============================================================================ # total_input Tests # ============================================================================ @@ -126,6 +144,166 @@ def test_audio_continuity_aggregation(): assert p99 is not None and p99 > 0.4 +def test_unmeasured_duplex_latency_does_not_add_zero_samples(): + measured = _make_output(100, output_tokens=2) + measured.ttft = 0.1 + measured.audio_ttfp = 0.2 + measured.audio_rtf = 0.5 + measured.duplex_session_metrics = { + "mean_ttft_ms": 100.0, + "mean_ttfp_ms": 200.0, + "mean_rtf": 0.5, + } + listen_only = _make_output(100, output_tokens=0) + listen_only.ttft = listen_only.audio_ttfp = listen_only.audio_rtf = 0.0 + listen_only.duplex_session_metrics = { + "mean_ttft_ms": None, + "mean_ttfp_ms": None, + "mean_rtf": None, + } + + metrics = _calculate_test_metrics([measured, listen_only], {"ttft": 150.0}) + + assert metrics.mean_ttft_ms == 100.0 + assert metrics.mean_audio_ttfp_ms == 200.0 + assert metrics.mean_audio_rtf == 0.5 + assert (metrics.num_ttft_samples, metrics.num_audio_ttfp_samples, metrics.num_audio_rtf_samples) == (1, 1, 1) + assert metrics.request_goodput == 1.0 + + +def test_all_unmeasured_duplex_latency_is_not_reported_as_zero(): + listen_only = _make_output(100, output_tokens=0) + listen_only.duplex_session_metrics = { + "mean_ttft_ms": None, + "mean_ttfp_ms": None, + "mean_rtf": None, + } + + metrics = _calculate_test_metrics( + [listen_only], + {"ttft": float("inf"), "audio_ttft": float("inf")}, + ) + + assert (metrics.num_ttft_samples, metrics.num_audio_ttfp_samples, metrics.num_audio_rtf_samples) == (0, 0, 0) + assert math.isnan(metrics.mean_ttft_ms) + assert math.isnan(metrics.mean_audio_ttfp_ms) + assert math.isnan(metrics.mean_audio_rtf) + assert metrics.request_goodput == 0.0 + + +def test_unmeasured_duplex_tpot_does_not_add_zero_or_misalign_goodput(): + missing_tpot = _make_output(100, output_tokens=5) + missing_tpot.itl = [] + missing_tpot.text_latency = missing_tpot.ttft = 0.1 + missing_tpot.tpot_measured = False + missing_tpot.duplex_session_metrics = {"mean_ttft_ms": 100.0} + + slow_ttft = _make_output(100, output_tokens=5) + slow_ttft.ttft = 1.0 + slow_ttft.text_latency = 1.4 + slow_ttft.itl = [0.1] * 4 + slow_ttft.duplex_session_metrics = {"mean_ttft_ms": 1000.0} + + metrics = _calculate_test_metrics( + [missing_tpot, slow_ttft], + {"ttft": 500.0, "tpot": 200.0}, + ) + + assert metrics.num_tpot_samples == 1 + assert metrics.mean_tpot_ms == pytest.approx(100.0) + assert metrics.request_goodput == 0.0 + + +def test_all_unmeasured_duplex_token_timing_is_not_reported_as_zero(): + output = _make_output(100, output_tokens=5) + output.itl = [] + output.text_latency = output.ttft + output.tpot_measured = False + output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + + metrics = _calculate_test_metrics([output], {"tpot": float("inf")}) + + assert (metrics.num_tpot_samples, metrics.num_itl_samples) == (0, 0) + assert math.isnan(metrics.mean_tpot_ms) + assert math.isnan(metrics.mean_itl_ms) + assert metrics.request_goodput == 0.0 + + +def test_duplex_response_timings_do_not_build_a_session_token_timeline(): + output = _make_output(100, output_tokens=5) + output.latency = 101.0 + output.duplex_request_metrics = [ + {"response_id": "r1", "stage0_tokens": {"itls_ms": [100.0, 100.0]}}, + {"response_id": "r2", "stage0_tokens": {"itls_ms": [100.0, 100.0]}}, + ] + output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + + metrics = _calculate_test_metrics([output]) + + assert math.isnan(metrics.max_output_tokens_per_s) + assert metrics.max_concurrent_requests == 1 + assert metrics.mean_tpot_ms == pytest.approx(100.0) + assert metrics.mean_itl_ms == pytest.approx(100.0) + + +def test_unmeasured_tpot_stays_missing_after_tokenizer_fallback(): + output = _make_output(100, output_tokens=0) + output.generated_text = "timing metadata missing" + output.itl = [] + output.text_latency = output.ttft + output.tpot_measured = False + output.duplex_session_metrics = {"mean_ttft_ms": 100.0} + + def tokenizer(text, *, add_special_tokens): + assert text == output.generated_text + assert add_special_tokens is False + return type("Tokenized", (), {"input_ids": [1, 2, 3]})() + + metrics, _ = calculate_metrics( + input_requests=[], + outputs=[output], + dur_s=1.0, + tokenizer=tokenizer, + selected_percentiles=[50.0], + goodput_config_dict={"tpot": float("inf")}, + task_type=TaskType.GENERATION, + selected_percentile_metrics=[], + max_concurrency=None, + request_rate=float("inf"), + benchmark_duration=1.0, + ) + + assert metrics.total_output == 3 + assert metrics.num_tpot_samples == 0 + assert math.isnan(metrics.mean_tpot_ms) + assert metrics.request_goodput == 0.0 + + +def test_measured_zero_itl_is_not_treated_as_missing(): + 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.mean_itl_ms == 0.0 + assert metrics.request_goodput == 1.0 + + +def test_duplex_goodput_does_not_pair_measurements_from_different_requests(): + text_only, audio_only = _make_output(100), _make_output(100) + text_only.ttft, text_only.audio_ttfp = 0.1, 0.0 + text_only.duplex_session_metrics = {"mean_ttft_ms": 100.0, "mean_ttfp_ms": None, "mean_rtf": None} + audio_only.ttft, audio_only.audio_ttfp = 0.0, 0.2 + audio_only.duplex_session_metrics = {"mean_ttft_ms": None, "mean_ttfp_ms": 200.0, "mean_rtf": None} + + metrics = _calculate_test_metrics([text_only, audio_only], {"ttft": 500.0, "audio_ttft": 500.0}) + + assert metrics.request_goodput == 0.0 + + # ============================================================================ # TTFT suppression for pure-audio (TTS) benchmarks # ============================================================================ diff --git a/tests/benchmarks/test_omniinteract.py b/tests/benchmarks/test_omniinteract.py new file mode 100644 index 00000000000..bb4fce251ca --- /dev/null +++ b/tests/benchmarks/test_omniinteract.py @@ -0,0 +1,778 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import io +import json +import subprocess +import tarfile +import threading +import time +import wave +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest +from vllm.benchmarks.lib.endpoint_request_func import RequestFuncInput +from vllm.benchmarks.serve import TaskType + +from vllm_omni.benchmarks import omniinteract as oi +from vllm_omni.benchmarks import serve as benchmark_serve +from vllm_omni.benchmarks.data_modules import omniinteract_dataset as data +from vllm_omni.benchmarks.metrics.metrics import calculate_metrics +from vllm_omni.benchmarks.patch import patch as benchmark_patch +from vllm_omni.entrypoints.cli.benchmark.cli_args import preprocess_serve_args +from vllm_omni.entrypoints.cli.benchmark.serve import OmniBenchmarkServingSubcommand +from vllm_omni.experimental.fullduplex.client import RealtimeEventCollector +from vllm_omni.utils.tracking_parser import TrackingArgumentParser + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.benchmark] +MODEL = "openbmb/MiniCPM-o-4_5" + + +def _case(tmp_path: Path, *, subset: str = "1q1a", name: str = "video.mp4") -> data.OmniInteractCase: + video, annotation = tmp_path / name, tmp_path / f"{Path(name).stem}.json" + video.parent.mkdir(parents=True, exist_ok=True) + video.touch() + annotation.write_text("{}") + return data.OmniInteractCase(subset, name, video, annotation, "multi_turn") + + +def _collector(*events: tuple[dict[str, object], float]) -> RealtimeEventCollector: + collector = RealtimeEventCollector() + for event, received_at in events: + collector.add(event, received_at_s=received_at) + return collector + + +def _created(response_id: str = "r1") -> dict[str, object]: + return {"type": "response.created", "response": {"id": response_id}} + + +def _done(response_id: str = "r1", status: str = "completed") -> dict[str, object]: + return {"type": "response.done", "response": {"id": response_id, "status": status}} + + +def _audio( + response_id: str = "r1", + *, + samples: int = 2400, + rate: int | None = 24_000, + value: int = 1, +) -> dict[str, object]: + event: dict[str, object] = { + "type": "response.audio.delta", + "response_id": response_id, + "format": "pcm16", + "delta": base64.b64encode(bytes((value, 0)) * samples).decode(), + } + if rate is not None: + event["sample_rate_hz"] = rate + return event + + +def _text(response_id: str = "r1", value: str = "hello") -> dict[str, object]: + return {"type": "response.audio_transcript.delta", "response_id": response_id, "delta": value} + + +def _listen(*, buffering: bool = False) -> dict[str, object]: + return { + "type": "response.listen", + "response": {"metadata": {"model_listen": True, "buffering": buffering}}, + } + + +def _session_options(tmp_path: Path) -> data.OmniInteractSessionOptions: + return data.OmniInteractSessionOptions(tmp_path / "artifacts", 10.0, 5.0, "reference.wav", True) + + +def _sample(case: data.OmniInteractCase, options: data.OmniInteractSessionOptions): + return data.OmniInteractSampleRequest( + prompt="", + prompt_len=0, + expected_output_len=0, + multi_modal_data=None, + request_id="measured-0", + omniinteract_case=case, + omniinteract_options=options, + ) + + +def _write_dataset(root: Path) -> Path: + data_root = root / "data" + for subset in ("1q1a", "1q1a_math"): + subset_root = data_root / subset + (subset_root / "videos").mkdir(parents=True) + (subset_root / "annotations").mkdir() + (subset_root / "videos" / f"{subset}.mp4").touch() + (subset_root / "annotations" / f"{subset}.json").write_text("{}") + (subset_root / "video_json_map.json").write_text( + json.dumps( + { + "entries": [ + { + "video": f"videos/{subset}.mp4", + "annotation": f"annotations/{subset}.json", + "scene_type": "multi_turn", + } + ] + } + ) + ) + one_to_many = data_root / "1qna" + (one_to_many / "videos_bench" / "nested").mkdir(parents=True) + (one_to_many / "annotations" / "nested").mkdir(parents=True) + (one_to_many / "videos_bench" / "nested" / "guide.mp4").touch() + (one_to_many / "annotations" / "nested" / "guide.json").write_text("{}") + return data_root + + +def _successful_output( + tmp_path: Path, + *, + audio_time: float = 10.0, + status: str = "completed", +) -> tuple[data.OmniInteractCase, RealtimeEventCollector, oi.OmniInteractCaseResult]: + case = _case(tmp_path) + collector = _collector( + (_created(), 9.9), + (_audio(), audio_time), + (_text(), audio_time + 0.01), + (_done(status=status), audio_time + 0.1), + ) + result = oi.OmniInteractCaseResult(case.subset, str(case.video_path), "") + return case, collector, result + + +def _write_success( + root: Path, + case: data.OmniInteractCase, + collector: RealtimeEventCollector, + result: oi.OmniInteractCaseResult, + **kwargs, +): + require_response = kwargs.pop("require_response", True) + return oi.prepare_success_artifacts( + root, + case, + collector, + stream_start=10.0, + video_duration_s=1.0, + require_response=require_response, + result=result, + **kwargs, + ) + + +def test_dataset_discovers_official_layouts_and_total_selection(tmp_path: Path): + root = _write_dataset(tmp_path) + all_cases = data.discover_omniinteract_cases(root, data.OMNIINTERACT_SUBSETS, num_prompts=0, disable_shuffle=True) + selected = data.discover_omniinteract_cases(root, data.OMNIINTERACT_SUBSETS, num_prompts=2, disable_shuffle=True) + assert [case.subset for case in all_cases] == ["1q1a", "1q1a_math", "1qna"] + assert all_cases[-1].video_rel == "videos_bench/nested/guide.mp4" + assert [case.subset for case in selected] == ["1q1a", "1q1a_math"] + + +def _archive(path: Path, member: str) -> None: + with tarfile.open(path, "w") as archive: + content = b"{}" + item = tarfile.TarInfo(member) + item.size = len(content) + archive.addfile(item, io.BytesIO(content)) + + +def test_archive_is_safe_and_atomically_shared(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + bad = tmp_path / "bad.tar" + _archive(bad, "../escape") + with pytest.raises(ValueError, match="Unsafe path"): + data._extract_archive(bad, tmp_path / "bad-cache") + + good = tmp_path / "good.tar" + _archive(good, "data/1q1a/video_json_map.json") + target = tmp_path / "cache" + barrier, safe_extract = threading.Barrier(2), data._safe_extract + monkeypatch.setattr(data, "_safe_extract", lambda *args: (barrier.wait(), safe_extract(*args))) + with ThreadPoolExecutor(2) as executor: + roots = list(executor.map(lambda _: data._extract_archive(good, target), range(2))) + assert roots[0] == roots[1] + assert (roots[0] / "1q1a" / "video_json_map.json").is_file() + assert not list(target.glob(".tmp-*")) + + +def test_hub_archive_uses_vllm_filesystem(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + source = tmp_path / "source.tar" + _archive(source, "data/1q1a/video_json_map.json") + + class FS: + def get_file(self, remote: str, local: str) -> None: + assert remote == "datasets/org/repo/data.tar.gz" + Path(local).write_bytes(source.read_bytes()) + + monkeypatch.setenv("HF_HOME", str(tmp_path / "hf")) + monkeypatch.setattr(data, "hf_fs", lambda: FS()) + assert (data.resolve_omniinteract_root(None, "org/repo") / "1q1a").is_dir() + + +def test_media_commands_are_bounded(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + calls: list[float] = [] + commands: list[list[str]] = [] + bounded_run = oi._run_media_command + + def run(command, *, timeout_s, text=False): + calls.append(timeout_s) + commands.append(command) + if command[0] == "ffprobe": + return subprocess.CompletedProcess(command, 0, "1.0", "") + return subprocess.CompletedProcess(command, 0, b"", b"") + + monkeypatch.setattr(oi, "_run_media_command", run) + duration, pcm, frames = oi.prepare_media(tmp_path / "video.mp4", 1.0, timeout_s=3.0, max_duration_s=1.0) + assert (duration, len(pcm), frames, calls) == (1.0, 32_000, [None], [3.0, 3.0, 3.0]) + assert commands[1][commands[1].index("-t") + 1] == "1.0" + assert commands[1][commands[1].index("-fs") + 1] == "32000" + + monkeypatch.setattr(subprocess, "run", lambda *a, **k: (_ for _ in ()).throw(subprocess.TimeoutExpired("x", 3))) + with pytest.raises(TimeoutError, match="timed out"): + bounded_run(["ffmpeg"], timeout_s=3.0) + + +def test_media_duration_is_bounded_before_decode(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + calls = 0 + + def run(command, *, timeout_s, text=False): + nonlocal calls + calls += 1 + return subprocess.CompletedProcess(command, 0, "601.0", "") + + monkeypatch.setattr(oi, "_run_media_command", run) + with pytest.raises(ValueError, match="Invalid video duration"): + oi.prepare_media(tmp_path / "video.mp4", 1.0, timeout_s=3.0, max_duration_s=600.0) + assert calls == 1 + + +def test_final_commit_keeps_a_partial_model_unit(): + """Exact units retain a sample for the server's correlated final flush.""" + + events = [{"session": {"capabilities": {"chunk_period_ms": 1000}}}] + exact_unit = b"\0\0" * 16_000 + assert oi._ensure_final_commit_tail(exact_unit, events) == exact_unit[:-2] + assert oi._ensure_final_commit_tail(exact_unit + b"\0\0", events) == exact_unit + b"\0\0" + + +@pytest.mark.parametrize( + ("events", "match"), + [ + (((_done(), 0.0),), "without response.created"), + (((_created(), 0.0), (_created(), 0.1)), "duplicate response.created"), + (((_created(), 0.0), (_done(), 0.1), (_done(), 0.2)), "duplicate response.done"), + ( + ( + (_created(), 0.0), + ({"type": "response.done", "response": {"id": "r1", "status_details": {"type": "failed"}}}, 0.1), + ), + "reports failure", + ), + ], +) +def test_response_ledger_rejects_identity_errors(events, match: str): + with pytest.raises(ValueError, match=match): + oi.response_ledger(_collector(*events)) + + +class _CompletionClient: + def __init__(self, collector: RealtimeEventCollector): + self.events, self.acks = collector, [] + + def raise_if_reader_stopped(self) -> None: + return None + + async def send_playback_ack(self, response_id: str, played_ms: int) -> None: + self.acks.append((response_id, played_ms)) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("tail", "succeeds"), + [ + (((_created(), 1.1), (_done(), 1.2)), True), + (((_listen(), 1.1),), True), + (((_listen(buffering=True), 1.1),), False), + ((({"type": "session.closed", "reason": "timeout"}, 1.1),), False), + ], +) +async def test_completion_requires_a_final_model_decision(tail, succeeds: bool): + events = _collector(({"type": "input_audio_buffer.committed"}, 1.0), *tail) + coroutine = oi.wait_for_session_completion( + _CompletionClient(events), oi._Playback(), commit_from=0, timeout_s=0.08, settle_s=0.0 + ) + if succeeds: + assert await coroutine == 0 + else: + with pytest.raises((TimeoutError, RuntimeError)): + await coroutine + + +@pytest.mark.asyncio +async def test_deferred_commit_ignores_the_prior_response_terminal(): + events = _collector( + (_created("old"), 0.0), + ({"type": "input_audio_buffer.committed", "event": {"overlap_deferred": True}}, 1.0), + (_done("old"), 1.1), + (_listen(), 1.2), + ) + assert ( + await oi.wait_for_session_completion( + _CompletionClient(events), oi._Playback(), commit_from=1, timeout_s=0.1, settle_s=0.0 + ) + == 1 + ) + + +@pytest.mark.asyncio +async def test_playback_acks_only_after_serial_audio_drain(): + collector = _collector((_audio(samples=2400), 10.0), (_done(), 10.01)) + client, playback = _CompletionClient(collector), oi._Playback() + await playback.acknowledge(client, now=10.05) + assert client.acks == [] + await playback.acknowledge(client, now=10.1) + assert client.acks == [("r1", 100)] + + +@pytest.mark.parametrize( + ("event", "match"), + [ + (_audio(rate=16_000), "24000 Hz"), + ({**_audio(), "format": "opus"}, "pcm16"), + ({**_audio(), "delta": "!"}, "base64"), + ], +) +def test_playback_rejects_non_official_audio(event: dict[str, object], match: str): + playback = oi._Playback() + with pytest.raises(ValueError, match=match): + playback.ingest(_collector((event, 1.0))) + + +def test_artifacts_publish_official_bundle_and_sparse_deferred_state(tmp_path: Path): + case, collector, result = _successful_output(tmp_path) + summary = _write_success( + tmp_path / "out", + case, + collector, + result, + ) + context = result._artifact_context + assert result.success and result.eligible_for_official_eval + assert summary["scene_type"] == "multi_turn" + assert context and context.spans + assert all("delta" not in event for event in context.events if event["type"] == "response.audio.delta") + oi.publish_deferred_case_artifacts(tmp_path / "out", case, result) + directory = oi._output_dir(tmp_path / "out", case) + assert all((directory / name).is_file() for name in oi.SUCCESS_ARTIFACTS) + with wave.open(str(directory / "output.wav")) as output: + assert (output.getframerate(), output.getnframes()) == (24_000, 24_000) + transcript = json.loads((directory / "wav_transcript.json").read_text()) + assert transcript["text"] == "hello" and transcript["chunks"][0]["timestamp"] == [0.0, 0.1] + assert "serialized playback queue" in transcript["timestamp_semantics"] + assert result._artifact_context is None + + +@pytest.mark.parametrize( + ("audio_time", "status", "reason"), + [(10.95, "completed", "audio_clipped"), (10.0, "cancelled", "cancelled_response")], +) +def test_ineligible_outputs_are_excluded_from_manifest(tmp_path: Path, audio_time: float, status: str, reason: str): + case, collector, result = _successful_output(tmp_path, audio_time=audio_time, status=status) + _write_success(tmp_path / "out", case, collector, result, require_response=False) + assert reason in result.official_eval_ineligible_reasons + oi.write_batch_artifacts(tmp_path / "out", [case], [result]) + assert (tmp_path / "out" / "official_eval_manifest.jsonl").read_text() == "" + + +def test_artifacts_require_complete_audio_and_transcript_response(tmp_path: Path): + case, _, result = _successful_output(tmp_path) + collector = _collector((_created(), 10.0), (_audio(), 10.0), (_done(), 10.1)) + with pytest.raises(ValueError, match="audio and transcript"): + _write_success(tmp_path, case, collector, result) + + +def test_publication_failure_removes_partial_success_bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + case, collector, result = _successful_output(tmp_path) + _write_success(tmp_path, case, collector, result) + original = oi._atomic_write_json + + def fail(path: Path, value: object) -> None: + if path.name == "events.json": + raise OSError("disk full") + original(path, value) + + monkeypatch.setattr(oi, "_atomic_write_json", fail) + with pytest.raises(OSError, match="disk full"): + oi.publish_deferred_case_artifacts(tmp_path, case, result) + directory = oi._output_dir(tmp_path, case) + assert not any((directory / name).exists() for name in oi.SUCCESS_ARTIFACTS) + assert not result.success and "artifact_write_failed" in result.official_eval_ineligible_reasons + + +def test_atomic_write_preserves_destination_on_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + destination = tmp_path / "output.wav" + destination.write_bytes(b"old") + monkeypatch.setattr(oi, "write_pcm16_wav", lambda *a, **k: (_ for _ in ()).throw(OSError("write"))) + with pytest.raises(OSError): + oi._atomic_write_wav(destination, b"\0\0", 24_000) + assert destination.read_bytes() == b"old" + assert not list(tmp_path.glob(".*.tmp")) + + +class _RealtimeClient: + instances: list[_RealtimeClient] = [] + + def __init__(self, url: str, **kwargs): + self.url, self.events, self.acks, self.configure_kwargs = url, RealtimeEventCollector(), [], {} + self.instances.append(self) + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + async def configure(self, model: str, **kwargs) -> None: + self.configure_kwargs = kwargs + self.events.add({"type": "session.created", "session": {"capabilities": {"chunk_period_ms": 1000}}}) + + async def send(self, event: dict[str, object]) -> None: + return None + + async def commit(self) -> None: + now = time.monotonic() + for event, offset in ( + ({"type": "input_audio_buffer.committed"}, 0.0), + (_created(), 0.001), + (_audio(samples=24), 0.002), + (_text(value="answer"), 0.003), + (_done(), 0.004), + ): + self.events.add(event, received_at_s=now + offset) + + async def send_playback_ack(self, response_id: str, played_ms: int) -> None: + self.acks.append((response_id, played_ms)) + + async def close_session(self, **kwargs) -> None: + self.events.add({"type": "session.closed"}) + + def raise_if_reader_stopped(self) -> None: + return None + + +@pytest.mark.asyncio +async def test_public_runner_executes_one_prepared_session(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + case = _case(tmp_path) + config = oi.OmniInteractBenchmarkConfig( + model=MODEL, + output_root=tmp_path / "out", + ref_audio="ref.wav", + require_response=True, + extra_body={"custom": "value"}, + ) + prepared = data.OmniInteractPreparedInput(1.0, b"\0\0\0\0", ("frame",), "data:audio/wav;base64,ref") + monkeypatch.setattr(oi, "RealtimeDuplexClient", _RealtimeClient) + monkeypatch.setattr(oi, "_COMPLETION_SETTLE_S", 0.0) + result = await oi.run_omniinteract_case( + case, + config, + request_index=0, + prepared_input=prepared, + ) + assert result.success and result.transcript == "answer" and result._artifact_context is not None + assert not config.output_root.exists() + assert "autostart=0" in _RealtimeClient.instances[-1].url + assert _RealtimeClient.instances[-1].configure_kwargs["extra_body"] == {"custom": "value"} + assert _RealtimeClient.instances[-1].acks == [("r1", 1)] + + +def test_standard_sample_loading_prepares_media_before_timing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + _write_dataset(tmp_path) + ref = tmp_path / "ref.wav" + ref.touch() + parser = TrackingArgumentParser() + OmniBenchmarkServingSubcommand.add_cli_args(parser) + args = parser.parse_args( + [ + "--backend", + "openai-realtime-duplex", + "--dataset-name", + "omniinteract", + "--dataset-path", + str(tmp_path), + "--model", + MODEL, + "--endpoint", + "/v1/realtime", + "--num-prompts", + "0", + "--omniinteract-ref-audio", + str(ref), + ] + ) + preprocess_serve_args(args) + assert args.endpoint == "/v1/realtime" + monkeypatch.setattr(benchmark_patch, "reference_audio_data_url", lambda _: "data:audio/wav;base64,ref") + monkeypatch.setattr(benchmark_patch, "prepare_media", lambda *a, **k: (1.0, b"pcm", ["frame"])) + samples = benchmark_patch.get_samples(args, None) + assert len(samples) == args.num_prompts == 3 + assert all(sample.omniinteract_prepared_input.video_frames == ("frame",) for sample in samples) + + +def _request(case: data.OmniInteractCase, options: data.OmniInteractSessionOptions) -> RequestFuncInput: + request = RequestFuncInput( + model=MODEL, + model_name=MODEL, + prompt="", + api_url="http://server:8000/v1/realtime", + prompt_len=0, + output_len=0, + logprobs=None, + multi_modal_content=None, + ignore_eos=False, + request_id="measured-0", + ) + request.omniinteract_case, request.omniinteract_options = case, options + return request + + +@pytest.mark.asyncio +async def test_adapter_requires_and_forwards_exact_prepared_payload(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + case, options = _case(tmp_path), _session_options(tmp_path) + request = _request(case, options) + output = await benchmark_patch.async_request_openai_realtime_duplex(request, session=None) + assert not output.success and "prepared before benchmark request timing" in output.error + + prepared = data.OmniInteractPreparedInput(1.0, b"pcm", ("frame",), "data:audio/wav;base64,ref") + request.omniinteract_prepared_input = prepared + request.extra_headers = {"X-Proxy": "test"} + request.extra_body = {"custom": "value"} + monkeypatch.setenv("OPENAI_API_KEY", "secret") + seen = {} + + async def run(_case_arg, config, **kwargs): + seen["headers"] = config.extra_headers + seen["extra_body"] = config.extra_body + seen.update(kwargs) + return oi.OmniInteractCaseResult( + case.subset, + str(case.video_path), + str(options.output_root), + success=True, + transcript="timing metadata missing", + output_tokens=0, + duplex_request_metrics=[{"request_metrics": {"ttft_ms": 1.0}}], + duplex_session_metrics={"mean_ttft_ms": 10.0}, + ) + + monkeypatch.setattr(benchmark_patch, "run_omniinteract_case", run) + output = await benchmark_patch.async_request_openai_realtime_duplex(request, session=None) + assert output.success and seen["prepared_input"] is prepared + assert seen["headers"] == {"Authorization": "Bearer secret", "X-Proxy": "test", "x-request-id": "measured-0"} + assert seen["extra_body"] == {"custom": "value"} + assert output.itl == [] + assert output.tpot_measured is False + assert seen["capture_artifacts"] is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("request_metrics", "expected_itl", "expected_text_latency"), + [ + ( + [ + {"stage0_tokens": {"output_token_count": 3, "itls_ms": [10.0, 20.0]}}, + {"stage0_tokens": {"output_token_count": 2, "itls_ms": [30.0]}}, + ], + [0.01, 0.02, 0.03], + 0.16, + ), + ( + [ + {"stage0_tokens": {"output_token_count": 3, "tpot_ms": 10.0}}, + {"stage0_tokens": {"output_token_count": 2, "tpot_ms": 40.0}}, + ], + [], + 0.18, + ), + ( + [ + {"stage0_tokens": {"output_token_count": 3, "tpot_ms": 0.0}}, + {"stage0_tokens": {"output_token_count": 2, "tpot_ms": 0.0}}, + ], + [], + 0.1, + ), + ], +) +async def test_adapter_reports_exact_or_weighted_token_timing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + request_metrics: list[dict[str, object]], + expected_itl: list[float], + expected_text_latency: float, +): + case, options = _case(tmp_path), _session_options(tmp_path) + request = _request(case, options) + request.omniinteract_prepared_input = data.OmniInteractPreparedInput( + 1.0, b"pcm", ("frame",), "data:audio/wav;base64,ref" + ) + + async def run(*args, **kwargs): + return oi.OmniInteractCaseResult( + case.subset, + str(case.video_path), + str(options.output_root), + success=True, + output_tokens=5, + duplex_request_metrics=request_metrics, + duplex_session_metrics={"mean_ttft_ms": 100.0}, + ) + + monkeypatch.setattr(benchmark_patch, "run_omniinteract_case", run) + output = await benchmark_patch.async_request_openai_realtime_duplex(request, session=None) + + assert output.success + assert output.itl == expected_itl + assert output.text_latency == pytest.approx(expected_text_latency) + assert output.tpot_measured is True + + +def test_batch_finalization_publishes_only_measured_results(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + case, options = _case(tmp_path), _session_options(tmp_path) + options.output_root.mkdir() + sample = _sample(case, options) + result = oi.OmniInteractCaseResult( + case.subset, str(case.video_path), str(options.output_root), success=True, eligible_for_official_eval=True + ) + output = benchmark_patch.MixRequestFuncOutput(success=True) + output.omniinteract_case_result = result + published = [] + monkeypatch.setattr(benchmark_patch, "publish_deferred_case_artifacts", lambda *args: published.append(args)) + summary = benchmark_patch._finalize_omniinteract_batch([sample], [output]) + assert summary and summary["total"] == summary["success"] == summary["eligible_for_official_eval"] == 1 + assert summary["failed"] == summary["successful_but_ineligible"] == summary["audio_clipped_bytes"] == 0 + assert published == [(options.output_root, case, result)] + + +def test_case_artifact_failures_do_not_discard_serving_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + case, options = _case(tmp_path), _session_options(tmp_path) + sample = _sample(case, options) + result = oi.OmniInteractCaseResult( + case.subset, str(case.video_path), str(options.output_root), success=True, eligible_for_official_eval=True + ) + output = benchmark_patch.MixRequestFuncOutput(success=True, error="inference warning") + output.omniinteract_case_result = result + monkeypatch.setattr( + benchmark_patch, + "publish_deferred_case_artifacts", + lambda *args: (_ for _ in ()).throw(OSError("case disk full")), + ) + monkeypatch.setattr( + benchmark_patch, + "write_failure_artifacts", + lambda *args: (_ for _ in ()).throw(OSError("failure disk full")), + ) + monkeypatch.setattr(benchmark_patch, "write_omniinteract_batch_artifacts", lambda *args: None) + + summary = benchmark_patch._finalize_omniinteract_batch([sample], [output]) + metrics, _ = calculate_metrics( + [], + [output], + 1.0, + None, + [50.0], + {}, + TaskType.GENERATION, + [], + None, + float("inf"), + 1.0, + ) + + assert summary and summary["failed"] == 1 + assert summary["artifacts_complete"] is False + assert len(summary["artifact_errors"]) == 2 + assert output.success and metrics.completed == 1 + assert output.error.startswith("inference warning\n") + assert "Artifact publication failed: case disk full" in output.error + + +def test_batch_artifact_failure_removes_partial_batch_files(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + case, options = _case(tmp_path), _session_options(tmp_path) + sample = _sample(case, options) + result = oi.OmniInteractCaseResult( + case.subset, str(case.video_path), str(options.output_root), success=True, eligible_for_official_eval=True + ) + output = benchmark_patch.MixRequestFuncOutput(success=True) + output.omniinteract_case_result = result + monkeypatch.setattr(benchmark_patch, "publish_deferred_case_artifacts", lambda *args: None) + + def fail_batch(root, *args): + root.mkdir(parents=True, exist_ok=True) + (root / "batch_summary.json").write_text("partial") + raise OSError("batch disk full") + + monkeypatch.setattr(benchmark_patch, "write_omniinteract_batch_artifacts", fail_batch) + + summary = benchmark_patch._finalize_omniinteract_batch([sample], [output]) + + assert summary and summary["success"] == 1 and summary["artifacts_complete"] is False + assert summary["artifact_errors"] == ["Batch artifact publication failed: batch disk full"] + assert output.success + assert not (options.output_root / "batch_summary.json").exists() + + +@pytest.mark.parametrize( + ("field", "value", "match"), + [ + ("omniinteract_ref_audio", None, "ref-audio"), + ("endpoint", "/v1/completions", "v1/realtime"), + ("max_concurrency", 0, "positive"), + ("skip_tokenizer_init", True, "skip-tokenizer-init"), + ], +) +def test_cli_rejects_incompatible_omniinteract_options(field: str, value, match: str): + args = argparse.Namespace( + dataset_name="omniinteract", + backend="openai-realtime-duplex", + endpoint="/v1/realtime", + omniinteract_ref_audio="ref.wav", + max_concurrency=1, + skip_tokenizer_init=False, + ) + setattr(args, field, value) + with pytest.raises(ValueError, match=match): + preprocess_serve_args(args) + + +def test_whole_benchmark_lock_serializes_a_shared_output_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + entered, release = threading.Event(), threading.Event() + calls: list[int] = [] + + async def fake_main_async(args): + calls.append(len(calls)) + if len(calls) == 1: + entered.set() + await asyncio.to_thread(release.wait) + return {"total": 1} + + monkeypatch.setattr(benchmark_serve, "main_async", fake_main_async) + args = argparse.Namespace(dataset_name="omniinteract", omniinteract_output_dir=tmp_path, extra_body=None) + with ThreadPoolExecutor(2) as executor: + first = executor.submit(benchmark_serve.main, args) + assert entered.wait(1) + second = executor.submit(benchmark_serve.main, args) + time.sleep(0.05) + assert calls == [0] + release.set() + assert first.result() == second.result() == {"total": 1} + assert calls == [0, 1] diff --git a/tests/benchmarks/test_serve_cli.py b/tests/benchmarks/test_serve_cli.py index 7dd12d0003b..5365d56ac7d 100644 --- a/tests/benchmarks/test_serve_cli.py +++ b/tests/benchmarks/test_serve_cli.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + import argparse import json import os @@ -22,6 +25,7 @@ "dataset_name", [ "daily-omni", + "omniinteract", "seed-tts", "seed-tts-text", "seed-tts-design", @@ -55,6 +59,7 @@ def test_extend_omni_choices_updates_tracking_parser_shadow(dataset_name: str) - [ (["--print-stage"], "print_stage", True), (["--daily-omni-input-mode", "audio"], "daily_omni_input_mode", "audio"), + (["--omniinteract-subsets", "1qna"], "omniinteract_subsets", ["1qna"]), (["--seed-tts-locale", "zh"], "seed_tts_locale", "zh"), ], ) @@ -80,12 +85,14 @@ def test_add_omni_args_preserves_implicit_defaults() -> None: args = parser.parse_args([]) assert args.print_stage is False assert args.daily_omni_input_mode == "all" + assert args.omniinteract_subsets == ["1q1a", "1q1a_math", "1qna"] assert args.seed_tts_locale == "en" assert args.explicit_keys == set() def test_update_omni_help_updates_upstream_actions() -> None: parser = TrackingArgumentParser() + parser.add_argument("--num-prompts", default=1000, help="Number of prompts.") parser.add_argument("--percentile-metrics", help="Upstream percentile help.") parser.add_argument("--random-mm-limit-mm-per-prompt", help="Upstream limit help.") parser.add_argument("--random-mm-bucket-config", help="Upstream bucket help.") @@ -93,6 +100,7 @@ def test_update_omni_help_updates_upstream_actions() -> None: update_omni_help(parser) actions = {action.dest: action for action in parser._actions} + assert "OmniInteract uses 3" in actions["num_prompts"].help assert all(metric in actions["percentile_metrics"].help for metric in ("ttfc", "tpop", "audio_rtf")) assert "probabilities are renormalized" in actions["random_mm_limit_mm_per_prompt"].help assert "Currently allows for 3 modalities" in actions["random_mm_bucket_config"].help @@ -120,6 +128,31 @@ def test_preprocess_serve_args_merges_bot_task_without_overriding_extra_body( assert args.extra_body == expected +@pytest.mark.parametrize( + ("argv", "expected"), + [ + ([], 3), + (["--num-prompts", "0"], 0), + (["--num-prompts", "12"], 12), + ], +) +def test_preprocess_serve_args_applies_safe_omniinteract_prompt_default( + argv: list[str], + expected: int, +) -> None: + parser = TrackingArgumentParser() + parser.add_argument("--dataset-name", default="omniinteract") + parser.add_argument("--backend", default="openai-realtime-duplex") + parser.add_argument("--endpoint", default="/v1/realtime") + parser.add_argument("--omniinteract-ref-audio", default="reference.wav") + parser.add_argument("--num-prompts", type=int, default=1000) + args = parser.parse_args(argv) + + preprocess_serve_args(args) + + assert args.num_prompts == expected + + @pytest.mark.parametrize( ("argv", "expected_extra_body", "expected_explicit"), [ diff --git a/tests/e2e/features/fullduplex/test_client.py b/tests/e2e/features/fullduplex/test_client.py index d90c9e5aa3c..601bb1d1c6a 100644 --- a/tests/e2e/features/fullduplex/test_client.py +++ b/tests/e2e/features/fullduplex/test_client.py @@ -1,3 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +import asyncio import base64 import wave from urllib.parse import parse_qs, urlsplit @@ -20,7 +24,7 @@ def test_realtime_client_builds_explicit_native_duplex_url(): url = build_realtime_url( - "ws://localhost:8099/v1/realtime?custom=1", + "ws://localhost:8099/v1/realtime?custom=1&duplex=0&model=stale&minicpmo45_native_duplex=0&session_id=stale", "openbmb/MiniCPM-o-4_5", session_id="session-a", ) @@ -50,7 +54,7 @@ def test_seed_tts_initial_text_is_part_of_native_duplex_context(): def test_realtime_client_builds_resume_only_url_when_autostart_disabled(): url = build_realtime_url( - "ws://localhost:8099/v1/realtime?duplex=1", + "ws://localhost:8099/v1/realtime?duplex=1&autostart=1", "openbmb/MiniCPM-o-4_5", autostart=False, ) @@ -73,10 +77,15 @@ async def send(self, event): client = Client() - await client.configure("openbmb/MiniCPM-o-4_5", timeout_s=1) + await client.configure( + "openbmb/MiniCPM-o-4_5", + idle_timeout_s=900, + timeout_s=1, + ) session = client.sent[0]["session"] assert "ref_audio" not in session + assert session["idle_timeout_s"] == 900 @pytest.mark.asyncio @@ -100,6 +109,74 @@ async def send(self, event): session = client.sent[0]["session"] assert session["ref_audio"] == "data:audio/wav;base64,AAAA" + assert "idle_timeout_s" not in session + + +@pytest.mark.asyncio +async def test_realtime_client_reports_reader_exit_to_event_waiters(): + client = RealtimeDuplexClient("ws://unused") + + async def stopped(): + return None + + client._reader_task = asyncio.create_task(stopped()) + await client._reader_task + + with pytest.raises(ConnectionError, match="closed before"): + client.raise_if_reader_stopped() + + +@pytest.mark.asyncio +async def test_realtime_client_configure_reports_reader_failure_immediately(): + class Client(RealtimeDuplexClient): + async def send(self, event): + async def fail(): + raise RuntimeError("reader failed") + + self._reader_task = asyncio.create_task(fail()) + await asyncio.sleep(0) + + client = Client("ws://unused") + with pytest.raises(ConnectionError, match="reader failed"): + await client.configure("model", timeout_s=10.0) + + +@pytest.mark.asyncio +async def test_realtime_client_close_reports_reader_cancellation_immediately(): + class Client(RealtimeDuplexClient): + async def send(self, event): + async def wait_forever(): + await asyncio.Future() + + self._reader_task = asyncio.create_task(wait_forever()) + self._reader_task.cancel() + with pytest.raises(asyncio.CancelledError): + await self._reader_task + + client = Client("ws://unused") + with pytest.raises(ConnectionError, match="reader was cancelled"): + await client.close_session(timeout_s=10.0) + + +@pytest.mark.asyncio +async def test_realtime_client_close_waits_for_a_new_session_closed_event(): + class Client(RealtimeDuplexClient): + async def send(self, event): + assert event["type"] == "session.close" + + async def acknowledge_close(): + await asyncio.sleep(0.02) + self.events.add({"type": "session.closed"}) + + asyncio.create_task(acknowledge_close()) + + client = Client("ws://unused") + client.events.add({"type": "session.closed", "reason": "old"}) + started_at = asyncio.get_running_loop().time() + + await client.close_session(timeout_s=0.2) + + assert asyncio.get_running_loop().time() - started_at >= 0.01 @pytest.mark.asyncio @@ -235,6 +312,7 @@ def test_realtime_event_collector_reports_engine_token_and_audio_intervals(): "output_token_count": 4, "ttft_ms": 120.0, "tpot_ms": 15.0, + "itls_ms": [10.0, 14.0, 18.0], "inter_token_interval_ms": { "count": 3, "mean": 14.0, diff --git a/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py b/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py index d5355887dad..9ac816057f0 100644 --- a/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py +++ b/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py @@ -58,18 +58,14 @@ read_pcm16_wav, summarize_session_request_metrics, ) +from vllm_omni.experimental.fullduplex.client import ( # noqa: E402 + reference_audio_data_url as _ref_audio_data_url, +) _url_with_model = build_realtime_url _read_wav_pcm16 = read_pcm16_wav -def _ref_audio_data_url(path: str | None) -> str | None: - if path is None: - return None - ref_path = Path(path).expanduser() - return "data:audio/wav;base64," + base64.b64encode(ref_path.read_bytes()).decode("ascii") - - @dataclass class DemoState: events: list[dict[str, object]] = field(default_factory=list) diff --git a/vllm_omni/benchmarks/data_modules/omniinteract_dataset.py b/vllm_omni/benchmarks/data_modules/omniinteract_dataset.py new file mode 100644 index 00000000000..04726eee6ad --- /dev/null +++ b/vllm_omni/benchmarks/data_modules/omniinteract_dataset.py @@ -0,0 +1,306 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""OmniInteract dataset discovery for the standard serving benchmark.""" + +from __future__ import annotations + +import json +import logging +import os +import random +import shutil +import tarfile +import tempfile +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from vllm.benchmarks.datasets import BenchmarkDataset, SampleRequest +from vllm.tokenizers import TokenizerLike +from vllm.transformers_utils.repo_utils import hf_fs + +OMNIINTERACT_SUBSETS = ("1q1a", "1q1a_math", "1qna") +DEFAULT_OMNIINTERACT_REPO = "lucky-lance/OmniInteract" +DEFAULT_MAX_VIDEO_DURATION_S = 3600.0 +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class OmniInteractCase: + subset: str + video_rel: str + video_path: Path + annotation_path: Path + scene_type: str + + +@dataclass(frozen=True) +class OmniInteractSessionOptions: + output_root: Path + timeout_s: float + media_timeout_s: float + ref_audio: str + require_response: bool = False + max_video_duration_s: float = DEFAULT_MAX_VIDEO_DURATION_S + + +@dataclass(frozen=True) +class OmniInteractPreparedInput: + duration_s: float + pcm16: bytes = field(repr=False) + video_frames: tuple[str | None, ...] = field(repr=False) + ref_audio_data_url: str = field(repr=False) + + +@dataclass +class OmniInteractSampleRequest(SampleRequest): + omniinteract_case: OmniInteractCase | None = None + omniinteract_options: OmniInteractSessionOptions | None = None + omniinteract_prepared_input: OmniInteractPreparedInput | None = field(default=None, repr=False) + + +class OmniInteractDataset(BenchmarkDataset): + def __init__( + self, + *, + data_root: str | None, + dataset_repo: str = DEFAULT_OMNIINTERACT_REPO, + subsets: Sequence[str] = OMNIINTERACT_SUBSETS, + random_seed: int = 0, + disable_shuffle: bool = False, + ) -> None: + super().__init__( + dataset_path=data_root or dataset_repo, + random_seed=random_seed, + disable_shuffle=disable_shuffle, + ) + self.root = resolve_omniinteract_root(data_root, dataset_repo) + self.subsets = tuple(subsets) + + def sample( + self, + tokenizer: TokenizerLike | None, + num_requests: int, + *, + request_id_prefix: str = "", + options: OmniInteractSessionOptions, + **_: Any, + ) -> list[SampleRequest]: + del tokenizer + cases = discover_omniinteract_cases( + self.root, + self.subsets, + num_prompts=num_requests, + seed=self.random_seed, + disable_shuffle=self.disable_shuffle, + ) + return [ + OmniInteractSampleRequest( + prompt="", + prompt_len=0, + expected_output_len=0, + multi_modal_data=None, + request_id=f"{request_id_prefix}{index}", + omniinteract_case=case, + omniinteract_options=options, + ) + for index, case in enumerate(cases) + ] + + +def _data_dir(root: Path) -> Path: + for candidate in (root, root / "data"): + if any((candidate / subset).is_dir() for subset in OMNIINTERACT_SUBSETS): + return candidate + raise FileNotFoundError(f"OmniInteract data not found under {root}") + + +def _confined_path(root: Path, relative: object, *, field: str) -> Path: + text, resolved_root = str(relative or ""), root.resolve() + path = Path(text) + destination = (root / path).resolve() + if not text or path.is_absolute() or ".." in path.parts or not destination.is_relative_to(resolved_root): + raise ValueError(f"Unsafe OmniInteract {field} path: {text!r}") + return destination + + +def _safe_extract(archive: tarfile.TarFile, target: Path) -> None: + root = target.resolve() + members = archive.getmembers() + for member in members: + path = Path(member.name) + if ( + path.is_absolute() + or ".." in path.parts + or not (root / path).resolve().is_relative_to(root) + or not (member.isdir() or member.isfile()) + ): + raise ValueError(f"Unsafe path in OmniInteract archive: {member.name!r}") + archive.extractall(target, members=members, filter="data") + + +def _extract_archive(archive: Path, target: Path) -> Path: + if target.is_symlink(): + raise ValueError(f"Refusing to extract through symlink: {target}") + stat = archive.stat() + published = target / f"{stat.st_size}-{stat.st_mtime_ns}" + if published.is_symlink(): + raise ValueError(f"Refusing to use symlinked extraction: {published}") + if published.is_dir(): + try: + return _data_dir(published) + except FileNotFoundError: + pass + target.mkdir(parents=True, exist_ok=True) + staging = Path(tempfile.mkdtemp(prefix=".tmp-", dir=target)) + try: + with tarfile.open(archive, "r:*") as handle: + _safe_extract(handle, staging) + root = _data_dir(staging) + try: + staging.replace(published) + return published / root.relative_to(staging) + except OSError: + if not published.is_dir(): + raise + return _data_dir(published) + finally: + shutil.rmtree(staging, ignore_errors=True) + + +def resolve_omniinteract_root(data_root: str | None, dataset_repo: str = DEFAULT_OMNIINTERACT_REPO) -> Path: + """Resolve a local tree/archive or download the official archive.""" + + if data_root: + local = Path(data_root).expanduser().resolve() + if local.is_file(): + return _extract_archive(local, local.parent / f".{local.stem}.vllm_omni_extracted") + if not local.is_dir(): + raise FileNotFoundError(f"--dataset-path does not exist: {local}") + try: + return _data_dir(local) + except FileNotFoundError: + for name in ("data.tar.gz", "data.tar"): + if (archive := local / name).is_file(): + return _extract_archive(archive, local / ".vllm_omni_extracted") + raise + + cache = Path(os.environ.get("HF_HOME", Path.home() / ".cache" / "huggingface")) + target = cache / "vllm_omni" / "omniinteract" / dataset_repo.replace("/", "__") + downloads = target / ".downloads" + downloads.mkdir(parents=True, exist_ok=True) + filesystem, errors = hf_fs(), [] + for name in ("data.tar.gz", "data.tar"): + archive = downloads / name + try: + if not archive.is_file(): + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{name}.", dir=downloads) + os.close(descriptor) + temporary = Path(temporary_name) + try: + filesystem.get_file(f"datasets/{dataset_repo}/{name}", str(temporary)) + if not temporary.stat().st_size: + raise ValueError("downloaded archive is empty") + temporary.replace(archive) + finally: + temporary.unlink(missing_ok=True) + return _extract_archive(archive, target) + except Exception as exc: # noqa: BLE001, PERF203 - Hub errors vary by version + errors.append(f"{name}: {exc}") + raise FileNotFoundError(f"Could not download OmniInteract from {dataset_repo!r}: {'; '.join(errors)}") + + +def _mapping_cases(root: Path, subset: str) -> list[OmniInteractCase]: + mapping_path = root / "video_json_map.json" + if not mapping_path.is_file(): + raise FileNotFoundError(f"Missing OmniInteract mapping: {mapping_path}") + mapping = json.loads(mapping_path.read_text()) + entries = mapping.get("entries") if isinstance(mapping, dict) else None + if not isinstance(entries, list): + raise ValueError(f"Invalid OmniInteract mapping: {mapping_path}") + cases = [] + for row in entries: + if not isinstance(row, dict): + raise ValueError(f"Invalid OmniInteract mapping row in {mapping_path}") + video_rel, annotation_rel = str(row.get("video") or ""), str(row.get("annotation") or "") + video = _confined_path(root, video_rel, field="video") + annotation = _confined_path(root, annotation_rel, field="annotation") + if not video.is_file() or not annotation.is_file(): + raise FileNotFoundError(f"OmniInteract mapping references missing files: {video_rel!r}, {annotation_rel!r}") + cases.append( + OmniInteractCase( + subset, + video_rel, + video, + annotation, + str(row.get("scene_type") or "multi_turn").lower(), + ) + ) + return cases + + +def _one_to_many_cases(root: Path) -> list[OmniInteractCase]: + videos, annotations = root / "videos_bench", root / "annotations" + if not videos.is_dir() or not annotations.is_dir(): + raise FileNotFoundError(f"Invalid OmniInteract 1qna layout under {root}") + cases = [] + for video in sorted(videos.rglob("*.mp4")): + if not video.resolve().is_relative_to(videos.resolve()): + raise ValueError(f"Unsafe OmniInteract video path: {video}") + relative = video.relative_to(videos) + annotation = (annotations / relative).with_suffix(".json").resolve() + if not annotation.is_relative_to(annotations.resolve()): + raise ValueError(f"Unsafe OmniInteract annotation path: {relative}") + if not annotation.is_file(): + raise FileNotFoundError(f"Missing OmniInteract annotation: {annotation}") + cases.append(OmniInteractCase("1qna", f"videos_bench/{relative.as_posix()}", video, annotation, "1qna")) + return cases + + +def discover_omniinteract_cases( + root: Path, + subsets: Sequence[str], + *, + num_prompts: int, + seed: int = 0, + disable_shuffle: bool = False, +) -> list[OmniInteractCase]: + invalid = set(subsets) - set(OMNIINTERACT_SUBSETS) + if invalid: + raise ValueError(f"Unsupported OmniInteract subsets: {sorted(invalid)}") + if not subsets: + raise ValueError("At least one OmniInteract subset is required") + if len(set(subsets)) != len(subsets): + raise ValueError("OmniInteract subsets must not contain duplicates") + if num_prompts < 0: + raise ValueError("num_prompts must be non-negative") + data_root, cases = _data_dir(root.resolve()), [] + for subset in subsets: + subset_root = data_root / subset + selected = _one_to_many_cases(subset_root) if subset == "1qna" else _mapping_cases(subset_root, subset) + if not selected: + raise ValueError(f"No OmniInteract sessions found for requested subset {subset!r}") + cases.extend(selected) + if len({case.video_path for case in cases}) != len(cases): + raise ValueError("OmniInteract dataset contains duplicate video paths") + if not disable_shuffle: + random.Random(seed).shuffle(cases) + if num_prompts: + if num_prompts > len(cases): + logger.warning( + "Requested %d OmniInteract prompts but only %d are available; using all cases", num_prompts, len(cases) + ) + cases = cases[:num_prompts] + return cases + + +def case_manifest(case: OmniInteractCase, output_dir: Path) -> dict[str, Any]: + return { + "sample_id": f"{case.subset}__{output_dir.name}", + "video": str(case.video_path), + "gt_json": str(case.annotation_path.resolve()), + "model_json": str((output_dir / "wav_transcript.json").resolve()), + "scene_type": "1QnA" if case.scene_type == "1qna" else case.scene_type, + } diff --git a/vllm_omni/benchmarks/metrics/metrics.py b/vllm_omni/benchmarks/metrics/metrics.py index 824b810f3ef..32716c7afa7 100644 --- a/vllm_omni/benchmarks/metrics/metrics.py +++ b/vllm_omni/benchmarks/metrics/metrics.py @@ -1,5 +1,9 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + import warnings from collections import defaultdict +from collections.abc import Sequence from dataclasses import field, make_dataclass import numpy as np @@ -15,6 +19,11 @@ _INT_LIST_TYPE = list[int] _MULTIMODAL_BENCHMARK_FIELDS = [ + ("num_ttft_samples", int, field(default=0)), + ("num_tpot_samples", int, field(default=0)), + ("num_itl_samples", int, field(default=0)), + ("num_audio_ttfp_samples", int, field(default=0)), + ("num_audio_rtf_samples", int, field(default=0)), (defs.MEAN_AUDIO_TTFP_MS, float, field(default=0.0)), (defs.MEDIAN_AUDIO_TTFP_MS, float, field(default=0.0)), (defs.STD_AUDIO_TTFP_MS, float, field(default=0.0)), @@ -176,6 +185,17 @@ def _stage_modality_flags( ) +def has_metric_samples(metrics: object, metric_name: str) -> bool: + sample_count_attr = { + "ttft": "num_ttft_samples", + "tpot": "num_tpot_samples", + "itl": "num_itl_samples", + defs.AUDIO_TTFP: "num_audio_ttfp_samples", + defs.AUDIO_RTF: "num_audio_rtf_samples", + }.get(metric_name) + return sample_count_attr is None or getattr(metrics, sample_count_attr, 0) > 0 + + def print_metrics( task_type, selected_percentile_metrics, @@ -308,6 +328,9 @@ def process_one_metric( metric_attribute_name: str, metrics: MultiModalsBenchmarkMetrics, ): + if not has_metric_samples(metrics, metric_attribute_name): + return + metric_header_map = { "ttft": "Time to First Token", "tpot": "Time per Output Token (excl. 1st token)", @@ -718,10 +741,12 @@ def calculate_metrics( good_completed = 0 itls: list[float] = [] tpots: list[float] = [] - all_tpots: list[float] = [] + all_tpots: list[float | None] = [] ttfts: list[float] = [] + goodput_ttfts: list[float | None] = [] e2els: list[float] = [] audio_ttfps: list[float] = [] + goodput_audio_ttfps: list[float | None] = [] audio_rtfs: list[float] = [] audio_duration: list[float] = [] audio_frames: list[int] = [] @@ -748,7 +773,7 @@ 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 = 0 + tpot: float | None = 0.0 if output_len > 1: if outputs[i].itl: # Use mean(ITL) directly so per-request TPOT == mean(ITL). @@ -756,19 +781,33 @@ def calculate_metrics( # 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) - else: + 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) - tpots.append(tpot) + else: + tpot = None + if tpot is not None: + tpots.append(tpot) # Note: if output_len <= 1, we regard tpot as 0 for goodput all_tpots.append(tpot) itls += outputs[i].itl - ttfts.append(outputs[i].ttft) - audio_ttfps.append(getattr(outputs[i], defs.AUDIO_TTFP, 0.0)) - audio_rtfs.append(getattr(outputs[i], defs.AUDIO_RTF, 0.0)) + session_metrics = getattr(outputs[i], "duplex_session_metrics", None) + ttft_measured = not isinstance(session_metrics, dict) or session_metrics.get("mean_ttft_ms") is not None + audio_ttfp_measured = ( + not isinstance(session_metrics, dict) or session_metrics.get("mean_ttfp_ms") is not None + ) + audio_rtf_measured = not isinstance(session_metrics, dict) or session_metrics.get("mean_rtf") is not None + if ttft_measured: + ttfts.append(outputs[i].ttft) + if audio_ttfp_measured: + audio_ttfps.append(getattr(outputs[i], defs.AUDIO_TTFP, 0.0)) + if audio_rtf_measured: + audio_rtfs.append(getattr(outputs[i], defs.AUDIO_RTF, 0.0)) + goodput_ttfts.append(outputs[i].ttft if ttft_measured else None) + goodput_audio_ttfps.append(getattr(outputs[i], defs.AUDIO_TTFP, 0.0) if audio_ttfp_measured else None) audio_duration.append(getattr(outputs[i], defs.AUDIO_DURATION, 0.0)) audio_frames.append(getattr(outputs[i], defs.AUDIO_FRAMES, 0.0)) image_count = int(getattr(outputs[i], defs.IMAGE_COUNT, 0) or 0) @@ -789,14 +828,14 @@ def calculate_metrics( actual_output_lens.append(0) if goodput_config_dict: - valid_metrics = [] + valid_metrics: list[Sequence[float | None]] = [] slo_values = [] if "ttft" in goodput_config_dict: - valid_metrics.append(ttfts) + valid_metrics.append(goodput_ttfts) slo_values.append(goodput_config_dict["ttft"] / MILLISECONDS_TO_SECONDS_CONVERSION) if "audio_ttft" in goodput_config_dict: - valid_metrics.append(audio_ttfps) + valid_metrics.append(goodput_audio_ttfps) slo_values.append(goodput_config_dict["audio_ttft"] / MILLISECONDS_TO_SECONDS_CONVERSION) if "tpot" in goodput_config_dict: valid_metrics.append(all_tpots) @@ -806,7 +845,7 @@ def calculate_metrics( slo_values.append(goodput_config_dict["e2el"] / MILLISECONDS_TO_SECONDS_CONVERSION) for req_metric in zip(*valid_metrics): - is_good_req = all([s >= r for s, r in zip(slo_values, req_metric)]) + is_good_req = all(r is not None and s >= r for s, r in zip(slo_values, req_metric)) if is_good_req: good_completed += 1 @@ -822,6 +861,7 @@ def calculate_metrics( # Calculate max output tokens per second metric max_output_tokens_per_s = 0.0 max_concurrent_requests = 0 + token_timeline_available = True # Find the time range across all successful requests successful_outputs = [output for output in outputs if output.success] @@ -836,19 +876,26 @@ def calculate_metrics( concurrent_requests_per_second = np.zeros(duration_seconds) for i, output in enumerate(successful_outputs): - # Calculate token generation timestamp using - # start_time, ttft, and itl - token_times = [output.start_time + output.ttft] - current_time = token_times[0] - for itl_value in output.itl: - current_time += itl_value - token_times.append(current_time) - - # Add tokens to second buckets - for token_time in token_times: - second_bucket = int(token_time - min_start_time) - if 0 <= second_bucket < duration_seconds: - tokens_per_second[second_bucket] += 1 + session_metrics = getattr(output, "duplex_session_metrics", None) + if isinstance(session_metrics, dict): + # Duplex TTFT/ITL samples are response-local. Without each + # response.created offset they cannot form a session-global + # token timeline, so do not publish a misleading peak rate. + token_timeline_available = False + else: + # Calculate token generation timestamp using + # start_time, ttft, and itl + token_times = [output.start_time + output.ttft] + current_time = token_times[0] + for itl_value in output.itl: + current_time += itl_value + token_times.append(current_time) + + # Add tokens to second buckets + for token_time in token_times: + second_bucket = int(token_time - min_start_time) + if 0 <= second_bucket < duration_seconds: + tokens_per_second[second_bucket] += 1 # Track concurrent requests for each second this request was active request_start_second = int(output.start_time - min_start_time) @@ -860,18 +907,22 @@ def calculate_metrics( # Find the maximum tokens per second and corresponding # concurrent requests if len(tokens_per_second) > 0: - max_output_tokens_per_s = float(np.max(tokens_per_second)) + if token_timeline_available: + max_output_tokens_per_s = float(np.max(tokens_per_second)) + else: + max_output_tokens_per_s = float("nan") max_concurrent_requests = int(np.max(concurrent_requests_per_second)) if TERM_PLOTLIB_AVAILABLE: import termplotlib as tpl fig = tpl.figure() - fig.plot( - np.arange(len(tokens_per_second)), - tokens_per_second, - title="Output tokens per second", - ) + if token_timeline_available: + fig.plot( + np.arange(len(tokens_per_second)), + tokens_per_second, + title="Output tokens per second", + ) fig.plot( np.arange(len(concurrent_requests_per_second)), concurrent_requests_per_second, @@ -881,6 +932,10 @@ def calculate_metrics( else: print("tip: install termplotlib and gnuplot to plot the metrics") + duplex_metrics_present = any( + isinstance(getattr(output, "duplex_session_metrics", None), dict) for output in outputs + ) + missing_duplex_value = float("nan") if duplex_metrics_present else 0 metrics = MultiModalsBenchmarkMetrics( completed=completed, failed=len(failed_outputs), @@ -890,18 +945,18 @@ def calculate_metrics( request_goodput=good_completed / dur_s, output_throughput=sum(actual_output_lens) / dur_s, total_token_throughput=(total_input + sum(actual_output_lens)) / dur_s, - mean_ttft_ms=np.mean(ttfts or 0) * 1000, # ttfts is empty if streaming is not supported by the endpoint - std_ttft_ms=np.std(ttfts or 0) * 1000, - median_ttft_ms=np.median(ttfts or 0) * 1000, - percentiles_ttft_ms=[(p, np.percentile(ttfts or 0, p) * 1000) for p in selected_percentiles], - mean_tpot_ms=np.mean(tpots or 0) * 1000, - std_tpot_ms=np.std(tpots or 0) * 1000, - median_tpot_ms=np.median(tpots or 0) * 1000, - percentiles_tpot_ms=[(p, np.percentile(tpots or 0, p) * 1000) for p in selected_percentiles], - mean_itl_ms=np.mean(itls or 0) * 1000, - std_itl_ms=np.std(itls or 0) * 1000, - median_itl_ms=np.median(itls or 0) * 1000, - percentiles_itl_ms=[(p, np.percentile(itls or 0, p) * 1000) for p in selected_percentiles], + mean_ttft_ms=np.mean(ttfts or missing_duplex_value) * 1000, + 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_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, + percentiles_itl_ms=[(p, np.percentile(itls or missing_duplex_value, p) * 1000) for p in selected_percentiles], mean_e2el_ms=np.mean(e2els or 0) * 1000, std_e2el_ms=np.std(e2els or 0) * 1000, median_e2el_ms=np.median(e2els or 0) * 1000, @@ -910,11 +965,16 @@ def calculate_metrics( max_concurrent_requests=max_concurrent_requests, rtfx=input_audio_duration / dur_s, **{ - defs.MEAN_AUDIO_TTFP_MS: np.mean(audio_ttfps or 0) * 1000, - defs.STD_AUDIO_TTFP_MS: np.std(audio_ttfps or 0) * 1000, - defs.MEDIAN_AUDIO_TTFP_MS: np.median(audio_ttfps or 0) * 1000, + "num_ttft_samples": len(ttfts), + "num_tpot_samples": len(tpots), + "num_itl_samples": len(itls), + "num_audio_ttfp_samples": len(audio_ttfps), + "num_audio_rtf_samples": len(audio_rtfs), + defs.MEAN_AUDIO_TTFP_MS: np.mean(audio_ttfps or missing_duplex_value) * 1000, + defs.STD_AUDIO_TTFP_MS: np.std(audio_ttfps or missing_duplex_value) * 1000, + defs.MEDIAN_AUDIO_TTFP_MS: np.median(audio_ttfps or missing_duplex_value) * 1000, defs.PERCENTILES_AUDIO_TTFP_MS: [ - (p, np.percentile(audio_ttfps or 0, p) * 1000) for p in selected_percentiles + (p, np.percentile(audio_ttfps or missing_duplex_value, p) * 1000) for p in selected_percentiles ], defs.MEAN_AUDIO_DURATION_S: np.mean(audio_duration or 0), defs.STD_AUDIO_DURATION_S: np.std(audio_duration or 0), @@ -925,10 +985,12 @@ def calculate_metrics( defs.TOTAL_AUDIO_DURATION_S: sum(audio_duration), defs.TOTAL_AUDIO_FRAMES: sum(audio_frames), defs.AUDIO_THROUGHPUT: sum(audio_duration) / dur_s, - defs.MEAN_AUDIO_RTF: np.mean(audio_rtfs or 0), - defs.STD_AUDIO_RTF: np.std(audio_rtfs or 0), - defs.MEDIAN_AUDIO_RTF: np.median(audio_rtfs or 0), - defs.PERCENTILES_AUDIO_RTF: [(p, np.percentile(audio_rtfs or 0, p)) for p in selected_percentiles], + defs.MEAN_AUDIO_RTF: np.mean(audio_rtfs or missing_duplex_value), + defs.STD_AUDIO_RTF: np.std(audio_rtfs or missing_duplex_value), + defs.MEDIAN_AUDIO_RTF: np.median(audio_rtfs or missing_duplex_value), + defs.PERCENTILES_AUDIO_RTF: [ + (p, np.percentile(audio_rtfs or missing_duplex_value, p)) for p in selected_percentiles + ], defs.TOTAL_IMAGES: total_images, defs.IMAGE_THROUGHPUT: total_images / dur_s, defs.AVERAGE_PIXELS_PER_IMAGE: (total_image_pixels / total_images) if total_images > 0 else 0.0, diff --git a/vllm_omni/benchmarks/omniinteract.py b/vllm_omni/benchmarks/omniinteract.py new file mode 100644 index 00000000000..46822b486b2 --- /dev/null +++ b/vllm_omni/benchmarks/omniinteract.py @@ -0,0 +1,933 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""OmniInteract session execution and artifacts for serving benchmarks.""" + +from __future__ import annotations + +import asyncio +import binascii +import contextlib +import copy +import fcntl +import hashlib +import json +import logging +import math +import os +import subprocess +import tempfile +import time +from collections import Counter +from collections.abc import Callable, Iterator, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, cast +from urllib.parse import urljoin, urlsplit + +import pybase64 as base64 + +from vllm_omni.benchmarks.data_modules.omniinteract_dataset import ( + DEFAULT_MAX_VIDEO_DURATION_S, + OmniInteractCase, + OmniInteractPreparedInput, + case_manifest, +) +from vllm_omni.experimental.fullduplex.client import ( + PCM16_BYTES_PER_SAMPLE, + PCM16_SAMPLE_RATE, + RealtimeDuplexClient, + RealtimeEventCollector, + build_realtime_url, + chunk_period_ms, + has_residual_model_unit, + reference_audio_data_url, + summarize_session_request_metrics, + write_pcm16_wav, +) + +OUTPUT_SAMPLE_RATE = 24_000 +SUCCESS_ARTIFACTS = (".done", "output.wav", "wav_transcript.json", "events.json", "result.json") +BATCH_ARTIFACTS = ("batch_summary.json", "official_eval_manifest.jsonl") +ARTIFACT_LOCK_FILE = ".omniinteract.lock" +_INPUT_CHUNK_MS = 200 +VIDEO_FPS = 1.0 +_COMPLETION_SETTLE_S = 2.0 +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class OmniInteractBenchmarkConfig: + model: str + base_url: str = "http://127.0.0.1:8000" + endpoint: str = "/v1/realtime" + output_root: Path = Path("omniinteract-output") + timeout_s: float = 900.0 + media_timeout_s: float = 600.0 + max_video_duration_s: float = DEFAULT_MAX_VIDEO_DURATION_S + ref_audio: str | None = None + require_response: bool = False + extra_headers: dict[str, str] | None = None + extra_body: dict[str, object] | None = None + + +@dataclass +class OmniInteractCaseResult: + subset: str + video: str + output_dir: str + success: bool = False + error: str = "" + session_id: str = "" + latency_s: float = 0.0 + input_audio_chunks: int = 0 + input_video_frames: int = 0 + pacing_mean_lag_s: float = 0.0 + pacing_max_lag_s: float = 0.0 + responses: int = 0 + audio_bytes: int = 0 + audio_clipped_bytes: int = 0 + transcript: str = "" + eligible_for_official_eval: bool = False + official_eval_ineligible_reasons: list[str] = field(default_factory=list) + artifact_warnings: list[str] = field(default_factory=list) + output_tokens: int = 0 + duplex_request_metrics: list[dict[str, object]] = field(default_factory=list) + duplex_session_metrics: dict[str, object] = field(default_factory=dict) + _artifact_context: _DeferredArtifactContext | None = field(default=None, repr=False, compare=False) + + def as_dict(self) -> dict[str, Any]: + return copy.deepcopy({key: value for key, value in vars(self).items() if not key.startswith("_")}) + + +@dataclass(frozen=True) +class _DeferredArtifactContext: + horizon_bytes: int + spans: tuple[tuple[int, bytes], ...] + chunks: list[dict[str, object]] + events: list[dict[str, object]] + + +def benchmark_summary(results: list[OmniInteractCaseResult]) -> dict[str, Any]: + succeeded = sum(result.success for result in results) + eligible = sum(result.success and result.eligible_for_official_eval for result in results) + return { + "total": len(results), + "success": succeeded, + "failed": len(results) - succeeded, + "eligible_for_official_eval": eligible, + "successful_but_ineligible": succeeded - eligible, + "audio_clipped_bytes": sum(result.audio_clipped_bytes for result in results), + "results": [result.as_dict() for result in results], + } + + +def _run_media_command( + command: list[str], + *, + timeout_s: float, + text: bool = False, +) -> subprocess.CompletedProcess[Any]: + try: + result = subprocess.run( + command, + capture_output=True, + check=False, + text=text, + timeout=timeout_s, + ) + except subprocess.TimeoutExpired as exc: + raise TimeoutError(f"Media command timed out after {timeout_s:g}s: {command[0]}") from exc + except FileNotFoundError as exc: + raise RuntimeError(f"Required media command is unavailable: {command[0]}") from exc + if result.returncode: + error = result.stderr if text else result.stderr.decode("utf-8", "ignore") + raise RuntimeError(f"{command[0]} failed: {error.strip()}") + return result + + +def prepare_media( + video: Path, + fps: float, + *, + timeout_s: float, + max_duration_s: float = DEFAULT_MAX_VIDEO_DURATION_S, +) -> tuple[float, bytes, list[str | None]]: + probe = [ + "ffprobe", + *"-v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1".split(), + str(video), + ] + duration = float(_run_media_command(probe, text=True, timeout_s=timeout_s).stdout.strip()) + if not math.isfinite(duration) or duration <= 0 or duration > max_duration_s: + raise ValueError(f"Invalid video duration for {video}: {duration!r}") + target_bytes = math.ceil(duration) * PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE + audio = [ + "ffmpeg", + *"-loglevel error -i".split(), + str(video), + "-t", + str(duration), + *f"-vn -f s16le -ac 1 -ar {PCM16_SAMPLE_RATE} -fs".split(), + str(target_bytes), + "pipe:1", + ] + pcm = _run_media_command(audio, timeout_s=timeout_s).stdout + pcm = (pcm + bytes(max(0, target_bytes - len(pcm))))[:target_bytes] + + frame_count = math.ceil(duration * fps) + frames: list[str | None] = [None] * frame_count + with tempfile.TemporaryDirectory(prefix="vllm-omni-frames-") as temp_dir: + output_pattern = Path(temp_dir) / "frame-%06d.jpg" + frame_filter = f"select=gte(t\\,(selected_n+0.5)/{fps}),scale=640:640:force_original_aspect_ratio=decrease" + frame_command = [ + "ffmpeg", + *"-loglevel error -i".split(), + str(video), + "-an", + "-vf", + frame_filter, + *f"-frames:v {frame_count} -vsync vfr -q:v 5".split(), + str(output_pattern), + ] + _run_media_command(frame_command, timeout_s=timeout_s) + for index, frame_path in enumerate(sorted(Path(temp_dir).glob("frame-*.jpg"))[:frame_count]): + frames[index] = base64.b64encode(frame_path.read_bytes()).decode("ascii") + return duration, pcm, frames + + +@dataclass(frozen=True) +class _AudioSegment: + event_index: int + response_id: str + start_s: float + end_s: float + pcm16: bytes + + +class _Playback: + def __init__(self) -> None: + self.cursor = 0 + self.end_s = 0.0 + self.segments: list[_AudioSegment] = [] + self.completed: set[str] = set() + self.completion_acked: set[str] = set() + self.warnings: list[str] = [] + self._total_samples: dict[str, int] = {} + self._response_end_s: dict[str, float] = {} + + def _warn_once(self, warning: str) -> None: + if warning not in self.warnings: + self.warnings.append(warning) + + def ingest(self, events: RealtimeEventCollector) -> None: + while self.cursor < len(events.events): + index, self.cursor = self.cursor, self.cursor + 1 + event = events.events[index] + response_id = events.response_id(event) + if event.get("type") == "response.done": + if response_id: + self.completed.add(response_id) + continue + if event.get("type") != "response.audio.delta": + continue + encoded = event.get("delta") or event.get("audio") + if not response_id: + raise ValueError("response audio has no response_id") + if not isinstance(encoded, str): + raise ValueError("response audio payload is missing") + if event.get("format") is None: + self._warn_once("response.audio.delta omitted format; assumed pcm16") + elif event.get("format") != "pcm16": + raise ValueError("OmniInteract output must be pcm16") + rate = event.get("sample_rate_hz") + if rate is None: + rate = events.output_sample_rate_hz or OUTPUT_SAMPLE_RATE + self._warn_once(f"response.audio.delta omitted sample_rate_hz; assumed {rate}") + if rate != OUTPUT_SAMPLE_RATE: + raise ValueError(f"OmniInteract output must use {OUTPUT_SAMPLE_RATE} Hz audio") + try: + raw = base64.b64decode(encoded, validate=True) + except (ValueError, binascii.Error) as exc: + raise ValueError("response audio is not valid base64") from exc + if not raw or len(raw) % PCM16_BYTES_PER_SAMPLE: + raise ValueError("response audio is empty or not PCM16 aligned") + start = max(events.event_received_at_s[index], self.end_s) + samples = len(raw) // PCM16_BYTES_PER_SAMPLE + segment = _AudioSegment(index, response_id, start, start + samples / rate, raw) + self.segments.append(segment) + self.end_s = segment.end_s + self._total_samples[response_id] = self._total_samples.get(response_id, 0) + samples + self._response_end_s[response_id] = segment.end_s + + async def acknowledge(self, client: RealtimeDuplexClient, now: float | None = None) -> None: + events = client.events + self.ingest(events) + now = time.monotonic() if now is None else now + for response_id in self.completed - self.completion_acked: + total_samples = self._total_samples.get(response_id, 0) + if not total_samples or now < self._response_end_s[response_id]: + continue + played_ms = total_samples * 1000 // OUTPUT_SAMPLE_RATE + await client.send_playback_ack(response_id, played_ms) + self.completion_acked.add(response_id) + + +async def stream_inputs( + client: RealtimeDuplexClient, + pcm: bytes, + frames: Sequence[str | None], + playback: _Playback, +) -> tuple[int, int, float, float]: + bytes_per_second = PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE + chunk_bytes = bytes_per_second * _INPUT_CHUNK_MS // 1000 + started_at = time.monotonic() + frame_cursor = sent_frames = 0 + lags: list[float] = [] + for chunks, offset in enumerate(range(0, len(pcm), chunk_bytes), start=1): + end = min(offset + chunk_bytes, len(pcm)) + end_ms = end * 1000 // bytes_per_second + ready: list[str] = [] + while frame_cursor < len(frames) and end_ms >= (frame_cursor + 0.5) * 1000 / VIDEO_FPS: + if frames[frame_cursor]: + ready.append(frames[frame_cursor] or "") + frame_cursor += 1 + sent_frames += len(ready) + event: dict[str, object] = { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(pcm[offset:end]).decode("ascii"), + "input_audio_format": "pcm16", + "sample_rate_hz": PCM16_SAMPLE_RATE, + "duration_ms": (end - offset) * 1000 // bytes_per_second, + "audio_end_ms": end_ms, + } + if ready: + event["video_frames"] = ready + lags.append(max(0.0, time.monotonic() - started_at - offset / bytes_per_second)) + await client.send(event) + await playback.acknowledge(client) + await asyncio.sleep(max(0.0, started_at + end_ms / 1000 - time.monotonic())) + return chunks if pcm else 0, sent_frames, sum(lags) / len(lags) if lags else 0.0, max(lags, default=0.0) + + +def _response_status(event: dict[str, object]) -> str | None: + response = event.get("response") + if isinstance(response, dict) and response.get("status"): + return str(response["status"]) + status = event.get("status") + return str(status) if status else None + + +def response_ledger( + collector: RealtimeEventCollector, + *, + end: int | None = None, +) -> tuple[set[str], set[str]]: + created: set[str] = set() + done: set[str] = set() + for event in collector.events[:end]: + event_type = event.get("type") + if event_type not in {"response.created", "response.done"}: + continue + response_id = collector.response_id(event) + if not response_id: + raise ValueError(f"{event_type} has no response_id") + if event_type == "response.created": + if response_id in created: + raise ValueError(f"duplicate response.created for {response_id}") + created.add(response_id) + continue + if response_id not in created: + raise ValueError(f"response.done without response.created for {response_id}") + if response_id in done: + raise ValueError(f"duplicate response.done for {response_id}") + response = event.get("response") if isinstance(event.get("response"), dict) else {} + details = event.get("status_details") if isinstance(event.get("status_details"), dict) else {} + nested = response.get("status_details") if isinstance(response.get("status_details"), dict) else {} + if "failed" in {event.get("status"), response.get("status"), details.get("type"), nested.get("type")}: + raise ValueError(f"response.done reports failure for {response_id}") + done.add(response_id) + return created, done + + +def _is_model_listen_decision(event: dict[str, object]) -> bool: + response = event.get("response") + metadata = response.get("metadata") if isinstance(response, dict) else event.get("metadata") + return isinstance(metadata, dict) and metadata.get("buffering") is not True and metadata.get("model_listen") is True + + +def _has_post_commit_decision( + collector: RealtimeEventCollector, + events: list[dict[str, object]], + *, + prior_response_ids: set[str], +) -> bool: + pending_prior = set(prior_response_ids) + for event in events: + event_type = event.get("type") + if event_type == "response.done": + response_id = collector.response_id(event) + if response_id in pending_prior: + pending_prior.remove(response_id) + continue + if not pending_prior and _response_status(event) != "cancelled": + return True + elif event_type == "response.listen" and not pending_prior and _is_model_listen_decision(event): + return True + return False + + +def _raise_if_session_terminated( + collector: RealtimeEventCollector, + from_index: int, + *, + explicit_close_from: int | None = None, +) -> None: + errors = collector.errors() + if errors: + raise RuntimeError(str(errors[-1])) + for index, event in enumerate(collector.events[from_index:], start=from_index): + event_type = event.get("type") + if event_type not in {"session.expired", "session.closed"}: + continue + reason = event.get("reason") + nested = event.get("event") + if reason is None and isinstance(nested, dict): + reason = nested.get("reason") + expected = ( + event_type == "session.closed" + and explicit_close_from is not None + and index >= explicit_close_from + and reason is None + ) + if expected: + continue + detail = f": {reason}" if reason else "" + prefix = "Unexpected " if explicit_close_from is not None else "" + raise RuntimeError(f"{prefix}{event_type}{detail}") + + +def _ensure_final_commit_tail(pcm: bytes, events: list[dict[str, object]]) -> bytes: + """Reserve one PCM sample so an exact model unit is flushed by commit. + + A complete unit is otherwise emitted before commit, so its asynchronous + decision cannot be correlated with the accepted final input. Keeping an + almost-full unit buffered makes the server's final residual flush carry + that correlation; the server pads the missing sample back to a full unit. + """ + + period_ms = chunk_period_ms(events) + if len(pcm) >= PCM16_BYTES_PER_SAMPLE and not has_residual_model_unit(pcm, chunk_period_ms=period_ms): + return pcm[:-PCM16_BYTES_PER_SAMPLE] + return pcm + + +async def wait_for_session_completion( + client: RealtimeDuplexClient, + playback: _Playback, + *, + commit_from: int, + session_from: int | None = None, + timeout_s: float, + settle_s: float, +) -> int: + deadline = time.monotonic() + timeout_s + committed_index: int | None = None + prior_response_ids: set[str] = set() + last_event_count = len(client.events.events) + stable_since = time.monotonic() + while time.monotonic() < deadline: + client.raise_if_reader_stopped() + _raise_if_session_terminated(client.events, commit_from if session_from is None else session_from) + await playback.acknowledge(client) + if len(client.events.events) != last_event_count: + last_event_count = len(client.events.events) + stable_since = time.monotonic() + if committed_index is None: + committed_index = next( + ( + index + for index in range(commit_from, len(client.events.events)) + if client.events.events[index].get("type") == "input_audio_buffer.committed" + ), + None, + ) + if committed_index is not None: + committed_event = client.events.events[committed_index] + committed = committed_event.get("event") + if isinstance(committed, dict) and committed.get("overlap_deferred") is True: + prior_created, prior_done = response_ledger(client.events, end=committed_index) + prior_response_ids = prior_created - prior_done + stable_since = time.monotonic() + if committed_index is not None: + created, done = response_ledger(client.events) + post_commit = client.events.events[committed_index + 1 :] + decision = _has_post_commit_decision( + client.events, + post_commit, + prior_response_ids=prior_response_ids, + ) + if created == done and decision and time.monotonic() - stable_since >= settle_s: + return committed_index + await asyncio.sleep(0.05) + missing: set[str] = set() + with contextlib.suppress(ValueError): + created, done = response_ledger(client.events) + missing = created - done + raise TimeoutError( + "Timed out waiting for committed input and stable responses" + + (f"; unfinished response_ids={sorted(missing)}" if missing else "") + ) + + +def _output_dir(root: Path, case: OmniInteractCase) -> Path: + relative = case.video_rel.replace("\\", "/") + stem = Path(relative).with_suffix("").as_posix().replace("/", "__") + digest = hashlib.sha256(f"{case.subset}/{relative}".encode()).hexdigest()[:8] + return root / case.subset / f"{stem}--{digest}" + + +@contextlib.contextmanager +def omniinteract_output_lock(root: Path) -> Iterator[None]: + root.mkdir(parents=True, exist_ok=True) + with (root / ARTIFACT_LOCK_FILE).open("a+b") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _atomic_replace(path: Path, writer: Callable[[Path], None]) -> None: + descriptor, temporary_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) + os.close(descriptor) + temporary = Path(temporary_name) + try: + writer(temporary) + temporary.replace(path) + finally: + temporary.unlink(missing_ok=True) + + +def _atomic_write_text(path: Path, value: str) -> None: + _atomic_replace(path, lambda temporary: temporary.write_text(value, encoding="utf-8")) + + +def _atomic_write_json(path: Path, value: object) -> None: + _atomic_write_text(path, json.dumps(value, ensure_ascii=False, indent=2) + "\n") + + +def _atomic_write_wav(path: Path, pcm: bytes | bytearray, rate: int) -> None: + _atomic_replace( + path, + lambda temporary: write_pcm16_wav( + temporary, + cast(bytes, pcm), # wave.writeframes accepts bytearray. + sample_rate_hz=rate, + ), + ) + + +def _clear_artifacts(directory: Path, names: Sequence[str]) -> None: + directory.mkdir(parents=True, exist_ok=True) + for name in names: + (directory / name).unlink(missing_ok=True) + + +def _artifact_summary(case: OmniInteractCase, result: OmniInteractCaseResult) -> dict[str, Any]: + return { + **result.as_dict(), + "annotation": str(case.annotation_path), + "scene_type": "1QnA" if case.scene_type == "1qna" else case.scene_type, + } + + +def _publish_success_artifacts( + directory: Path, + result: OmniInteractCaseResult, + context: _DeferredArtifactContext, + summary: dict[str, Any], +) -> None: + pcm = bytearray(context.horizon_bytes) + for offset, chunk in context.spans: + pcm[offset : offset + len(chunk)] = chunk + try: + _clear_artifacts(directory, (*SUCCESS_ARTIFACTS, ".failed.json")) + _atomic_write_wav(directory / "output.wav", pcm, OUTPUT_SAMPLE_RATE) + _atomic_write_json( + directory / "wav_transcript.json", + { + "text": result.transcript, + "chunks": context.chunks, + "timestamp_semantics": "serialized playback queue time relative to input streaming start", + }, + ) + _atomic_write_json(directory / "events.json", context.events) + _atomic_write_json(directory / "result.json", result.as_dict()) + _atomic_write_json(directory / ".done", summary) + except Exception: + _clear_artifacts(directory, SUCCESS_ARTIFACTS) + result.success = result.eligible_for_official_eval = False + if "artifact_write_failed" not in result.official_eval_ineligible_reasons: + result.official_eval_ineligible_reasons.append("artifact_write_failed") + raise + + +def _collect_output( + collector: RealtimeEventCollector, + playback: _Playback, + *, + stream_start: float, + video_duration_s: float, + require_response: bool, + retain_events: bool, + result: OmniInteractCaseResult, +) -> _DeferredArtifactContext: + playback.ingest(collector) + created, done = response_ledger(collector) + if created != done: + raise ValueError(f"unfinished response_ids: {sorted(created - done)}") + done_events = [event for event in collector.events if event.get("type") == "response.done"] + cancelled = any(_response_status(event) == "cancelled" for event in done_events) + terminal_responses = { + response_id + for event in done_events + if _response_status(event) != "cancelled" and (response_id := collector.response_id(event)) + } + horizon_s = math.ceil(video_duration_s) + horizon_bytes = horizon_s * OUTPUT_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE + spans: list[tuple[int, bytes]] = [] + clipped_bytes = 0 + response_times: dict[str, list[float]] = {} + responses_with_audio: set[str] = set() + for segment in playback.segments: + response_id = segment.response_id + if response_id not in created: + raise ValueError("response audio has no matching response.created") + start_s = max(0.0, segment.start_s - stream_start) + end_s = max(start_s, segment.end_s - stream_start) + offset = round(start_s * OUTPUT_SAMPLE_RATE) * PCM16_BYTES_PER_SAMPLE + writable = max(0, min(len(segment.pcm16), horizon_bytes - offset)) + clipped_bytes += len(segment.pcm16) - writable + if writable: + clipped = segment.pcm16[:writable] + spans.append((offset, clipped)) + if any(clipped): + responses_with_audio.add(response_id) + timing = response_times.setdefault(response_id, [start_s, end_s]) + timing[0], timing[1] = min(timing[0], start_s), max(timing[1], end_s) + + text_event_types = {"response.audio_transcript.delta", "response.output_text.delta", "response.text.delta"} + text_ids = {collector.response_id(event) for event in collector.events if event.get("type") in text_event_types} + if None in text_ids or not text_ids <= created: + raise ValueError("response transcript has no matching response.created") + chunks: list[dict[str, object]] = [] + texts: list[str] = [] + for response_id in collector.response_ids: + text = collector.response_text(response_id).strip() + if not text: + continue + response_timing = response_times.get(response_id) + if response_timing is None or response_timing[0] >= horizon_s: + continue + texts.append(text) + chunks.append( + { + "response_id": response_id, + "text": text, + "timestamp": [round(response_timing[0], 6), round(min(response_timing[1], horizon_s), 6)], + } + ) + complete_outputs = terminal_responses & responses_with_audio & {str(chunk["response_id"]) for chunk in chunks} + if require_response and not complete_outputs: + raise ValueError("OmniInteract E2E requires a response with audio and transcript") + result.audio_bytes, result.audio_clipped_bytes = ( + sum(len(segment.pcm16) for segment in playback.segments), + clipped_bytes, + ) + result.transcript, result.responses, result.success = " ".join(texts).strip(), len(created), True + result.artifact_warnings = list(playback.warnings) + result.official_eval_ineligible_reasons = [] + if clipped_bytes: + result.official_eval_ineligible_reasons.append("audio_clipped") + if cancelled: + result.official_eval_ineligible_reasons.append("cancelled_response") + result.eligible_for_official_eval = not result.official_eval_ineligible_reasons + audio_sizes = {segment.event_index: len(segment.pcm16) for segment in playback.segments} + events = ( + [ + { + **{key: value for key, value in event.items() if key not in {"delta", "audio"}}, + "audio_bytes": audio_sizes[index], + } + if event.get("type") == "response.audio.delta" + else dict(event) + for index, event in enumerate(collector.events) + ] + if retain_events + else [] + ) + return _DeferredArtifactContext(horizon_bytes, tuple(spans), chunks, events) + + +def prepare_success_artifacts( + root: Path, + case: OmniInteractCase, + collector: RealtimeEventCollector, + *, + playback: _Playback | None = None, + stream_start: float, + video_duration_s: float, + require_response: bool, + result: OmniInteractCaseResult, + capture_artifacts: bool = True, +) -> dict[str, Any]: + directory = _output_dir(root, case) + context = _collect_output( + collector, + playback or _Playback(), + stream_start=stream_start, + video_duration_s=video_duration_s, + require_response=require_response, + retain_events=capture_artifacts, + result=result, + ) + result.output_dir = str(directory.resolve()) + summary = _artifact_summary(case, result) + if capture_artifacts: + result._artifact_context = context + return summary + + +def write_failure_artifacts(root: Path, case: OmniInteractCase, result: OmniInteractCaseResult) -> None: + directory = _output_dir(root, case) + result.output_dir = str(directory.resolve()) + result.success = result.eligible_for_official_eval = False + _clear_artifacts(directory, SUCCESS_ARTIFACTS) + _atomic_write_json(directory / ".failed.json", _artifact_summary(case, result)) + + +def publish_deferred_case_artifacts( + root: Path, + case: OmniInteractCase, + result: OmniInteractCaseResult, +) -> None: + context = result._artifact_context + result._artifact_context = None + if not result.success: + write_failure_artifacts(root, case, result) + return + if context is None: + raise RuntimeError("OmniInteract benchmark output lost its deferred artifact context") + directory = _output_dir(root, case) + _publish_success_artifacts(directory, result, context, _artifact_summary(case, result)) + + +def clear_case_artifacts(root: Path, case: OmniInteractCase) -> None: + directory = _output_dir(root, case) + _clear_artifacts(directory, (*SUCCESS_ARTIFACTS, ".failed.json")) + + +def clear_batch_artifacts(root: Path) -> None: + _clear_artifacts(root, BATCH_ARTIFACTS) + + +def write_batch_artifacts( + root: Path, + cases: list[OmniInteractCase], + results: list[OmniInteractCaseResult], +) -> None: + root.mkdir(parents=True, exist_ok=True) + rows = [ + case_manifest(case, _output_dir(root, case)) + for case, result in zip(cases, results, strict=True) + if result.success and result.eligible_for_official_eval + ] + ineligible = [result for result in results if result.success and not result.eligible_for_official_eval] + if ineligible: + reasons = Counter( + reason for result in ineligible for reason in (result.official_eval_ineligible_reasons or ["unspecified"]) + ) + logger.warning( + "%d successful OmniInteract cases were excluded from official evaluation: %s", + len(ineligible), + ", ".join(f"{reason}={count}" for reason, count in sorted(reasons.items())), + ) + _atomic_write_json(root / "batch_summary.json", benchmark_summary(results)) + _atomic_write_text( + root / "official_eval_manifest.jsonl", + "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), + ) + + +def _websocket_url(config: OmniInteractBenchmarkConfig, session_id: str) -> str: + endpoint = ( + config.endpoint + if urlsplit(config.endpoint).scheme + else urljoin(config.base_url.rstrip("/") + "/", config.endpoint.lstrip("/")) + ) + return build_realtime_url( + endpoint, + config.model, + autostart=False, + native_duplex=True, + session_id=session_id, + ) + + +def _populate_response_metrics( + result: OmniInteractCaseResult, + collector: RealtimeEventCollector, + *, + stream_start: float, +) -> None: + measurement_origin = { + "ttft": "response.created client receive to first non-empty text delta", + "ttfp": "response.created client receive to first audio packet", + "rtf": "response.created client receive to last audio packet divided by emitted audio duration", + } + request_metrics: list[dict[str, object]] = [] + output_tokens = 0 + for request_index, response_id in enumerate(collector.response_ids): + timing = collector.timing_summary( + after_s=stream_start, + input_committed_at_s=None, + response_id=response_id, + measurement_origin=measurement_origin, + ) + raw_metric = timing.get("request_metrics") + stage0 = timing.get("stage0_tokens") + metric = { + "session_id": result.session_id, + "request_index": request_index, + "response_id": response_id, + **(raw_metric if isinstance(raw_metric, dict) else {}), + } + if isinstance(stage0, dict): + metric["stage0_tokens"] = dict(stage0) + output_tokens += int(stage0.get("output_token_count") or 0) + if isinstance(raw_metric, dict) or isinstance(stage0, dict): + request_metrics.append(metric) + result.output_tokens = output_tokens + result.duplex_request_metrics = request_metrics + result.duplex_session_metrics = summarize_session_request_metrics( + request_metrics, + session_id=result.session_id, + ) + + +async def run_omniinteract_case( + case: OmniInteractCase, + config: OmniInteractBenchmarkConfig, + *, + request_index: int | str, + capture_artifacts: bool = True, + prepared_input: OmniInteractPreparedInput | None = None, +) -> OmniInteractCaseResult: + if not config.ref_audio: + raise ValueError("ref_audio is required for MiniCPM-o native-duplex audio output") + for name, value in ( + ("timeout_s", config.timeout_s), + ("media_timeout_s", config.media_timeout_s), + ("max_video_duration_s", config.max_video_duration_s), + ): + if not math.isfinite(value) or value <= 0: + raise ValueError(f"{name} must be finite and positive") + session_id = f"omniinteract:{case.subset}:{request_index}:{time.monotonic_ns()}" + result = OmniInteractCaseResult( + case.subset, + str(case.video_path), + str(_output_dir(config.output_root, case).resolve()), + session_id=session_id, + ) + started_at = time.monotonic() + try: + if prepared_input is None: + reference_audio = reference_audio_data_url(config.ref_audio) + duration, pcm, frames = await asyncio.to_thread( + prepare_media, + case.video_path, + VIDEO_FPS, + timeout_s=config.media_timeout_s, + max_duration_s=config.max_video_duration_s, + ) + else: + reference_audio = prepared_input.ref_audio_data_url + duration = prepared_input.duration_s + pcm = prepared_input.pcm16 + frames = prepared_input.video_frames + if not any(frames): + raise ValueError(f"No video frames were decoded from {case.video_path}") + async with RealtimeDuplexClient( + _websocket_url(config, session_id), additional_headers=config.extra_headers + ) as client: + session_from = len(client.events.events) + await client.configure( + config.model, + ref_audio=reference_audio, + session_id=session_id, + extra_body=config.extra_body, + idle_timeout_s=config.timeout_s, + timeout_s=min(config.timeout_s, 20.0), + ) + pcm = _ensure_final_commit_tail(pcm, client.events.events) + playback = _Playback() + stream_start = time.monotonic() + try: + input_duration_s = len(pcm) / (PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE) + upload_timeout_s = config.timeout_s + input_duration_s + try: + chunks, frame_count, mean_lag, max_lag = await asyncio.wait_for( + stream_inputs(client, pcm, frames, playback), timeout=upload_timeout_s + ) + except asyncio.TimeoutError as exc: + raise TimeoutError(f"Realtime upload timed out after {upload_timeout_s:g}s") from exc + commit_from = len(client.events.events) + await client.commit() + await wait_for_session_completion( + client, + playback, + commit_from=commit_from, + session_from=session_from, + timeout_s=config.timeout_s, + settle_s=_COMPLETION_SETTLE_S, + ) + await playback.acknowledge(client, playback.end_s) + _raise_if_session_terminated(client.events, session_from) + close_from = len(client.events.events) + await client.close_session(timeout_s=min(config.timeout_s, 20.0)) + except Exception: + with contextlib.suppress(Exception): + await client.close_session(timeout_s=min(config.timeout_s, 20.0)) + raise + errors = client.events.errors() + if errors: + raise RuntimeError(str(errors[-1])) + _raise_if_session_terminated(client.events, session_from, explicit_close_from=close_from) + result.latency_s = time.monotonic() - started_at + result.input_audio_chunks, result.input_video_frames = chunks, frame_count + result.pacing_mean_lag_s, result.pacing_max_lag_s = mean_lag, max_lag + _populate_response_metrics(result, client.events, stream_start=stream_start) + await asyncio.to_thread( + prepare_success_artifacts, + config.output_root, + case, + client.events, + playback=playback, + stream_start=stream_start, + video_duration_s=duration, + require_response=config.require_response, + result=result, + capture_artifacts=capture_artifacts, + ) + except Exception as exc: + result.success = result.eligible_for_official_eval = False + if "case_failed" not in result.official_eval_ineligible_reasons: + result.official_eval_ineligible_reasons.append("case_failed") + result.error = str(exc) + result.latency_s = max(0.0, time.monotonic() - started_at) + return result diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index 3f83a1b3814..942a9963b18 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + import asyncio import contextlib import io @@ -16,7 +19,6 @@ from datetime import datetime from pathlib import Path from typing import Any, Literal -from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit import aiohttp import numpy as np @@ -31,6 +33,7 @@ RequestFuncOutput, StreamedResponseHandler, _get_chat_content, + _get_headers, _update_headers_common, _update_payload_common, _validate_api_url, @@ -46,6 +49,13 @@ daily_omni_local_videos_dir, resolve_daily_omni_local_root, ) +from vllm_omni.benchmarks.data_modules.omniinteract_dataset import ( + DEFAULT_OMNIINTERACT_REPO, + OmniInteractDataset, + OmniInteractPreparedInput, + OmniInteractSampleRequest, + OmniInteractSessionOptions, +) from vllm_omni.benchmarks.data_modules.random_multi_modal_dataset import OmniRandomMultiModalDataset from vllm_omni.benchmarks.data_modules.seed_tts_dataset import ( SEED_TTS_DEFAULT_OMNI_SYSTEM_PROMPT, @@ -56,8 +66,27 @@ ) from vllm_omni.benchmarks.data_modules.sound_effect_dataset import SoundEffectDataset from vllm_omni.benchmarks.data_modules.ttsd_dataset import TTSDDataset +from vllm_omni.benchmarks.omniinteract import ( + VIDEO_FPS, + OmniInteractBenchmarkConfig, + OmniInteractCaseResult, + clear_batch_artifacts, + clear_case_artifacts, + prepare_media, + publish_deferred_case_artifacts, + run_omniinteract_case, + write_failure_artifacts, +) +from vllm_omni.benchmarks.omniinteract import ( + benchmark_summary as omniinteract_benchmark_summary, +) +from vllm_omni.benchmarks.omniinteract import ( + write_batch_artifacts as write_omniinteract_batch_artifacts, +) from vllm_omni.experimental.fullduplex.client import ( RealtimeDuplexClient, + build_realtime_url, + reference_audio_data_url, summarize_session_request_metrics, wait_for, ) @@ -221,6 +250,91 @@ def _attach_seed_tts_to_request_func_input(sample: SampleRequest, rfi: RequestFu rfi.extra_body = base +def _attach_omniinteract_to_request_func_input(sample: SampleRequest, rfi: RequestFuncInput) -> None: + if not isinstance(sample, OmniInteractSampleRequest): + return + setattr(rfi, "omniinteract_case", sample.omniinteract_case) + setattr(rfi, "omniinteract_options", sample.omniinteract_options) + setattr(rfi, "omniinteract_prepared_input", sample.omniinteract_prepared_input) + + +def _append_error(existing: str, message: str) -> str: + return f"{existing}\n{message}" if existing else message + + +def _finalize_omniinteract_batch( + input_requests: list[SampleRequest], + outputs: list[RequestFuncOutput], +) -> dict[str, object] | None: + rows = [ + (sample, output) + for sample, output in zip(input_requests, outputs, strict=True) + if isinstance(sample, OmniInteractSampleRequest) + ] + if not rows: + return None + options = rows[0][0].omniinteract_options + if not isinstance(options, OmniInteractSessionOptions): + raise RuntimeError("OmniInteract benchmark output lost its artifact options") + cases, results = [], [] + artifact_errors: list[str] = [] + for sample, output in rows: + case = sample.omniinteract_case + result = getattr(output, "omniinteract_case_result", None) + if case is None or not isinstance(result, OmniInteractCaseResult): + raise RuntimeError("OmniInteract benchmark output lost its dataset identity") + cases.append(case) + results.append(result) + try: + publish_deferred_case_artifacts(options.output_root, case, result) + except Exception as exc: # noqa: BLE001 - artifact backends raise heterogeneous errors + result.success = result.eligible_for_official_eval = False + if "artifact_write_failed" not in result.official_eval_ineligible_reasons: + result.official_eval_ineligible_reasons.append("artifact_write_failed") + artifact_error = f"Artifact publication failed: {exc}" + result.error = _append_error(result.error, artifact_error) + output.error = _append_error(output.error, artifact_error) + message = f"{case.subset}/{case.video_rel}: {artifact_error}" + artifact_errors.append(message) + logger.exception(message) + try: + write_failure_artifacts(options.output_root, case, result) + except Exception as recovery_exc: # noqa: BLE001 - preserve benchmark metrics after I/O failure + message = f"{case.subset}/{case.video_rel}: failure artifact publication failed: {recovery_exc}" + artifact_errors.append(message) + logger.exception(message) + try: + write_omniinteract_batch_artifacts(options.output_root, cases, results) + except Exception as exc: # noqa: BLE001 - preserve benchmark metrics after I/O failure + message = f"Batch artifact publication failed: {exc}" + artifact_errors.append(message) + logger.exception(message) + try: + clear_batch_artifacts(options.output_root) + except Exception as cleanup_exc: # noqa: BLE001 - best-effort removal of partial batch artifacts + message = f"Partial batch artifact cleanup failed: {cleanup_exc}" + artifact_errors.append(message) + logger.exception(message) + summary = omniinteract_benchmark_summary(results) + compact_summary = {key: value for key, value in summary.items() if key != "results"} + compact_summary["artifacts_complete"] = not artifact_errors + if artifact_errors: + compact_summary["artifact_errors"] = artifact_errors + return compact_summary + + +def _prepare_omniinteract_batch(input_requests: list[SampleRequest]) -> None: + roots: set[Path] = set() + for sample in input_requests: + if not isinstance(sample, OmniInteractSampleRequest): + continue + root = sample.omniinteract_options.output_root.resolve() + roots.add(root) + clear_case_artifacts(sample.omniinteract_options.output_root, sample.omniinteract_case) + for root in roots: + clear_batch_artifacts(root) + + def _daily_omni_repo_from_args(args) -> str | None: """Resolve HuggingFace repo id for Daily-Omni from CLI args. @@ -250,6 +364,7 @@ def get_samples(args, tokenizer): "ttsd", "sound-effect", ) + is_omniinteract = args.dataset_name == "omniinteract" # Check if we need to handle omni-related backends/datasets is_omni_backend = args.backend in [ @@ -258,12 +373,68 @@ def get_samples(args, tokenizer): "openai-realtime-duplex", "daily-omni", ] - is_omni_dataset = is_daily_omni or is_seed_tts or args.dataset_name == "random-mm" + is_omni_dataset = is_daily_omni or is_seed_tts or is_omniinteract or args.dataset_name == "random-mm" if not is_omni_backend and not is_omni_dataset: # Not an omni-related request, delegate to original implementation return get_samples_old(args, tokenizer) + if is_omniinteract: + dataset_path = getattr(args, "dataset_path", None) + data_root = None + dataset_repo = DEFAULT_OMNIINTERACT_REPO + if dataset_path: + candidate = Path(dataset_path).expanduser() + if candidate.exists() or candidate.is_absolute(): + data_root = str(candidate) + else: + dataset_repo = str(dataset_path) + dataset = OmniInteractDataset( + data_root=data_root, + dataset_repo=dataset_repo, + subsets=tuple(getattr(args, "omniinteract_subsets")), + random_seed=args.seed, + disable_shuffle=getattr(args, "disable_shuffle", False), + ) + options = OmniInteractSessionOptions( + output_root=Path(getattr(args, "omniinteract_output_dir")), + timeout_s=float(getattr(args, "omniinteract_timeout_s")), + media_timeout_s=float(getattr(args, "omniinteract_media_timeout_s")), + ref_audio=str(getattr(args, "omniinteract_ref_audio")), + require_response=bool(getattr(args, "omniinteract_require_response")), + max_video_duration_s=float(getattr(args, "omniinteract_max_video_duration_s")), + ) + requests = dataset.sample( + tokenizer, + args.num_prompts, + request_id_prefix=args.request_id_prefix, + options=options, + ) + if not requests: + raise ValueError("No OmniInteract sessions were selected") + encoded_ref_audio = reference_audio_data_url(options.ref_audio) + assert encoded_ref_audio is not None + for request in requests: + case = request.omniinteract_case + assert isinstance(request, OmniInteractSampleRequest) and case is not None + duration, pcm, frames = prepare_media( + case.video_path, + VIDEO_FPS, + timeout_s=options.media_timeout_s, + max_duration_s=options.max_video_duration_s, + ) + if not any(frames): + raise ValueError(f"No video frames were decoded from {case.video_path}") + request.omniinteract_prepared_input = OmniInteractPreparedInput( + duration_s=duration, + pcm16=pcm, + video_frames=tuple(frames), + ref_audio_data_url=encoded_ref_audio, + ) + # Replace OmniInteract's ``0`` (all) with the measured request count. + args.num_prompts = len(requests) + return requests + # Handle Daily-Omni dataset if is_daily_omni: # Support: @@ -476,6 +647,7 @@ class MixRequestFuncOutput(RequestFuncOutput): image_pixels: int = 0 denoise_step_latency_ms: float = 0.0 text_latency: float = 0.0 + tpot_measured: bool = True #: Worst-case streaming-audio underrun (wall-clock seconds the player #: would have been starved). Populated by the audio-speech backend; ``0.0`` #: for backends that do not run continuity analysis. @@ -1313,11 +1485,115 @@ async def async_request_openai_audio_speech( def _realtime_websocket_url(api_url: str) -> str: - parts = urlsplit(api_url) - scheme = "wss" if parts.scheme == "https" else "ws" - query = dict(parse_qsl(parts.query, keep_blank_values=True)) - query.setdefault("duplex", "1") - return urlunsplit((scheme, parts.netloc, parts.path, urlencode(query), parts.fragment)) + return build_realtime_url(api_url, None, native_duplex=None) + + +def _nonnegative_number(value: object) -> bool: + return isinstance(value, int | float) and not isinstance(value, bool) and np.isfinite(value) and value >= 0 + + +async def _async_request_omniinteract( + request_func_input: RequestFuncInput, + *, + pbar: tqdm | None, +) -> MixRequestFuncOutput: + case = getattr(request_func_input, "omniinteract_case", None) + options = getattr(request_func_input, "omniinteract_options", None) + output = MixRequestFuncOutput() + output.prompt_len = request_func_input.prompt_len + output.start_time = time.perf_counter() + try: + if case is None or not isinstance(options, OmniInteractSessionOptions): + raise ValueError("OmniInteract RequestFuncInput is missing its dataset session layout") + prepared_input = getattr(request_func_input, "omniinteract_prepared_input", None) + if not isinstance(prepared_input, OmniInteractPreparedInput): + raise ValueError("OmniInteract media must be prepared before benchmark request timing") + headers = _get_headers() + _update_headers_common(headers, request_func_input) + config = OmniInteractBenchmarkConfig( + endpoint=request_func_input.api_url, + model=request_func_input.model_name or request_func_input.model, + output_root=options.output_root, + timeout_s=options.timeout_s, + media_timeout_s=options.media_timeout_s, + max_video_duration_s=options.max_video_duration_s, + ref_audio=options.ref_audio, + require_response=options.require_response, + extra_headers=headers or None, + extra_body=dict(request_func_input.extra_body or {}) or None, + ) + case_result = await run_omniinteract_case( + case, + config, + request_index=request_func_input.request_id or uuid.uuid4().hex, + capture_artifacts=request_func_input.request_id is not None, + prepared_input=prepared_input, + ) + output.latency = case_result.latency_s + output.generated_text = case_result.transcript + output.output_tokens = case_result.output_tokens + output.audio_duration = case_result.audio_bytes / (24_000 * 2) + output.audio_frames = case_result.audio_bytes // 2 + session_metrics = case_result.duplex_session_metrics + output.ttft = float(session_metrics.get("mean_ttft_ms") or 0.0) / 1000.0 + output.audio_ttfp = float(session_metrics.get("mean_ttfp_ms") or 0.0) / 1000.0 + output.audio_rtf = float(session_metrics.get("mean_rtf") or 0.0) + 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 + ) + 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 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", + case_result.session_id, + ) + output.duplex_request_metrics = case_result.duplex_request_metrics + output.duplex_session_metrics = session_metrics + output.success = case_result.success + output.error = case_result.error + setattr(output, "omniinteract_case", case) + setattr(output, "omniinteract_case_result", case_result) + except Exception: + output.success = False + output.error = traceback.format_exc() + logger.error("OmniInteract Realtime request failed: %s", output.error) + if case is not None: + case_result = OmniInteractCaseResult( + subset=case.subset, + video=str(case.video_path), + output_dir="", + error=output.error, + ) + setattr(output, "omniinteract_case", case) + setattr(output, "omniinteract_case_result", case_result) + if pbar: + pbar.update(1) + return output async def async_request_openai_realtime_duplex( @@ -1325,8 +1601,9 @@ async def async_request_openai_realtime_duplex( session: aiohttp.ClientSession, pbar: tqdm | None = None, ) -> MixRequestFuncOutput: - """Run one or more Seed-TTS turns through one explicit Realtime TTS session.""" del session + if getattr(request_func_input, "omniinteract_case", None) is not None: + return await _async_request_omniinteract(request_func_input, pbar=pbar) output = MixRequestFuncOutput() output.prompt_len = request_func_input.prompt_len output.start_time = time.perf_counter() @@ -1387,8 +1664,10 @@ async def async_request_openai_realtime_duplex( ) await client.send({"type": "response.create"}) await wait_for( - lambda: client.events.count("response.done") > done_before - or len(client.events.errors()) > errors_before, + lambda: ( + client.events.count("response.done") > done_before + or len(client.events.errors()) > errors_before + ), timeout_s=180.0, label=f"Seed-TTS Realtime TTS turn {request_index} response.done", ) @@ -1433,19 +1712,7 @@ async def async_request_openai_realtime_duplex( channels=1, ) ) - turn_transcripts.append( - "".join( - str(event.get("delta") or "") - for event in client.events.events - if client.events.response_id(event) == response_id - and event.get("type") - in { - "response.audio_transcript.delta", - "response.output_text.delta", - "response.text.delta", - } - ) - ) + turn_transcripts.append(client.events.response_text(response_id)) await client.acknowledge_playback() request_finished_at = time.perf_counter() session_metrics = summarize_session_request_metrics( @@ -1520,6 +1787,7 @@ async def async_request_openai_realtime_duplex( from vllm_omni.benchmarks.metrics.metrics import ( MultiModalsBenchmarkMetrics, calculate_metrics, + has_metric_samples, ) # ruff: noqa: E402 @@ -1626,6 +1894,7 @@ async def benchmark( ) _attach_daily_omni_to_request_func_input(input_requests[0], test_input) _attach_seed_tts_to_request_func_input(input_requests[0], test_input) + _attach_omniinteract_to_request_func_input(input_requests[0], test_input) if ready_check_timeout_sec > 0: test_output = await wait_for_endpoint( @@ -1741,6 +2010,7 @@ async def probe_loop(): print(f"Probe request rate: {probe_request_rate} req/s") probe_task = asyncio.create_task(probe_loop()) + _prepare_omniinteract_batch(input_requests) benchmark_start_time = time.perf_counter() tasks: list[asyncio.Task] = [] @@ -1801,6 +2071,7 @@ async def probe_loop(): ) _attach_daily_omni_to_request_func_input(request, request_func_input) _attach_seed_tts_to_request_func_input(request, request_func_input) + _attach_omniinteract_to_request_func_input(request, request_func_input) tasks.append( asyncio.create_task(limited_request_func(request_func_input=request_func_input, session=session, pbar=pbar)) ) @@ -1815,6 +2086,8 @@ async def probe_loop(): benchmark_duration = time.perf_counter() - benchmark_start_time + omniinteract_summary = _finalize_omniinteract_batch(input_requests, outputs) + if task_type == TaskType.GENERATION: metrics, actual_output_lens = calculate_metrics( input_requests=input_requests, @@ -1839,6 +2112,13 @@ async def probe_loop(): actual_output_lens = 0 if isinstance(metrics, MultiModalsBenchmarkMetrics): + + def measured_ttft(output: RequestFuncOutput) -> float | None: + session_metrics = getattr(output, "duplex_session_metrics", None) + if isinstance(session_metrics, dict) and session_metrics.get("mean_ttft_ms") is None: + return None + return output.ttft + result = { "duration": benchmark_duration, "completed": metrics.completed, @@ -1859,7 +2139,7 @@ async def probe_loop(): "input_lens": [output.prompt_len for output in outputs], "start_times": [output.start_time for output in outputs], "output_lens": actual_output_lens, - "ttfts": [output.ttft for output in outputs], + "ttfts": [measured_ttft(output) for output in outputs], "itls": [output.itl for output in outputs], "generated_texts": [output.generated_text for output in outputs], "errors": [output.error for output in outputs], @@ -1893,6 +2173,8 @@ async def probe_loop(): ] if duplex_session_metrics: result["duplex_session_metrics"] = duplex_session_metrics + if omniinteract_summary is not None: + result["omniinteract"] = omniinteract_summary from vllm_omni.benchmarks.data_modules.daily_omni_eval import ( compute_daily_omni_accuracy_metrics, @@ -1949,6 +2231,8 @@ def process_one_metric( # metric. if metric_attribute_name not in result_percentile_metrics: return + if not has_metric_samples(metrics, metric_attribute_name): + return # No text tokens generated (e.g. pure TTS speech endpoint): per-token # latency metrics (ttft/tpot/itl) are undefined, so skip them. is_text_token_metric = not (metric_attribute_name == "e2el" or metric_attribute_name.startswith("audio")) diff --git a/vllm_omni/benchmarks/serve.py b/vllm_omni/benchmarks/serve.py index 8e3b19ea90b..2c4f8d13043 100644 --- a/vllm_omni/benchmarks/serve.py +++ b/vllm_omni/benchmarks/serve.py @@ -1,6 +1,13 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from __future__ import annotations + import argparse import asyncio +import contextlib import os +from pathlib import Path from typing import Any from vllm.benchmarks.serve import main_async @@ -9,6 +16,7 @@ # This monkey-patches vllm.benchmarks.datasets.get_samples before it's used # Must be imported before any vllm.benchmarks module usage import vllm_omni.benchmarks.patch.patch # noqa: F401 +from vllm_omni.benchmarks.omniinteract import omniinteract_output_lock from vllm_omni.benchmarks.patch.patch import ( maybe_enable_stage_metrics, set_print_stage, @@ -28,4 +36,10 @@ def main(args: argparse.Namespace) -> dict[str, Any]: getattr(args, "extra_body", None), enabled=should_request_stage_metrics(args), ) - return asyncio.run(main_async(args)) + lock = ( + omniinteract_output_lock(Path(args.omniinteract_output_dir)) + if getattr(args, "dataset_name", None) == "omniinteract" + else contextlib.nullcontext() + ) + with lock: + return asyncio.run(main_async(args)) diff --git a/vllm_omni/entrypoints/cli/benchmark/cli_args.py b/vllm_omni/entrypoints/cli/benchmark/cli_args.py index c7f22087013..2cf66876931 100644 --- a/vllm_omni/entrypoints/cli/benchmark/cli_args.py +++ b/vllm_omni/entrypoints/cli/benchmark/cli_args.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + """vLLM-Omni extensions for the ``vllm bench serve`` CLI. Core functions: @@ -16,6 +19,63 @@ """ import argparse +import math +from pathlib import Path + +_DEFAULT_OMNIINTERACT_NUM_PROMPTS = 3 + + +def _positive_finite_float(value: str) -> float: + parsed = float(value) + if not math.isfinite(parsed) or parsed <= 0: + raise argparse.ArgumentTypeError(f"must be a finite positive number, got {value!r}") + return parsed + + +def _existing_file(value: str) -> str: + path = Path(value).expanduser() + if not path.is_file(): + raise argparse.ArgumentTypeError(f"file does not exist: {value!r}") + return str(path) + + +def add_omniinteract_cli_args(parser: argparse.ArgumentParser) -> None: + from vllm_omni.benchmarks.data_modules.omniinteract_dataset import ( + DEFAULT_MAX_VIDEO_DURATION_S, + OMNIINTERACT_SUBSETS, + ) + + group = parser.add_argument_group("OmniInteract Benchmark Options") + group.add_argument( + "--omniinteract-subsets", nargs="+", choices=OMNIINTERACT_SUBSETS, default=list(OMNIINTERACT_SUBSETS) + ) + group.add_argument( + "--omniinteract-timeout-s", type=_positive_finite_float, default=900.0, help="Complete session timeout." + ) + group.add_argument( + "--omniinteract-media-timeout-s", + type=_positive_finite_float, + default=600.0, + help="Per-command media timeout.", + ) + group.add_argument( + "--omniinteract-max-video-duration-s", + type=_positive_finite_float, + default=DEFAULT_MAX_VIDEO_DURATION_S, + help="Reject media longer than this safety limit before decoding.", + ) + group.add_argument( + "--omniinteract-ref-audio", type=_existing_file, help="Reference WAV for native-duplex audio output." + ) + group.add_argument( + "--omniinteract-require-response", action="store_true", help="Fail LISTEN-only functional E2E cases." + ) + group.add_argument( + "--omniinteract-output-dir", + type=Path, + default=Path("omniinteract-output"), + help="Directory for case and evaluator artifacts.", + ) def add_multi_stage_cli_args(parser: argparse.ArgumentParser) -> None: @@ -188,6 +248,7 @@ def add_seed_tts_cli_args(parser: argparse.ArgumentParser) -> None: _OMNI_BENCH_DATASET_CHOICES = ( "daily-omni", + "omniinteract", "seed-tts", "seed-tts-text", "seed-tts-design", @@ -218,6 +279,11 @@ def extend_omni_choices(parser: argparse.ArgumentParser) -> None: def update_omni_help(parser: argparse.ArgumentParser) -> None: """Update upstream argument help text to describe Omni-specific behavior.""" for action in parser._actions: + if action.dest == "num_prompts": + action.help = ( + f"{action.help} OmniInteract uses {_DEFAULT_OMNIINTERACT_NUM_PROMPTS} when this option is omitted; " + "0 selects all available cases." + ) if action.dest == "percentile_metrics": action.help = ( "Comma-separated list of selected metrics to report percentiles. " @@ -254,6 +320,7 @@ def update_omni_help(parser: argparse.ArgumentParser) -> None: def add_omni_args(parser: argparse.ArgumentParser) -> None: """Register all vLLM-Omni serving benchmark arguments.""" add_daily_omni_cli_args(parser) + add_omniinteract_cli_args(parser) add_seed_tts_cli_args(parser) add_multi_stage_cli_args(parser) add_diffusion_cli_args(parser) @@ -261,6 +328,28 @@ def add_omni_args(parser: argparse.ArgumentParser) -> None: def preprocess_serve_args(args: argparse.Namespace) -> None: """Apply serving benchmark CLI transformations after parsing.""" + if getattr(args, "dataset_name", None) == "omniinteract": + if getattr(args, "backend", None) != "openai-realtime-duplex": + raise ValueError("OmniInteract requires --backend openai-realtime-duplex") + if getattr(args, "endpoint", None) != "/v1/realtime": + raise ValueError("OmniInteract requires --endpoint /v1/realtime") + if not getattr(args, "omniinteract_ref_audio", None): + raise ValueError("OmniInteract requires --omniinteract-ref-audio") + if getattr(args, "ignore_eos", False): + raise ValueError("OmniInteract does not support --ignore-eos") + if getattr(args, "profile", False): + raise ValueError("OmniInteract does not support --profile") + if getattr(args, "skip_tokenizer_init", False): + raise ValueError("OmniInteract does not support --skip-tokenizer-init") + if float(getattr(args, "probe_request_rate", 0.0) or 0.0) > 0: + raise ValueError("OmniInteract does not support --probe-request-rate") + if "num_prompts" not in getattr(args, "explicit_keys", ()): + args.num_prompts = _DEFAULT_OMNIINTERACT_NUM_PROMPTS + max_concurrency = getattr(args, "max_concurrency", None) + if max_concurrency is None: + args.max_concurrency = 1 + elif max_concurrency <= 0: + raise ValueError("OmniInteract requires --max-concurrency to be positive") extra_body = dict(getattr(args, "extra_body", None) or {}) bot_task = getattr(args, "bot_task", None) if getattr(args, "backend", None) == "openai-image-edits-omni" and bot_task is not None: diff --git a/vllm_omni/entrypoints/cli/benchmark/serve.py b/vllm_omni/entrypoints/cli/benchmark/serve.py index 5e46aecd2b4..7c7b0522175 100644 --- a/vllm_omni/entrypoints/cli/benchmark/serve.py +++ b/vllm_omni/entrypoints/cli/benchmark/serve.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + """Online serving benchmark subcommand for vLLM-Omni. ``OmniBenchmarkServingSubcommand`` starts from vLLM's serving benchmark @@ -24,7 +27,7 @@ class OmniBenchmarkServingSubcommand(OmniBenchmarkSubcommandBase): """The `serve` subcommand for vllm bench.""" name = "serve" - help = "Benchmark the online serving throughput. Supports Daily-Omni and Seed-TTS datasets." + help = "Benchmark online serving. Supports Daily-Omni, OmniInteract, and Seed-TTS datasets." @classmethod def add_cli_args(cls, parser: argparse.ArgumentParser) -> None: diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py index 97391a5c9a3..78ec2cd3dec 100644 --- a/vllm_omni/experimental/fullduplex/client.py +++ b/vllm_omni/experimental/fullduplex/client.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + """Reusable Realtime WebSocket, PCM, and event helpers for MiniCPM-o demos.""" from __future__ import annotations @@ -30,6 +33,13 @@ def _rounded_ms(value: float) -> float: return round(float(value), 3) +def _finite_number(value: object, *, nonnegative: bool = False) -> float | None: + if isinstance(value, bool) or not isinstance(value, int | float): + return None + parsed = float(value) + return parsed if math.isfinite(parsed) and (not nonnegative or parsed >= 0) else None + + 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: @@ -56,11 +66,7 @@ def summarize_session_request_metrics( """Average client-observed metrics across turns that emitted audio.""" def mean(metric: str, *, digits: int = 3) -> float | None: - values = [ - float(request[metric]) - for request in request_metrics - if isinstance(request.get(metric), int | float) and math.isfinite(float(request[metric])) - ] + values = [value for request in request_metrics if (value := _finite_number(request.get(metric))) is not None] return round(sum(values) / len(values), digits) if values else None return { @@ -93,24 +99,57 @@ def _event_stage_metrics(event: dict[str, object]) -> dict[str, object] | None: def build_realtime_url( url: str, - model: str, + model: str | None, *, autostart: bool | None = None, + native_duplex: bool | None = True, session_id: str | None = None, ) -> str: """Add the explicit native-duplex query parameters to a Realtime URL.""" parts = urlsplit(url) + if parts.scheme in {"http", "https"}: + parts = parts._replace(scheme="ws" if parts.scheme == "http" else "wss") + if parts.scheme not in {"ws", "wss"} or not parts.netloc: + raise ValueError(f"Unsupported Realtime URL: {url!r}") query = dict(parse_qsl(parts.query, keep_blank_values=True)) - query.setdefault("duplex", "1") - query.setdefault("model", model) - query.setdefault("minicpmo45_native_duplex", "1") + query["duplex"] = "1" + if model: + query["model"] = model + if native_duplex is not None: + query["minicpmo45_native_duplex"] = "1" if native_duplex else "0" if autostart is not None: - query.setdefault("autostart", "1" if autostart else "0") + query["autostart"] = "1" if autostart else "0" if session_id: - query.setdefault("session_id", session_id) + query["session_id"] = session_id return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment)) +def reference_audio_data_url(path: str | None) -> str | None: + """Encode a local reference WAV for a Realtime session update.""" + if path is None: + return None + audio = Path(path).expanduser().resolve() + if not audio.is_file(): + raise FileNotFoundError(f"Reference audio does not exist: {audio}") + return "data:audio/wav;base64," + base64.b64encode(audio.read_bytes()).decode("ascii") + + +def chunk_period_ms(events: list[dict[str, object]], *, default: int = 1000) -> int: + """Read the negotiated native-duplex model-unit duration.""" + for event in reversed(events): + session = event.get("session") + capabilities = session.get("capabilities") if isinstance(session, dict) else None + value = capabilities.get("chunk_period_ms") if isinstance(capabilities, dict) else None + if isinstance(value, int) and not isinstance(value, bool) and value > 0: + return value + return default + + +def has_residual_model_unit(pcm16: bytes, *, chunk_period_ms: int) -> bool: + unit_bytes = PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * chunk_period_ms // 1000 + return bool(unit_bytes and len(pcm16) % unit_bytes) + + def read_pcm16_wav(path: Path) -> bytes: """Read a mono, uncompressed, 16 kHz PCM16 WAV file.""" with wave.open(str(path), "rb") as wav_file: @@ -202,6 +241,20 @@ def audio_bytes(self, response_id: str | None = None) -> bytes: chunk for response_id in self.response_ids for chunk in self.response_audio.get(response_id, ()) ) + def response_text(self, response_id: str) -> str: + """Join all text/transcript deltas for one response identity.""" + return "".join( + str(event.get("delta") or "") + for event in self.events + if self.response_id(event) == response_id + and event.get("type") + in { + "response.audio_transcript.delta", + "response.output_text.delta", + "response.text.delta", + } + ) + def errors(self) -> list[dict[str, object]]: return [event for event in self.events if event.get("type") == "error"] @@ -282,15 +335,16 @@ def timing_summary( 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)] + [parsed for value in raw_itls if (parsed := _finite_number(value, nonnegative=True)) is not None] 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), + "output_token_count": int(_finite_number(stage0_metrics.get("num_tokens_out"), nonnegative=True) or 0), + "ttft_ms": _finite_number(stage0_metrics.get("vllm_ttft_ms"), nonnegative=True) or 0.0, + "tpot_ms": _finite_number(stage0_metrics.get("vllm_tpot_ms"), nonnegative=True), + "itls_ms": itls, "inter_token_interval_ms": _interval_summary(itls), } @@ -368,15 +422,26 @@ def timing_summary( class RealtimeDuplexClient: """Small async client used by the user demo and reusable smoke probes.""" - def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024) -> None: + def __init__( + self, + url: str, + *, + max_size: int = 64 * 1024 * 1024, + additional_headers: dict[str, str] | None = None, + ) -> None: self.url = url self.max_size = max_size + self.additional_headers = additional_headers self.events = RealtimeEventCollector() self._ws: Any = None self._reader_task: asyncio.Task[None] | None = None async def __aenter__(self) -> RealtimeDuplexClient: - self._ws = await websockets.connect(self.url, max_size=self.max_size) + self._ws = await websockets.connect( + self.url, + max_size=self.max_size, + additional_headers=self.additional_headers, + ) self._reader_task = asyncio.create_task(self._read_events()) return self @@ -405,6 +470,19 @@ async def _read_events(self) -> None: async def send(self, event: dict[str, object]) -> None: await self._ws.send(json.dumps(event)) + def raise_if_reader_stopped(self) -> None: + """Fail a caller waiting on events after the WebSocket reader exits.""" + + task = self._reader_task + if task is None or not task.done(): + return + if task.cancelled(): + raise ConnectionError("Realtime WebSocket reader was cancelled") + error = task.exception() + if error is not None: + raise ConnectionError("Realtime WebSocket reader failed") from error + raise ConnectionError("Realtime WebSocket closed before the requested event arrived") + async def configure( self, model: str, @@ -418,6 +496,7 @@ async def configure( temperature: float | None = None, extra_body: dict[str, object] | None = None, session_id: str | None = None, + idle_timeout_s: float | None = None, timeout_s: float = 20.0, ) -> None: session_extra_body = dict(extra_body or {}) @@ -448,14 +527,23 @@ async def configure( if instructions is not None: session["instructions"] = instructions if initial_user_text is not None: - extra_body = session["extra_body"] - assert isinstance(extra_body, dict) - extra_body["duplex_initial_user_text"] = initial_user_text + session_extra = session["extra_body"] + assert isinstance(session_extra, dict) + session_extra["duplex_initial_user_text"] = initial_user_text if session_id: session["session_id"] = session_id + if idle_timeout_s is not None: + session["idle_timeout_s"] = idle_timeout_s await self.send({"type": "session.update", "session": session}) + + def session_created() -> bool: + if self.events.count("session.created") > 0: + return True + self.raise_if_reader_stopped() + return False + await wait_for( - lambda: self.events.count("session.created") > 0, + session_created, timeout_s=timeout_s, label="session.created", ) @@ -498,20 +586,31 @@ async def acknowledge_playback(self) -> None: if not pcm16: continue played_ms = len(pcm16) * 1000 // (self.events.output_sample_rate_hz * PCM16_BYTES_PER_SAMPLE) - await self.send( - { - "type": "playback.ack", - "response_id": response_id, - "item_id": f"item_{response_id}", - "played_ms": played_ms, - "committed_ms": played_ms, - } - ) + await self.send_playback_ack(response_id, played_ms) + + async def send_playback_ack(self, response_id: str, played_ms: int) -> None: + await self.send( + { + "type": "playback.ack", + "response_id": response_id, + "item_id": f"item_{response_id}", + "played_ms": played_ms, + "committed_ms": played_ms, + } + ) async def close_session(self, *, timeout_s: float = 20.0) -> None: + from_index = len(self.events.events) await self.send({"type": "session.close"}) + + def session_closed() -> bool: + if any(event.get("type") == "session.closed" for event in self.events.events[from_index:]): + return True + self.raise_if_reader_stopped() + return False + await wait_for( - lambda: self.events.count("session.closed") > 0, + session_closed, timeout_s=timeout_s, label="session.closed", )