[CI] Qwen3-Omni online test (#257)

Signed-off-by: Rein Yang <ruiruyang2@gmail.com>

.buildkite/pipeline.yml

@@ -19,7 +19,7 @@ steps:
     timeout_in_minutes: 15
     depends_on: image-build
     commands:
-      - pytest -s -v tests/single_stage/test_diffusion_model.py
+      - pytest -s -v tests/e2e/offline_inference/test_t2i_model.py
     agents:
       queue: "gpu_1_queue" # g6.4xlarge instance on AWS, has 1 L4 GPU
     plugins:
@@ -38,7 +38,7 @@ steps:
     commands:
       - export VLLM_LOGGING_LEVEL=DEBUG
       - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-      - pytest -s -v tests/multi_stages/
+      - pytest -s -v tests/e2e/offline_inference/test_qwen2_5_omni.py
     agents:
       queue: "gpu_4_queue" # g6.4xlarge instance on AWS, has 1 L4 GPU
     plugins:
@@ -56,7 +56,7 @@ steps:
     depends_on: image-build
     commands:
       - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-      - pytest -s -v tests/multi_stages_h100/
+      - pytest -s -v tests/e2e/offline_inference/test_qwen3_omni.py tests/e2e/online_serving/test_qwen3_omni.py
     agents:
       queue: "mithril-h100-pool"
     plugins:

tests/e2e/offline_inference/stage_configs/qwen3_omni_ci.yaml

@@ -1,6 +1,6 @@
 # Stage config for running Qwen3-Omni-MoE with 3-stage architecture
 # Stage 0: Thinker (multimodal understanding + text generation)
-# Stage 1: Talker (text embeddings → 8-layer RVQ codec codes)
+# Stage 1: Talker (text embeddings → 16-layer RVQ codec codes)
 # Stage 2: Code2Wav (8-layer RVQ codes → audio waveform)
 
 # The following config has been verified on 2x H100-80G GPUs.

tests/e2e/offline_inference/test_qwen2_5_omni.py

@@ -18,15 +18,19 @@ from .utils import create_new_process_for_each_test
 models = ["Qwen/Qwen2.5-Omni-3B"]
 
 # CI stage config optimized for 24GB GPU (L4/RTX3090)
-CI_STAGE_CONFIG_PATH = str(Path(__file__).parent / "stage_configs" / "qwen2_5_omni_ci.yaml")
+stage_configs = [str(Path(__file__).parent / "stage_configs" / "qwen2_5_omni_ci.yaml")]
+
+# Create parameter combinations for model and stage config
+test_params = [(model, stage_config) for model in models for stage_config in stage_configs]
 
 
 @pytest.mark.core_model
-@pytest.mark.parametrize("model", models)
+@pytest.mark.parametrize("test_config", test_params)
 @create_new_process_for_each_test()
-def test_mixed_modalities_to_audio(omni_runner: type[OmniRunner], model: str) -> None:
+def test_mixed_modalities_to_audio(omni_runner: type[OmniRunner], test_config: tuple[str, str]) -> None:
     """Test processing audio, image, and video together, generating audio output."""
-    with omni_runner(model, seed=42, stage_configs_path=CI_STAGE_CONFIG_PATH) as runner:
+    model, stage_config_path = test_config
+    with omni_runner(model, seed=42, stage_configs_path=stage_config_path) as runner:
         # Prepare multimodal inputs
         question = "What is recited in the audio? What is in this image? Describe the video briefly."
         audio = AudioAsset("mary_had_lamb").audio_and_sample_rate

tests/e2e/offline_inference/test_qwen3_omni.py

@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 """
-E2E tests for Qwen3-Omni model with audio/video/image input and audio output.
+E2E offline tests for Omni model with video input and audio output.
 """
 
 import os
@@ -10,20 +10,24 @@ from pathlib import Path
 import pytest
 from vllm.assets.video import VideoAsset
 
-from ..multi_stages.conftest import OmniRunner
+from .conftest import OmniRunner
 
 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
 
 models = ["Qwen/Qwen3-Omni-30B-A3B-Instruct"]
 
 # CI stage config for 2xH100-80G GPUs
-CI_STAGE_CONFIG_PATH = str(Path(__file__).parent / "stage_configs" / "qwen3_omni_ci.yaml")
+stage_configs = [str(Path(__file__).parent / "stage_configs" / "qwen3_omni_ci.yaml")]
+
+# Create parameter combinations for model and stage config
+test_params = [(model, stage_config) for model in models for stage_config in stage_configs]
 
 
-@pytest.mark.parametrize("model", models)
-def test_video_to_audio(omni_runner: type[OmniRunner], model: str) -> None:
+@pytest.mark.parametrize("test_config", test_params)
+def test_video_to_audio(omni_runner: type[OmniRunner], test_config) -> None:
     """Test processing video, generating audio output."""
-    with omni_runner(model, seed=42, stage_configs_path=CI_STAGE_CONFIG_PATH) as runner:
+    model, stage_config_path = test_config
+    with omni_runner(model, seed=42, stage_configs_path=stage_config_path) as runner:
         # Prepare inputs
         question = "Describe the video briefly."
         video = VideoAsset(name="baby_reading", num_frames=4).np_ndarrays

tests/e2e/online_serving/stage_configs/qwen3_omni_ci.yaml

@@ -0,0 +1,98 @@
+# Stage config for running Qwen3-Omni-MoE with 3-stage architecture
+# Stage 0: Thinker (multimodal understanding + text generation)
+# Stage 1: Talker (text embeddings → 16-layer RVQ codec codes)
+# Stage 2: Code2Wav (8-layer RVQ codes → audio waveform)
+
+# The following config has been verified on 2x H100-80G GPUs.
+stage_args:
+  - stage_id: 0
+    runtime:
+      devices: "0,1"
+      max_batch_size: 1
+    engine_args:
+      model_stage: thinker
+      model_arch: Qwen3OmniMoeForConditionalGeneration
+      worker_cls: vllm_omni.worker.gpu_ar_worker.GPUARWorker
+      scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
+      gpu_memory_utilization: 0.6
+      enforce_eager: true
+      trust_remote_code: true
+      engine_output_type: latent  # Output hidden states for talker
+      distributed_executor_backend: "mp"
+      enable_prefix_caching: false
+      hf_config_name: thinker_config
+      tensor_parallel_size: 2
+      load_format: dummy
+    final_output: true
+    final_output_type: text
+    is_comprehension: true
+    default_sampling_params:
+      temperature: 0.4
+      top_p: 0.9
+      top_k: 1
+      max_tokens: 100
+      seed: 42
+      detokenize: True
+      repetition_penalty: 1.05
+
+  - stage_id: 1
+    runtime:
+       devices: "1"
+       max_batch_size: 1
+    engine_args:
+       model_stage: talker
+       model_arch: Qwen3OmniMoeForConditionalGeneration
+       worker_cls: vllm_omni.worker.gpu_ar_worker.GPUARWorker
+       scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
+       gpu_memory_utilization: 0.3
+       enforce_eager: true
+       trust_remote_code: true
+       engine_output_type: latent  # Output codec codes for code2wav
+      #  tensor_parallel_size: 2
+       enable_prefix_caching: false
+       distributed_executor_backend: "mp"
+       hf_config_name: talker_config
+       load_format: dummy
+    engine_input_source: [0]
+    custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker
+    # final_output: true
+    # final_output_type: text
+    default_sampling_params:
+      temperature: 0.9
+      top_k: 50
+      max_tokens: 100
+      seed: 42
+      detokenize: False
+      repetition_penalty: 1.05
+      stop_token_ids: [2150]
+
+  - stage_id: 2
+    runtime:
+      devices: "0"
+      max_batch_size: 1
+    engine_args:
+      model_stage: code2wav
+      model_arch: Qwen3OmniMoeForConditionalGeneration
+      worker_cls: vllm_omni.worker.gpu_generation_worker.GPUGenerationWorker
+      scheduler_cls: vllm_omni.core.sched.omni_generation_scheduler.OmniGenerationScheduler
+      enforce_eager: true
+      trust_remote_code: true
+      enable_prefix_caching: false
+      engine_output_type: audio  # Final output: audio waveform
+      gpu_memory_utilization: 0.1
+      distributed_executor_backend: "mp"
+      max_num_batched_tokens: 1000000
+      hf_config_name: thinker_config
+      load_format: dummy
+    engine_input_source: [1]
+    custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen3_omni.talker2code2wav
+    final_output: true
+    final_output_type: audio
+    default_sampling_params:
+      temperature: 0.0
+      top_p: 1.0
+      top_k: -1
+      max_tokens: 200
+      seed: 42
+      detokenize: True
+      repetition_penalty: 1.1

tests/e2e/online_serving/test_qwen3_omni.py

@@ -0,0 +1,206 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+E2E Online tests for Qwen3-Omni model with video input and audio output.
+"""
+
+import os
+import socket
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import openai
+import pytest
+from vllm.assets.video import VideoAsset
+from vllm.utils import get_open_port
+
+os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
+
+models = ["Qwen/Qwen3-Omni-30B-A3B-Instruct"]
+
+# CI stage config for 2*H100-80G GPUs
+stage_configs = [str(Path(__file__).parent / "stage_configs" / "qwen3_omni_ci.yaml")]
+
+# Create parameter combinations for model and stage config
+test_params = [(model, stage_config) for model in models for stage_config in stage_configs]
+
+
+class OmniServer:
+    """Omniserver for vLLM-Omni tests."""
+
+    def __init__(
+        self,
+        model: str,
+        serve_args: list[str],
+        *,
+        env_dict: dict[str, str] | None = None,
+    ) -> None:
+        self.model = model
+        self.serve_args = serve_args
+        self.env_dict = env_dict
+        self.proc: subprocess.Popen | None = None
+        self.host = "127.0.0.1"
+        self.port = get_open_port()
+
+    def _start_server(self) -> None:
+        """Start the vLLM-Omni server subprocess."""
+        env = os.environ.copy()
+        env["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
+        if self.env_dict is not None:
+            env.update(self.env_dict)
+
+        cmd = [
+            sys.executable,
+            "-m",
+            "vllm_omni.entrypoints.cli.main",
+            "serve",
+            self.model,
+            "--omni",
+            "--host",
+            self.host,
+            "--port",
+            str(self.port),
+        ] + self.serve_args
+
+        print(f"Launching OmniServer with: {' '.join(cmd)}")
+        self.proc = subprocess.Popen(
+            cmd,
+            env=env,
+            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),  # Set working directory to vllm-omni root
+        )
+
+        # Wait for server to be ready
+        max_wait = 600  # 10 minutes
+        start_time = time.time()
+        while time.time() - start_time < max_wait:
+            try:
+                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                    sock.settimeout(1)
+                    result = sock.connect_ex((self.host, self.port))
+                    if result == 0:
+                        print(f"Server ready on {self.host}:{self.port}")
+                        return
+            except Exception:
+                pass
+            time.sleep(2)
+
+        raise RuntimeError(f"Server failed to start within {max_wait} seconds")
+
+    def __enter__(self):
+        self._start_server()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.proc:
+            self.proc.terminate()
+            try:
+                self.proc.wait(timeout=30)
+            except subprocess.TimeoutExpired:
+                self.proc.kill()
+                self.proc.wait()
+
+
+@pytest.fixture
+def omni_server(request):
+    """Start vLLM-Omni server as a subprocess with actual model weights.
+    Uses module scope so the server starts only once for all tests.
+    Multi-stage initialization can take 10-20+ minutes.
+    """
+    model, stage_config_path = request.param
+    with OmniServer(model, ["--stage-configs-path", stage_config_path]) as server:
+        yield server
+
+
+@pytest.fixture
+def client(omni_server):
+    """OpenAI client for the running vLLM-Omni server."""
+    return openai.OpenAI(
+        base_url=f"http://{omni_server.host}:{omni_server.port}/v1",
+        api_key="EMPTY",
+    )
+
+
+@pytest.fixture(scope="session")
+def base64_encoded_video() -> str:
+    """Base64 encoded video for testing."""
+    import base64
+
+    video = VideoAsset(name="baby_reading", num_frames=4)
+    with open(video.video_path, "rb") as f:
+        content = f.read()
+        return base64.b64encode(content).decode("utf-8")
+
+
+def get_system_prompt():
+    return {
+        "role": "system",
+        "content": [
+            {
+                "type": "text",
+                "text": (
+                    "You are Qwen, a virtual human developed by the Qwen Team, "
+                    "Alibaba Group, capable of perceiving auditory and visual inputs, "
+                    "as well as generating text and speech."
+                ),
+            }
+        ],
+    }
+
+
+def dummy_messages_from_video_data(
+    video_data_url: str,
+    content_text: str = "Describe the video briefly.",
+):
+    """Create messages with video data URL for OpenAI API."""
+    return [
+        get_system_prompt(),
+        {
+            "role": "user",
+            "content": [
+                {"type": "video_url", "video_url": {"url": video_data_url}},
+                {"type": "text", "text": content_text},
+            ],
+        },
+    ]
+
+
+@pytest.mark.parametrize("omni_server", test_params, indirect=True)
+def test_video_to_audio(
+    client: openai.OpenAI,
+    omni_server,
+    base64_encoded_video: str,
+) -> None:
+    """Test processing video, generating audio output via OpenAI API."""
+    # Create data URL for the base64 encoded video
+    video_data_url = f"data:video/mp4;base64,{base64_encoded_video}"
+
+    messages = dummy_messages_from_video_data(video_data_url)
+
+    # Test single completion
+    chat_completion = client.chat.completions.create(
+        model=omni_server.model,
+        messages=messages,
+    )
+
+    assert len(chat_completion.choices) == 2  # 1 for text output, 1 for audio output
+
+    # Verify text output
+    text_choice = chat_completion.choices[0]
+    assert text_choice.finish_reason == "length"
+
+    # Verify we got a response
+    text_message = text_choice.message
+    assert text_message.content is not None and len(text_message.content) >= 10
+    assert text_message.role == "assistant"
+
+    # Verify audio output
+    audio_choice = chat_completion.choices[1]
+    assert audio_choice.finish_reason == "stop"
+    audio_message = audio_choice.message
+
+    # Check if audio was generated
+    if hasattr(audio_message, "audio") and audio_message.audio:
+        assert audio_message.audio.data is not None
+        assert len(audio_message.audio.data) > 0

tests/multi_stages_h100/conftest.py

@@ -1,14 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""
-Pytest configuration for multi_stages_h100 tests.
-
-This conftest imports fixtures from the parent multi_stages conftest.
-"""
-
-# Import the omni_runner fixture from the parent directory's conftest
-# This makes the fixture available to tests in this directory
-from ..multi_stages.conftest import omni_runner  # noqa: F401
-
-# Explicitly expose the fixture for pytest discovery
-__all__ = ["omni_runner"]