From f75c21bcd02667d9505bddb8bc66ff69b71f5481 Mon Sep 17 00:00:00 2001 From: Canlin Guo Date: Sat, 6 Dec 2025 14:20:26 +0800 Subject: [PATCH] [E2E] Add Qwen2.5-Omni model test with OmniRunner (#168) Signed-off-by: gcanlin Signed-off-by: Isotr0py Co-authored-by: Isotr0py Co-authored-by: Gao Han --- .buildkite/pipeline.yml | 21 ++ pyproject.toml | 1 + tests/multi_stages/__init__.py | 0 tests/multi_stages/conftest.py | 251 ++++++++++++++++++ .../stage_configs/qwen2_5_omni_ci.yaml | 105 ++++++++ tests/multi_stages/test_qwen_omni.py | 73 +++++ tests/multi_stages/utils.py | 206 ++++++++++++++ vllm_omni/engine/arg_utils.py | 2 + 8 files changed, 659 insertions(+) create mode 100644 tests/multi_stages/__init__.py create mode 100644 tests/multi_stages/conftest.py create mode 100644 tests/multi_stages/stage_configs/qwen2_5_omni_ci.yaml create mode 100644 tests/multi_stages/test_qwen_omni.py create mode 100644 tests/multi_stages/utils.py diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index 403675563..0b7b0f44a 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -14,7 +14,9 @@ steps: - ".buildkite/scripts/simple_test.sh" agents: queue: "cpu_queue_premerge" + - label: "Diffusion Model Test" + timeout_in_minutes: 15 depends_on: image-build commands: - pytest -s -v tests/single_stage/test_diffusion_model.py @@ -29,3 +31,22 @@ steps: - "HF_HOME=/fsx/hf_cache" volumes: - "/fsx/hf_cache:/fsx/hf_cache" + + - label: "Omni Model Test" + timeout_in_minutes: 15 + depends_on: image-build + commands: + - export VLLM_LOGGING_LEVEL=DEBUG + - export VLLM_WORKER_MULTIPROC_METHOD=spawn + - pytest -s -v tests/multi_stages/ + agents: + queue: "gpu_1_queue" # g6.4xlarge instance on AWS, has 1 L4 GPU + plugins: + - docker#v5.2.0: + image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT + always-pull: true + propagate-environment: true + environment: + - "HF_HOME=/fsx/hf_cache" + volumes: + - "/fsx/hf_cache:/fsx/hf_cache" diff --git a/pyproject.toml b/pyproject.toml index c994b252f..de3200df3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -156,6 +156,7 @@ markers = [ "integration: Integration tests", "benchmark: Benchmark tests", "slow: Slow tests", + "core_model: enable this model test in each PR instead of only nightly", ] [tool.typos.default] diff --git a/tests/multi_stages/__init__.py b/tests/multi_stages/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/multi_stages/conftest.py b/tests/multi_stages/conftest.py new file mode 100644 index 000000000..521061c25 --- /dev/null +++ b/tests/multi_stages/conftest.py @@ -0,0 +1,251 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +""" +Pytest configuration and fixtures for vllm-omni tests. +""" + +from typing import Any + +import pytest +from vllm.distributed.parallel_state import cleanup_dist_env_and_memory +from vllm.sampling_params import SamplingParams + +from vllm_omni.entrypoints.omni import Omni + +PromptAudioInput = list[tuple[Any, int]] | tuple[Any, int] | None +PromptImageInput = list[Any] | Any | None +PromptVideoInput = list[Any] | Any | None + + +class OmniRunner: + """ + Test runner for Omni models. + """ + + def __init__( + self, + model_name: str, + seed: int = 42, + init_sleep_seconds: int = 20, + batch_timeout: int = 10, + init_timeout: int = 300, + shm_threshold_bytes: int = 65536, + log_stats: bool = False, + stage_configs_path: str | None = None, + **kwargs, + ) -> None: + """ + Initialize an OmniRunner for testing. + + Args: + model_name: The model name or path + seed: Random seed for reproducibility + init_sleep_seconds: Sleep time after starting each stage + batch_timeout: Timeout for batching in seconds + init_timeout: Timeout for initializing stages in seconds + shm_threshold_bytes: Threshold for using shared memory + log_stats: Enable detailed statistics logging + stage_configs_path: Optional path to YAML stage config file + **kwargs: Additional arguments passed to Omni + """ + self.model_name = model_name + self.seed = seed + + self.omni = Omni( + model=model_name, + log_stats=log_stats, + init_sleep_seconds=init_sleep_seconds, + batch_timeout=batch_timeout, + init_timeout=init_timeout, + shm_threshold_bytes=shm_threshold_bytes, + stage_configs_path=stage_configs_path, + **kwargs, + ) + + def get_default_sampling_params_list(self) -> list[SamplingParams]: + """ + Get a list of default sampling parameters for all stages. + + Returns: + List of SamplingParams with default decoding for each stage + """ + return [st.default_sampling_params for st in self.omni.instance.stage_list] + + def get_omni_inputs( + self, + prompts: list[str] | str, + system_prompt: str | None = None, + audios: PromptAudioInput = None, + images: PromptImageInput = None, + videos: PromptVideoInput = None, + mm_processor_kwargs: dict[str, Any] | None = None, + ) -> list[dict[str, Any]]: + """ + Construct Omni input format from prompts and multimodal data. + + Args: + prompts: Text prompt(s) - either a single string or list of strings + system_prompt: Optional system prompt (defaults to Qwen system prompt) + audios: Audio input(s) - tuple of (audio_array, sample_rate) or list of tuples + images: Image input(s) - PIL Image or list of PIL Images + videos: Video input(s) - numpy array or list of numpy arrays + mm_processor_kwargs: Optional processor kwargs (e.g., use_audio_in_video) + + Returns: + List of prompt dictionaries suitable for Omni.generate() + """ + if system_prompt is None: + system_prompt = ( + "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." + ) + + if isinstance(prompts, str): + prompts = [prompts] + + def _normalize_mm_input(mm_input, num_prompts): + if mm_input is None: + return [None] * num_prompts + if isinstance(mm_input, list): + if len(mm_input) != num_prompts: + raise ValueError( + f"Multimodal input list length ({len(mm_input)}) must match prompts length ({num_prompts})" + ) + return mm_input + return [mm_input] * num_prompts + + num_prompts = len(prompts) + audios_list = _normalize_mm_input(audios, num_prompts) + images_list = _normalize_mm_input(images, num_prompts) + videos_list = _normalize_mm_input(videos, num_prompts) + + omni_inputs = [] + for i, prompt_text in enumerate(prompts): + user_content = "" + multi_modal_data = {} + + audio = audios_list[i] + if audio is not None: + if isinstance(audio, list): + for _ in audio: + user_content += "<|audio_bos|><|AUDIO|><|audio_eos|>" + multi_modal_data["audio"] = audio + else: + user_content += "<|audio_bos|><|AUDIO|><|audio_eos|>" + multi_modal_data["audio"] = audio + + image = images_list[i] + if image is not None: + if isinstance(image, list): + for _ in image: + user_content += "<|vision_bos|><|IMAGE|><|vision_eos|>" + multi_modal_data["image"] = image + else: + user_content += "<|vision_bos|><|IMAGE|><|vision_eos|>" + multi_modal_data["image"] = image + + video = videos_list[i] + if video is not None: + if isinstance(video, list): + for _ in video: + user_content += "<|vision_bos|><|VIDEO|><|vision_eos|>" + multi_modal_data["video"] = video + else: + user_content += "<|vision_bos|><|VIDEO|><|vision_eos|>" + multi_modal_data["video"] = video + + user_content += prompt_text + + full_prompt = ( + f"<|im_start|>system\n{system_prompt}<|im_end|>\n" + f"<|im_start|>user\n{user_content}<|im_end|>\n" + f"<|im_start|>assistant\n" + ) + + input_dict: dict[str, Any] = {"prompt": full_prompt} + if multi_modal_data: + input_dict["multi_modal_data"] = multi_modal_data + if mm_processor_kwargs: + input_dict["mm_processor_kwargs"] = mm_processor_kwargs + + omni_inputs.append(input_dict) + + return omni_inputs + + def generate( + self, + prompts: list[dict[str, Any]], + sampling_params_list: list[SamplingParams] | None = None, + ) -> list[Any]: + """ + Generate outputs for the given prompts. + + Args: + prompts: List of prompt dictionaries with 'prompt' and optionally + 'multi_modal_data' keys + sampling_params_list: List of sampling parameters for each stage. + If None, uses default parameters. + + Returns: + List of OmniRequestOutput objects from stages with final_output=True + """ + if sampling_params_list is None: + sampling_params_list = self.get_default_sampling_params_list() + + return self.omni.generate(prompts, sampling_params_list) + + def generate_multimodal( + self, + prompts: list[str] | str, + sampling_params_list: list[SamplingParams] | None = None, + system_prompt: str | None = None, + audios: PromptAudioInput = None, + images: PromptImageInput = None, + videos: PromptVideoInput = None, + mm_processor_kwargs: dict[str, Any] | None = None, + ) -> list[Any]: + """ + Convenience method to generate with multimodal inputs. + + Args: + prompts: Text prompt(s) + sampling_params_list: List of sampling parameters for each stage + system_prompt: Optional system prompt + audios: Audio input(s) + images: Image input(s) + videos: Video input(s) + mm_processor_kwargs: Optional processor kwargs + + Returns: + List of OmniRequestOutput objects from stages with final_output=True + """ + omni_inputs = self.get_omni_inputs( + prompts=prompts, + system_prompt=system_prompt, + audios=audios, + images=images, + videos=videos, + mm_processor_kwargs=mm_processor_kwargs, + ) + return self.generate(omni_inputs, sampling_params_list) + + def __enter__(self): + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit - cleanup resources.""" + self.close() + del self.omni + cleanup_dist_env_and_memory() + + def close(self): + """Close and cleanup the Omni instance.""" + if hasattr(self.omni.instance, "close"): + self.omni.instance.close() + + +@pytest.fixture(scope="session") +def omni_runner(): + return OmniRunner diff --git a/tests/multi_stages/stage_configs/qwen2_5_omni_ci.yaml b/tests/multi_stages/stage_configs/qwen2_5_omni_ci.yaml new file mode 100644 index 000000000..3e5c2b0df --- /dev/null +++ b/tests/multi_stages/stage_configs/qwen2_5_omni_ci.yaml @@ -0,0 +1,105 @@ +# stage config for running qwen2.5-omni with architecture of OmniLLM. + +# The following config has been verified on 1x 24GB GPU (L4/RTX3090). +# This config is optimized for CI e2e tests. +stage_args: + - stage_id: 0 + runtime: + process: true # Run this stage in a separate process + devices: "0" # Visible devices for this stage (CUDA_VISIBLE_DEVICES/torch.cuda.set_device) + max_batch_size: 1 + engine_args: + model_stage: thinker + model_arch: Qwen2_5OmniForConditionalGeneration + worker_cls: vllm_omni.worker.gpu_ar_worker.GPUARWorker + scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler + max_model_len: 896 + max_num_batched_tokens: 896 + max_num_seqs: 1 + gpu_memory_utilization: 0.64 + skip_mm_profiling: true + enforce_eager: true # Now we only support eager mode + trust_remote_code: true + engine_output_type: latent + enable_prefix_caching: false + is_comprehension: true + final_output: true + final_output_type: text + default_sampling_params: + temperature: 0.0 + top_p: 1.0 + top_k: -1 + max_tokens: 128 + seed: 42 + detokenize: True + repetition_penalty: 1.1 + - stage_id: 1 + runtime: + process: true + devices: "0" + max_batch_size: 1 + engine_args: + model_stage: talker + model_arch: Qwen2_5OmniForConditionalGeneration + worker_cls: vllm_omni.worker.gpu_ar_worker.GPUARWorker + scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler + max_model_len: 896 + max_num_batched_tokens: 896 + max_num_seqs: 1 + gpu_memory_utilization: 0.28 + skip_mm_profiling: true + enforce_eager: true + trust_remote_code: true + enable_prefix_caching: false + engine_output_type: latent + engine_input_source: [0] + custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen2_5_omni.thinker2talker + default_sampling_params: + temperature: 0.9 + top_p: 0.8 + top_k: 40 + max_tokens: 128 + seed: 42 + detokenize: True + repetition_penalty: 1.05 + stop_token_ids: [8294] + - stage_id: 2 + runtime: + process: true + devices: "0" # Example: use a different GPU than the previous stage; use "0" if single GPU + max_batch_size: 1 + engine_args: + model_stage: code2wav + model_arch: Qwen2_5OmniForConditionalGeneration + worker_cls: vllm_omni.worker.gpu_generation_worker.GPUGenerationWorker + scheduler_cls: vllm_omni.core.sched.omni_generation_scheduler.OmniGenerationScheduler + gpu_memory_utilization: 0.07 + enforce_eager: true + trust_remote_code: true + enable_prefix_caching: false + engine_output_type: audio + engine_input_source: [1] + final_output: true + final_output_type: audio + default_sampling_params: + temperature: 0.0 + top_p: 1.0 + top_k: -1 + max_tokens: 128 + seed: 42 + detokenize: True + repetition_penalty: 1.1 + +# Top-level runtime config (concise): default windows and stage edges +runtime: + enabled: true + defaults: + window_size: -1 # Simplified: trigger downstream only after full upstream completion + max_inflight: 1 # Simplified: process serially within each stage + edges: + - from: 0 # thinker → talker: trigger only after receiving full input (-1) + to: 1 + window_size: -1 + - from: 1 # talker → code2wav: trigger only after receiving full input (-1) + to: 2 + window_size: -1 diff --git a/tests/multi_stages/test_qwen_omni.py b/tests/multi_stages/test_qwen_omni.py new file mode 100644 index 000000000..5bd967184 --- /dev/null +++ b/tests/multi_stages/test_qwen_omni.py @@ -0,0 +1,73 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +""" +E2E tests for Qwen2.5-Omni model with mixed modality inputs and audio output. +""" + +from pathlib import Path + +import pytest +from vllm.assets.audio import AudioAsset +from vllm.assets.image import ImageAsset +from vllm.assets.video import VideoAsset +from vllm.multimodal.image import convert_image_mode + +from .conftest import OmniRunner +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") + + +@pytest.mark.core_model +@pytest.mark.parametrize("model", models) +@create_new_process_for_each_test() +def test_mixed_modalities_to_audio(omni_runner: type[OmniRunner], model: 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: + # 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 + audio = (audio[0][: 16000 * 5], audio[1]) # Trim to first 5 seconds + image = convert_image_mode(ImageAsset("cherry_blossom").pil_image.resize((128, 128)), "RGB") + video = VideoAsset(name="baby_reading", num_frames=4).np_ndarrays + + outputs = runner.generate_multimodal( + prompts=question, + audios=audio, + images=image, + videos=video, + ) + + # Verify we got outputs from multiple stages + assert len(outputs) > 0 + + # Find and verify text output (thinker stage) + text_output = None + for stage_output in outputs: + if stage_output.final_output_type == "text": + text_output = stage_output + break + + assert text_output is not None + assert len(text_output.request_output) > 0 + text_content = text_output.request_output[0].outputs[0].text + assert text_content is not None + assert len(text_content.strip()) > 0 + + # Find and verify audio output (code2wav stage) + audio_output = None + for stage_output in outputs: + if stage_output.final_output_type == "audio": + audio_output = stage_output + break + + assert audio_output is not None + assert len(audio_output.request_output) > 0 + + # Verify audio tensor exists and has content + audio_tensor = audio_output.request_output[0].multimodal_output["audio"] + assert audio_tensor is not None + assert audio_tensor.numel() > 0 diff --git a/tests/multi_stages/utils.py b/tests/multi_stages/utils.py new file mode 100644 index 000000000..931e7b506 --- /dev/null +++ b/tests/multi_stages/utils.py @@ -0,0 +1,206 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +import contextlib +import functools +import os +import signal +import subprocess +import sys +import tempfile +from collections.abc import Callable +from contextlib import ExitStack, suppress +from pathlib import Path +from typing import Any, Literal + +import cloudpickle +from typing_extensions import ParamSpec +from vllm.platforms import current_platform + +VLLM_PATH = Path(__file__).parent.parent.parent +"""Path to root of the vLLM repository.""" + + +_P = ParamSpec("_P") + + +def fork_new_process_for_each_test(func: Callable[_P, None]) -> Callable[_P, None]: + """Decorator to fork a new process for each test function. + See https://github.com/vllm-project/vllm/issues/7053 for more details. + """ + + @functools.wraps(func) + def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None: + # Make the process the leader of its own process group + # to avoid sending SIGTERM to the parent process + os.setpgrp() + from _pytest.outcomes import Skipped + + # Create a unique temporary file to store exception info from child + # process. Use test function name and process ID to avoid collisions. + with ( + tempfile.NamedTemporaryFile( + delete=False, + mode="w+b", + prefix=f"vllm_test_{func.__name__}_{os.getpid()}_", + suffix=".exc", + ) as exc_file, + ExitStack() as delete_after, + ): + exc_file_path = exc_file.name + delete_after.callback(os.remove, exc_file_path) + + pid = os.fork() + print(f"Fork a new process to run a test {pid}") + if pid == 0: + # Parent process responsible for deleting, don't delete + # in child. + delete_after.pop_all() + try: + func(*args, **kwargs) + except Skipped as e: + # convert Skipped to exit code 0 + print(str(e)) + os._exit(0) + except Exception as e: + import traceback + + tb_string = traceback.format_exc() + + # Try to serialize the exception object first + exc_to_serialize: dict[str, Any] + try: + # First, try to pickle the actual exception with + # its traceback. + exc_to_serialize = {"pickled_exception": e} + # Test if it can be pickled + cloudpickle.dumps(exc_to_serialize) + except (Exception, KeyboardInterrupt): + # Fall back to string-based approach. + exc_to_serialize = { + "exception_type": type(e).__name__, + "exception_msg": str(e), + "traceback": tb_string, + } + try: + with open(exc_file_path, "wb") as f: + cloudpickle.dump(exc_to_serialize, f) + except Exception: + # Fallback: just print the traceback. + print(tb_string) + os._exit(1) + else: + os._exit(0) + else: + pgid = os.getpgid(pid) + _pid, _exitcode = os.waitpid(pid, 0) + # ignore SIGTERM signal itself + old_signal_handler = signal.signal(signal.SIGTERM, signal.SIG_IGN) + # kill all child processes + os.killpg(pgid, signal.SIGTERM) + # restore the signal handler + signal.signal(signal.SIGTERM, old_signal_handler) + if _exitcode != 0: + # Try to read the exception from the child process + exc_info = {} + if os.path.exists(exc_file_path): + with ( + contextlib.suppress(Exception), + open(exc_file_path, "rb") as f, + ): + exc_info = cloudpickle.load(f) + + if (original_exception := exc_info.get("pickled_exception")) is not None: + # Re-raise the actual exception object if it was + # successfully pickled. + assert isinstance(original_exception, Exception) + raise original_exception + + if (original_tb := exc_info.get("traceback")) is not None: + # Use string-based traceback for fallback case + raise AssertionError( + f"Test {func.__name__} failed when called with" + f" args {args} and kwargs {kwargs}" + f" (exit code: {_exitcode}):\n{original_tb}" + ) from None + + # Fallback to the original generic error + raise AssertionError( + f"function {func.__name__} failed when called with" + f" args {args} and kwargs {kwargs}" + f" (exit code: {_exitcode})" + ) from None + + return wrapper + + +def spawn_new_process_for_each_test(f: Callable[_P, None]) -> Callable[_P, None]: + """Decorator to spawn a new process for each test function.""" + + @functools.wraps(f) + def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None: + # Check if we're already in a subprocess + if os.environ.get("RUNNING_IN_SUBPROCESS") == "1": + # If we are, just run the function directly + return f(*args, **kwargs) + + import torch.multiprocessing as mp + + with suppress(RuntimeError): + mp.set_start_method("spawn") + + # Get the module + module_name = f.__module__ + + # Create a process with environment variable set + env = os.environ.copy() + env["RUNNING_IN_SUBPROCESS"] = "1" + + with tempfile.TemporaryDirectory() as tempdir: + output_filepath = os.path.join(tempdir, "new_process.tmp") + + # `cloudpickle` allows pickling complex functions directly + input_bytes = cloudpickle.dumps((f, output_filepath)) + + repo_root = str(VLLM_PATH.resolve()) + + env = dict(env or os.environ) + env["PYTHONPATH"] = repo_root + os.pathsep + env.get("PYTHONPATH", "") + + cmd = [sys.executable, "-m", f"{module_name}"] + + returned = subprocess.run(cmd, input=input_bytes, capture_output=True, env=env) + + # check if the subprocess is successful + try: + returned.check_returncode() + except Exception as e: + # wrap raised exception to provide more information + raise RuntimeError(f"Error raised in subprocess:\n{returned.stderr.decode()}") from e + + return wrapper + + +def create_new_process_for_each_test( + method: Literal["spawn", "fork"] | None = None, +) -> Callable[[Callable[_P, None]], Callable[_P, None]]: + """Creates a decorator that runs each test function in a new process. + + Args: + method: The process creation method. Can be either "spawn" or "fork". + If not specified, it defaults to "spawn" on ROCm and XPU + platforms and "fork" otherwise. + + Returns: + A decorator to run test functions in separate processes. + """ + if method is None: + use_spawn = current_platform.is_rocm() or current_platform.is_xpu() + method = "spawn" if use_spawn else "fork" + + assert method in ["spawn", "fork"], "Method must be either 'spawn' or 'fork'" + + if method == "fork": + return fork_new_process_for_each_test + + return spawn_new_process_for_each_test diff --git a/vllm_omni/engine/arg_utils.py b/vllm_omni/engine/arg_utils.py index fc6dd5d7b..202281e06 100644 --- a/vllm_omni/engine/arg_utils.py +++ b/vllm_omni/engine/arg_utils.py @@ -51,6 +51,8 @@ class OmniEngineArgs(EngineArgs): # Create OmniModelConfig by copying all base config attributes # and adding the new omni-specific fields config_dict = base_config.__dict__.copy() + # FIXME(Isotr0py): This is a temporary workaround for multimodal_config + config_dict = {**config_dict.pop("multimodal_config", {}).__dict__, **config_dict} # Add the new omni-specific fields config_dict["stage_id"] = self.stage_id