diff --git a/tests/e2e/online_serving/test_qwen3_omni_expansion.py b/tests/e2e/online_serving/test_qwen3_omni_expansion.py index 1ba94c011..99066468d 100644 --- a/tests/e2e/online_serving/test_qwen3_omni_expansion.py +++ b/tests/e2e/online_serving/test_qwen3_omni_expansion.py @@ -452,11 +452,31 @@ def test_audio_in_video_default_loader_sampling_regression(omni_server, openai_c @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_one_word_prompt_001(omni_server, openai_client) -> None: - """ + """Catastrophic-regression gate on one-word pronunciation. + + Whether a one-word answer is pronounced intelligibly is sampled, not guaranteed: the + talker runs at temperature 0.9 (the upstream default, see qwen3_omni_moe.yaml stage 1), + so asserting it per request gates on a random variable, which is what the old 10-retry + loop was compensating for. + + Sends a fixed 40 requests and requires at least 29 to transcribe as the expected + word. 85.6% is a predeclared minimum acceptable success probability, taken from a + measured HF reference run (77/90); vLLM-Omni itself measured 92.2%. Assuming 40 + independent Bernoulli trials, >=29/40 has a nominal rejection probability of 0.9% + at p=.856, with 92.9% power at p=.60 and 55.9% at p=.70. + + Those are conditional on p, not a measured false-failure rate for this CI: 77/90 is + one point estimate (Wilson 76.8%-91.4%), and at the low end of that interval the gate + would reject about 20% of runs. Pinning down a real rate would take repeated + report-only runs on the target hardware. + + A drop to 70% is caught only about half the time, so this is a smoke gate for gross + pipeline breakage, not a quality tracker -- resolving moderate drift needs ~80+ + samples and belongs in an accuracy benchmark. See #5395. + Input Modal: text only (one-word answer constraint). Output Modal: text, audio (default ``modalities``); ``key_words`` only assert on text. Input Setting: stream=True - Datasets: single request """ messages = dummy_messages_from_mix_data( system_prompt=get_system_prompt(), @@ -474,17 +494,13 @@ def test_one_word_prompt_001(omni_server, openai_client) -> None: "transcript_language": "en", } - # Retry only when assert_omni_response fails on text/audio cosine similarity (see tests/helpers/assertions.py). - _similarity_assert_msg = "The audio content is not same as the text" - _max_retries = 10 - for attempt in range(_max_retries): - try: - openai_client.send_omni_request(request_config, request_num=get_max_batch_size()) - break - except AssertionError as e: - if _similarity_assert_msg not in str(e) or attempt == _max_retries - 1: - raise - print(f"Similarity assertion failed, retrying {attempt + 2}/{_max_retries}: {e!r}") + # The judge runs per request so mismatches can be counted, which also transcribes + # concurrently where the previous request_num=5 path judged serially -- worth watching + # on tighter cards, since each judge loads whisper on the same device as the talker and + # code2wav stages. + openai_client.send_omni_request( + request_config, request_num=40, min_successes=29, max_concurrency=get_max_batch_size() + ) @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) diff --git a/tests/helpers/assertions.py b/tests/helpers/assertions.py index 3b6fbe137..70c772435 100644 --- a/tests/helpers/assertions.py +++ b/tests/helpers/assertions.py @@ -2,12 +2,16 @@ import io import json +import math import tempfile import threading import wave +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass from io import BytesIO from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal +from typing import TYPE_CHECKING, Any, Literal, TypeVar if TYPE_CHECKING: from tests.helpers.runtime import DiffusionResponse @@ -24,8 +28,109 @@ from tests.helpers.media import ( preprocess_text, ) +_ResponseT = TypeVar("_ResponseT") + _GENDER_PIPELINE = None _GENDER_PIPELINE_LOCK = threading.Lock() + +# Assertions that sample a random variable rather than testing whether serving works: +# whether TTS audio is intelligible enough to transcribe back, and whether a preset voice's +# timbre estimates as the expected gender. Both are decided by the model and the estimator, +# so a caller measuring a success rate has to tell them apart from a serving failure. The +# messages are constants because a caller matching the wording by hand would silently stop +# matching if it were reworded, reclassifying every quality failure as a hard failure. +AUDIO_MISMATCH_MESSAGE = "The audio content is not same as the text" +GENDER_MISMATCH_MESSAGE = "estimated gender is" +_QUALITY_FAILURE_MESSAGES = (AUDIO_MISMATCH_MESSAGE, GENDER_MISMATCH_MESSAGE) + + +@dataclass(frozen=True) +class SuccessRateGate: + """Gate a sampled quality check on how many requests pass, not on each one passing. + + Built by the send helpers from their own arguments, so a test states a policy and never + touches this class. See ``collect_at_success_rate``. + + Attributes: + min_successes: Fail below this many successes out of ``request_num``. Predeclare it + from a measured baseline; tuning it until CI passes defeats the gate. + concurrency: In-flight requests, in batches until the sample count is reached. + Defaults to sending all of them at once. + """ + + min_successes: int + concurrency: int | None = None + + +def is_quality_failure(exc: BaseException) -> bool: + """True when ``exc`` is one of the sampled quality assertions above. + + A caller absorbing a few of these into a success-rate budget must not absorb an HTTP + failure, missing audio or a text-keyword miss as well, or a broken server passes. + """ + if not isinstance(exc, AssertionError): + return False + return any(message in str(exc) for message in _QUALITY_FAILURE_MESSAGES) + + +def collect_at_success_rate( + sample: Callable[[], _ResponseT], + gate: SuccessRateGate, + *, + request_num: int, + report: Callable[[str], None] = print, +) -> list[_ResponseT]: + """Send ``request_num`` samples and gate on how many passed, returning the successes. + + ``sample`` sends one request and asserts it. Only the sampled quality assertions are + counted; anything else propagates immediately. Does not retry and does not exit early, + since both would bias the rate being measured. + """ + if gate.min_successes > request_num: + raise ValueError(f"min_successes={gate.min_successes} exceeds request_num={request_num}, gate can never pass") + concurrency = min(gate.concurrency or request_num, request_num) + responses: list[_ResponseT] = [] + failures: list[str] = [] + lock = threading.Lock() + + def _one_sample(_: int) -> None: + try: + response = sample() + except AssertionError as exc: + if not is_quality_failure(exc): + raise + with lock: + failures.append(str(exc)) + return + with lock: + responses.append(response) + + for start in range(0, request_num, concurrency): + size = min(concurrency, request_num - start) + with ThreadPoolExecutor(max_workers=size) as executor: + for _ in executor.map(_one_sample, range(size)): + pass + + successes = request_num - len(failures) + report(f"success rate: {successes}/{request_num} succeeded") + assert successes >= gate.min_successes, ( + f"success rate {successes}/{request_num} is below the " + f"{gate.min_successes}/{request_num} gate " + f"(95% CI {wilson_interval(successes, request_num)}). Failures:\n" + + "\n".join(f" - {failure}" for failure in failures) + ) + return responses + + +def wilson_interval(successes: int, total: int, z: float = 1.96) -> str: + """Format a 95% Wilson score interval, so a rate failure shows how marginal it is.""" + p = successes / total + denom = 1 + z * z / total + centre = (p + z * z / (2 * total)) / denom + half = z / denom * math.sqrt(p * (1 - p) / total + z * z / (4 * total * total)) + return f"{max(0.0, centre - half):.1%}-{min(1.0, centre + half):.1%}" + + # Transcript gates default to whisper ``small`` for speed. ``small`` mishears a # short TTS clip ~0.5% of the time (e.g. "Hello"->"fellow", or hallucinating a # leading SFX token), which flakes the deterministic similarity gate. Short @@ -474,7 +579,7 @@ def _assert_preset_voice_gender_from_audio( print(f"Preset voice gender check: preset={key!r}, estimated={estimated_gender!r}, expected={expected_gender!r}") if estimated_gender != "unknown": assert estimated_gender == expected_gender, ( - f"{voice_name!r} is expected {expected_gender}, but estimated gender is {estimated_gender!r}" + f"{voice_name!r} is expected {expected_gender}, but {GENDER_MISMATCH_MESSAGE} {estimated_gender!r}" ) @@ -655,7 +760,7 @@ def assert_omni_response(response: Any, request_config: dict[str, Any], run_leve shorter_clean = _re.sub(r"[^\w\s]", "", shorter).strip() longer_clean = _re.sub(r"[^\w\s]", "", longer).strip() assert shorter_clean and (shorter_clean in longer_clean), ( - f"The audio content is not same as the text " + f"{AUDIO_MISMATCH_MESSAGE} " f"(short-text containment check failed: " f"text={text_output!r}, transcript={transcript!r})" ) @@ -666,7 +771,7 @@ def assert_omni_response(response: Any, request_config: dict[str, Any], run_leve text_output.lower(), ) print(f"similarity is: {similarity}") - assert similarity > similarity_threshold, "The audio content is not same as the text" + assert similarity > similarity_threshold, AUDIO_MISMATCH_MESSAGE if audio_ref_text: assert transcript is not None, "No audio transcript for reference-text validation" audio_similarity = cosine_similarity_text( diff --git a/tests/helpers/runtime.py b/tests/helpers/runtime.py index cfc8cc75d..c48411c9b 100644 --- a/tests/helpers/runtime.py +++ b/tests/helpers/runtime.py @@ -38,11 +38,13 @@ from vllm.distributed.parallel_state import ( from vllm.logger import init_logger from tests.helpers.assertions import ( + SuccessRateGate, assert_audio_speech_response, assert_diffusion_response, assert_http_error, assert_images_generations_response, assert_omni_response, + collect_at_success_rate, ) from tests.helpers.env import run_post_test_cleanup, run_pre_test_cleanup from tests.helpers.media import ( @@ -1891,8 +1893,20 @@ class OpenAIClientHandler: ) ] - def send_omni_request(self, request_config: dict[str, Any], request_num: int = 1) -> list[OmniResponse]: - """Chat completions via the OpenAI Python SDK (not raw HTTP).""" + def send_omni_request( + self, + request_config: dict[str, Any], + request_num: int = 1, + *, + min_successes: int | None = None, + max_concurrency: int | None = None, + ) -> list[OmniResponse]: + """Chat completions via the OpenAI Python SDK (not raw HTTP). + + ``min_successes`` gates the batch on how many of the ``request_num`` requests pass + rather than requiring each one to, and returns only the successes; ``max_concurrency`` + caps how many are in flight at once. See ``SuccessRateGate``. + """ responses: list[OmniResponse] = [] stream = request_config.get("stream", False) modalities = request_config.get("modalities", ["text", "audio"]) @@ -1923,22 +1937,6 @@ class OpenAIClientHandler: if extra_body: create_kwargs["extra_body"] = extra_body - if request_num == 1: - wall_start = time.perf_counter() - chat_completion = self.client.chat.completions.create(**create_kwargs) - resp = ( - self._process_stream_omni_response(chat_completion, wall_start=wall_start) - if stream - else self._process_non_stream_omni_response(chat_completion, wall_start=wall_start) - ) - assert_omni_response(resp, request_config, run_level=self.run_level) - if resp.e2e_latency is not None: - self._print_client_stat(f"[omni] request#1 success in {resp.e2e_latency:.3f}s") - else: - self._print_client_stat("[omni] request#1 completed") - responses.append(resp) - return responses - def _one(): wall_start = time.perf_counter() chat_completion = self.client.chat.completions.create(**create_kwargs) @@ -1948,6 +1946,33 @@ class OpenAIClientHandler: else self._process_non_stream_omni_response(chat_completion, wall_start=wall_start) ) + if min_successes is not None: + gate = SuccessRateGate(min_successes=min_successes, concurrency=max_concurrency) + + def _sample() -> OmniResponse: + resp = _one() + assert_omni_response(resp, request_config, run_level=self.run_level) + return resp + + # Not _print_client_stat: the count is the gate's verdict, so it has to survive + # log_stats=False. + return collect_at_success_rate( + _sample, + gate, + request_num=request_num, + report=lambda line: print(f"[omni] {line}", flush=True), + ) + + if request_num == 1: + resp = _one() + assert_omni_response(resp, request_config, run_level=self.run_level) + if resp.e2e_latency is not None: + self._print_client_stat(f"[omni] request#1 success in {resp.e2e_latency:.3f}s") + else: + self._print_client_stat("[omni] request#1 completed") + responses.append(resp) + return responses + with concurrent.futures.ThreadPoolExecutor(max_workers=request_num) as executor: futures = {executor.submit(_one): i + 1 for i in range(request_num)} for future in concurrent.futures.as_completed(futures): diff --git a/tests/helpers/tests/test_success_rate_helpers.py b/tests/helpers/tests/test_success_rate_helpers.py new file mode 100644 index 000000000..c1b8f4417 --- /dev/null +++ b/tests/helpers/tests/test_success_rate_helpers.py @@ -0,0 +1,110 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +import threading +from types import SimpleNamespace + +import pytest + +from tests.helpers import assertions +from tests.helpers.assertions import ( + AUDIO_MISMATCH_MESSAGE, + SuccessRateGate, + assert_omni_response, + collect_at_success_rate, + is_quality_failure, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +_GATE = SuccessRateGate(min_successes=7, concurrency=2) + + +def _fake_response(text: str) -> SimpleNamespace: + return SimpleNamespace(success=True, text_content=text, audio_content=None, audio_bytes=b"fake-wav") + + +def _omni_response_failure(monkeypatch, *, text: str, transcript: str) -> AssertionError: + """Drive the real assert_omni_response to failure and hand back what it raised.""" + monkeypatch.setattr(assertions, "convert_audio_bytes_to_text", lambda *a, **kw: transcript) + request_config = {"modalities": ["text", "audio"], "key_words": {"text": ["london"]}} + with pytest.raises(AssertionError) as excinfo: + assert_omni_response(_fake_response(text), request_config, run_level="full_model") + return excinfo.value + + +def test_real_audio_mismatch_is_tolerated(monkeypatch): + # Ties the classifier to the actual producer: if assert_omni_response's wording drifts, + # this fails instead of the gate silently turning every quality failure into a hard one. + assert is_quality_failure(_omni_response_failure(monkeypatch, text="London", transcript="런던")) + + +def test_real_keyword_failure_is_not_tolerated(monkeypatch): + assert not is_quality_failure(_omni_response_failure(monkeypatch, text="Paris", transcript="London")) + + +@pytest.mark.parametrize("exc", [RuntimeError(AUDIO_MISMATCH_MESSAGE), ValueError("boom")]) +def test_non_assertion_failures_are_not_tolerated(exc): + # A transport error carrying the same text is still a serving failure, not a sample. + assert not is_quality_failure(exc) + + +class _ScriptedSampler: + """One scripted outcome per call: raised if it is an exception, else returned.""" + + def __init__(self, outcomes: list) -> None: + self._outcomes = outcomes + self._lock = threading.Lock() + self.calls = 0 + + def __call__(self): + with self._lock: + outcome = self._outcomes[self.calls] + self.calls += 1 + if isinstance(outcome, BaseException): + raise outcome + return outcome + + +def _mismatch(detail: str) -> AssertionError: + return AssertionError(f"{AUDIO_MISMATCH_MESSAGE} (transcript={detail!r})") + + +def test_tolerated_failures_are_counted_and_every_sample_is_sent(): + sampler = _ScriptedSampler(["ok"] * 7 + [_mismatch("a"), _mismatch("b"), _mismatch("c")]) + + responses = collect_at_success_rate(sampler, _GATE, request_num=10, report=lambda _: None) + + # No early exit once the gate is already satisfied: it would bias the measured rate. + assert sampler.calls == 10 + assert responses == ["ok"] * 7 + + +def test_gate_failure_reports_the_count_and_every_failing_sample(): + outcomes = ["ok"] * 6 + [_mismatch("first"), _mismatch("second"), _mismatch("third"), _mismatch("fourth")] + + with pytest.raises(AssertionError) as excinfo: + collect_at_success_rate(_ScriptedSampler(outcomes), _GATE, request_num=10, report=lambda _: None) + + message = str(excinfo.value) + assert "6/10" in message and "7/10 gate" in message + assert all(detail in message for detail in ("first", "second", "third", "fourth")) + + +def test_serving_failure_is_not_absorbed_into_the_budget(): + sampler = _ScriptedSampler(["ok", AssertionError("No audio output is generated"), *["ok"] * 8]) + + with pytest.raises(AssertionError, match="No audio output"): + collect_at_success_rate(sampler, _GATE, request_num=10, report=lambda _: None) + + # Stopped on the serving failure rather than spending the remaining samples. + assert sampler.calls < 10 + + +def test_unreachable_gate_is_rejected_before_sending(): + sampler = _ScriptedSampler(["ok"] * 3) + + with pytest.raises(ValueError, match="can never pass"): + collect_at_success_rate(sampler, _GATE, request_num=3, report=lambda _: None) + + assert sampler.calls == 0