[CI/Build] Gate one-word pronunciation on a success rate, not per request (#5681)

Signed-off-by: Shenglei Fu <sfu@confluent.io>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

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)

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(

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):

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