From 2e108a4109d9ee6291db87c59668f4b2e4678dc1 Mon Sep 17 00:00:00 2001 From: ChenWenjing <54166744+Shirley125@users.noreply.github.com> Date: Wed, 3 Jun 2026 18:57:16 +0800 Subject: [PATCH] [ci] add merge/ready ci for audio realtime api (#4069) Signed-off-by: CHEN <116010019@link.cuhk.edu.cn> --- .buildkite/test-ready.yml | 36 +++++ .../test_qwen3_omni_realtime_websocket.py | 146 +++++++++++++----- 2 files changed, 145 insertions(+), 37 deletions(-) diff --git a/.buildkite/test-ready.yml b/.buildkite/test-ready.yml index 2e28db312..417db45be 100644 --- a/.buildkite/test-ready.yml +++ b/.buildkite/test-ready.yml @@ -273,6 +273,42 @@ steps: path: /mnt/hf-cache type: DirectoryOrCreate + - label: "Omni · Qwen3-Omni Realtime WebSocket" + commands: + - | + timeout 30m bash -c " + export VLLM_WORKER_MULTIPROC_METHOD=spawn + pytest -s -v tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py::TestQwen3OmniRealtimeWebSocket::test_streaming_audio_input_pcm_output_async_chunk_smoke -m 'core_model' --run-level 'core_model' + " + agents: + queue: "mithril-h100-pool" + plugins: + - kubernetes: + podSpec: + containers: + - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT + resources: + limits: + nvidia.com/gpu: 2 + volumeMounts: + - name: devshm + mountPath: /dev/shm + - name: hf-cache + mountPath: /root/.cache/huggingface + env: + - name: HF_HOME + value: /root/.cache/huggingface + nodeSelector: + node.kubernetes.io/instance-type: gpu-h100-sxm + volumes: + - name: devshm + emptyDir: + medium: Memory + - name: hf-cache + hostPath: + path: /mnt/hf-cache + type: DirectoryOrCreate + - label: "Omni · VoxCPM2 Test" timeout_in_minutes: 20 commands: diff --git a/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py b/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py index 39b27103e..bbe50a925 100644 --- a/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py +++ b/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py @@ -2,6 +2,11 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project """ E2E online tests for Qwen3-Omni /v1/realtime WebSocket (streaming PCM in, audio out). + +Three scenarios: +- Ready CI: async_chunk on, smoke only (no send delay, no accuracy check). +- Merge CI: async_chunk on + send delay, full accuracy check. +- Merge CI: async_chunk off, no send delay, full accuracy check. """ from __future__ import annotations @@ -34,18 +39,32 @@ REALTIME_SYNTH_PHRASE_TEXT = ( "Translate into Chinese: Beijing is the Capital of China. It is the center of culture and politics" ) -# The new-schema CI overlay bakes in async_chunk: False and covers CUDA/ROCm/XPU -# via its ``platforms:`` section, so one path serves all three. +# Simulate realtime upload pacing (``openai_realtime_client.py --send-delay-ms``). +SEND_DELAY_MS = 200 + +# CI overlay bakes in async_chunk: False and covers CUDA/ROCm/XPU via ``platforms:``. default_stage_config = get_deploy_config_path("ci/qwen3_omni_moe.yaml") -realtime_server_params = [ +realtime_sync_server_params = [ pytest.param( OmniServerParams( model=MODEL, stage_config_path=default_stage_config, use_stage_cli=True, ), - id="default", + id="sync", + ), +] + +realtime_async_chunk_server_params = [ + pytest.param( + OmniServerParams( + model=MODEL, + stage_config_path=default_stage_config, + use_stage_cli=True, + server_args=["--async-chunk"], + ), + id="async_chunk", ), ] @@ -80,6 +99,7 @@ async def _run_realtime_audio_roundtrip( pcm16: bytes, *, chunk_ms: int = 100, + send_delay_ms: int = 0, ) -> dict: uri = f"ws://{host}:{port}/v1/realtime" incremental: list[bytes] = [] @@ -105,6 +125,8 @@ async def _run_realtime_audio_roundtrip( } ) ) + if send_delay_ms > 0: + await asyncio.sleep(send_delay_ms / 1000.0) await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True})) @@ -156,26 +178,49 @@ async def _run_realtime_audio_roundtrip( } +def _synthetic_pcm16_input() -> bytes: + syn = generate_synthetic_audio( + 10, + 1, + sample_rate=16000, + phrase_text=REALTIME_SYNTH_PHRASE_TEXT, + ) + wav_bytes = base64.b64decode(syn["base64"]) + return _pcm16_mono_16k_from_wav_bytes(wav_bytes) + + +def _assert_realtime_smoke(result: dict) -> None: + out_pcm = result["output_pcm"] + assert result["delta_events"] >= 1 + assert out_pcm, "No output PCM from response.audio.delta" + assert len(out_pcm) % 2 == 0 + assert len(out_pcm) >= 4096, "Output audio unexpectedly small" + assert result["output_sample_rate"] > 0 + + +def _assert_realtime_accuracy(result: dict) -> None: + final_text = (result["transcription_text"] or "").strip() + assert final_text, "Expected non-empty transcription (model text stream)" + + wav_out = _wav_bytes_from_pcm16(result["output_pcm"], result["output_sample_rate"]) + whisper_text = convert_audio_bytes_to_text(wav_out).strip() + assert whisper_text, "Whisper returned empty string for synthesized output audio" + + sim = cosine_similarity_text(whisper_text.lower(), final_text.lower()) + assert sim > 0.8, ( + f"Output audio transcript should match model text (sim={sim:.3f}): " + f"whisper={whisper_text!r}, model_text={final_text!r}" + ) + + class TestQwen3OmniRealtimeWebSocket: - @pytest.mark.advanced_model + @pytest.mark.core_model @pytest.mark.omni @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) - @pytest.mark.parametrize("omni_server", realtime_server_params, indirect=True) - def test_streaming_audio_input_pcm_output(self, omni_server) -> None: - """ - Short streamed 16 kHz mono PCM16 input; expect streamed PCM16 audio deltas and - transcription. Verify Whisper(output audio) aligns with model text (same idea - as multimodal omni e2e). Input speech is synthesized from - ``REALTIME_SYNTH_PHRASE_TEXT``. - """ - syn = generate_synthetic_audio( - 10, - 1, - sample_rate=16000, - phrase_text=REALTIME_SYNTH_PHRASE_TEXT, - ) - wav_bytes = base64.b64decode(syn["base64"]) - pcm16 = _pcm16_mono_16k_from_wav_bytes(wav_bytes) + @pytest.mark.parametrize("omni_server", realtime_async_chunk_server_params, indirect=True) + def test_streaming_audio_input_pcm_output_async_chunk_smoke(self, omni_server) -> None: + """Ready CI: async_chunk on, no send delay, smoke only.""" + pcm16 = _synthetic_pcm16_input() result = asyncio.run( _run_realtime_audio_roundtrip( @@ -184,25 +229,52 @@ class TestQwen3OmniRealtimeWebSocket: omni_server.model, pcm16, chunk_ms=100, + send_delay_ms=0, ) ) - out_pcm = result["output_pcm"] - assert result["delta_events"] >= 1 - assert out_pcm, "No output PCM from response.audio.delta" - assert len(out_pcm) % 2 == 0 - assert len(out_pcm) >= 4096, "Output audio unexpectedly small" - assert result["output_sample_rate"] > 0 + _assert_realtime_smoke(result) - final_text = (result["transcription_text"] or "").strip() - assert final_text, "Expected non-empty transcription (model text stream)" + @pytest.mark.advanced_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) + @pytest.mark.parametrize("omni_server", realtime_async_chunk_server_params, indirect=True) + def test_streaming_audio_input_pcm_output_async_chunk(self, omni_server) -> None: + """Merge CI: async_chunk on, paced upload, full accuracy check.""" + pcm16 = _synthetic_pcm16_input() - wav_out = _wav_bytes_from_pcm16(out_pcm, result["output_sample_rate"]) - whisper_text = convert_audio_bytes_to_text(wav_out).strip() - assert whisper_text, "Whisper returned empty string for synthesized output audio" - - sim = cosine_similarity_text(whisper_text.lower(), final_text.lower()) - assert sim > 0.8, ( - f"Output audio transcript should match model text (sim={sim:.3f}): " - f"whisper={whisper_text!r}, model_text={final_text!r}" + result = asyncio.run( + _run_realtime_audio_roundtrip( + omni_server.host, + omni_server.port, + omni_server.model, + pcm16, + chunk_ms=100, + send_delay_ms=SEND_DELAY_MS, + ) ) + + _assert_realtime_smoke(result) + _assert_realtime_accuracy(result) + + @pytest.mark.advanced_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) + @pytest.mark.parametrize("omni_server", realtime_sync_server_params, indirect=True) + def test_streaming_audio_input_pcm_output(self, omni_server) -> None: + """Merge CI: async_chunk off, no send delay, full accuracy check.""" + pcm16 = _synthetic_pcm16_input() + + result = asyncio.run( + _run_realtime_audio_roundtrip( + omni_server.host, + omni_server.port, + omni_server.model, + pcm16, + chunk_ms=100, + send_delay_ms=0, + ) + ) + + _assert_realtime_smoke(result) + _assert_realtime_accuracy(result)