[Benchmark] Add local OmniInteract realtime benchmark (#6522)

Signed-off-by: natureofnature <wzliu@connect.hku.hk>
This commit is contained in:
NATURE authored and GitHub committed 2026-08-25 15:32:58 +00:00
1 parent c608910d20
commit 497c537c6f
16 files changed
+3025 -149

No files matched your search

+2 -1
View File
@@ -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.
+41 -3
View File
@@ -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.
</details>
### 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
<details class="admonition abstract" markdown="1">
@@ -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 @@ if str(REPO_ROOT) not in sys.path:
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 @@ from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
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:
+179 -1
View File
@@ -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
# ============================================================================
+778
View File
@@ -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]
+33
View File
@@ -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 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.benchmark]
"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"),
[
+81 -3
View File
@@ -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 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
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 test_realtime_client_configure_omits_ref_audio_by_default():
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 test_realtime_client_configure_sends_explicit_ref_audio():
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,
@@ -58,18 +58,14 @@ from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
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)
@@ -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,
}
+111 -49
View File
@@ -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 @@ _FLOAT_LIST_TYPE = list[float]
_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)
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
# 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,
+933
View File
@@ -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
+308 -24
View File
@@ -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 dataclasses import dataclass, replace
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 @@ from vllm.benchmarks.lib.endpoint_request_func import (
RequestFuncOutput,
StreamedResponseHandler,
_get_chat_content,
_get_headers,
_update_headers_common,
_update_payload_common,
_validate_api_url,
@@ -46,6 +49,13 @@ from vllm_omni.benchmarks.data_modules.daily_omni_dataset import (
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.seed_tts_dataset import (
)
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 @@ from vllm.benchmarks.serve import TaskType, calculate_metrics_for_embeddings, ge
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 benchmark(
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 benchmark(
)
_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 benchmark(
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 benchmark(
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 benchmark(
"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 benchmark(
]
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 @@ async def benchmark(
# 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"))
+15 -1
View File
@@ -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 @@ from vllm.benchmarks.serve import main_async
# 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))
@@ -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 @@ by ``add_omni_args``.
"""
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:
+4 -1
View File
@@ -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:
+130 -31
View File
@@ -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 @@ class RealtimeEventCollector:
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 @@ class RealtimeEventCollector:
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 @@ class RealtimeEventCollector:
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 @@ class RealtimeDuplexClient:
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 @@ class RealtimeDuplexClient:
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 @@ class RealtimeDuplexClient:
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 @@ class RealtimeDuplexClient:
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",
)