diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index 88cfc22d5..d0566e18b 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -139,41 +139,41 @@ steps: - "/fsx/hf_cache:/fsx/hf_cache" - - label: "Benchmark&Engine Test" - timeout_in_minutes: 15 - depends_on: image-build - commands: - - export VLLM_WORKER_MULTIPROC_METHOD=spawn - - pytest -s -v tests/benchmarks/test_serve_cli.py - - pytest -s -v tests/engine/test_async_omni_engine_abort.py - agents: - queue: "mithril-h100-pool" - plugins: - - kubernetes: - podSpec: - containers: - - image: public.ecr.aws/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: "Benchmark&Engine Test" + # timeout_in_minutes: 15 + # depends_on: image-build + # commands: + # - export VLLM_WORKER_MULTIPROC_METHOD=spawn + # - pytest -s -v tests/benchmarks/test_serve_cli.py + # - pytest -s -v tests/engine/test_async_omni_engine_abort.py + # agents: + # queue: "mithril-h100-pool" + # plugins: + # - kubernetes: + # podSpec: + # containers: + # - image: public.ecr.aws/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 Model Test" timeout_in_minutes: 15 @@ -194,42 +194,42 @@ steps: volumes: - "/fsx/hf_cache:/fsx/hf_cache" - - label: "Omni Model Test with H100" - timeout_in_minutes: 20 - depends_on: image-build - commands: - - export VLLM_WORKER_MULTIPROC_METHOD=spawn - - export VLLM_TEST_CLEAN_GPU_MEMORY="1" - - pytest -s -v tests/e2e/offline_inference/test_qwen3_omni.py - - pytest -s -v tests/e2e/online_serving/test_qwen3_omni.py -m "core_model" --run-level "core_model" - agents: - queue: "mithril-h100-pool" - plugins: - - kubernetes: - podSpec: - containers: - - image: public.ecr.aws/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 Model Test with H100" + # timeout_in_minutes: 20 + # depends_on: image-build + # commands: + # - export VLLM_WORKER_MULTIPROC_METHOD=spawn + # - export VLLM_TEST_CLEAN_GPU_MEMORY="1" + # - pytest -s -v tests/e2e/offline_inference/test_qwen3_omni.py + # - pytest -s -v tests/e2e/online_serving/test_qwen3_omni.py -m "core_model" --run-level "core_model" + # agents: + # queue: "mithril-h100-pool" + # plugins: + # - kubernetes: + # podSpec: + # containers: + # - image: public.ecr.aws/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: "Qwen3-TTS E2E Test" timeout_in_minutes: 20 @@ -251,40 +251,40 @@ steps: volumes: - "/fsx/hf_cache:/fsx/hf_cache" - - label: "Diffusion Image Edit Test with H100 (1 GPU)" - timeout_in_minutes: 20 - depends_on: image-build - commands: - - export VLLM_WORKER_MULTIPROC_METHOD=spawn - - pytest -s -v tests/e2e/online_serving/test_image_gen_edit.py - agents: - queue: "mithril-h100-pool" - plugins: - - kubernetes: - podSpec: - containers: - - image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT - resources: - limits: - nvidia.com/gpu: 1 - 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: "Diffusion Image Edit Test with H100 (1 GPU)" + # timeout_in_minutes: 20 + # depends_on: image-build + # commands: + # - export VLLM_WORKER_MULTIPROC_METHOD=spawn + # - pytest -s -v tests/e2e/online_serving/test_image_gen_edit.py + # agents: + # queue: "mithril-h100-pool" + # plugins: + # - kubernetes: + # podSpec: + # containers: + # - image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT + # resources: + # limits: + # nvidia.com/gpu: 1 + # 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: "Bagel Text2Img Model Test with H100" # timeout_in_minutes: 30 diff --git a/.buildkite/test-nightly.yaml b/.buildkite/test-nightly.yaml index f665de270..70c6b1fc4 100644 --- a/.buildkite/test-nightly.yaml +++ b/.buildkite/test-nightly.yaml @@ -5,8 +5,8 @@ steps: if: build.env("NIGHTLY") == "1" commands: - export VLLM_WORKER_MULTIPROC_METHOD=spawn - - pytest -s -v tests/e2e/online_serving/test_qwen3_omni_expansion.py - - pytest -s -v tests/examples/online_serving/test_qwen3_omni.py + - pytest -s -v tests/e2e/online_serving/test_qwen3_omni_expansion.py -m "advanced_model" --run-level "advanced_model" + - pytest -s -v tests/examples/online_serving/test_qwen3_omni.py -m "advanced_model" --run-level "advanced_model" agents: queue: "mithril-h100-pool" plugins: @@ -43,7 +43,7 @@ steps: commands: - export VLLM_LOGGING_LEVEL=DEBUG - export VLLM_WORKER_MULTIPROC_METHOD=spawn - - pytest -s -v tests/examples/online_serving/test_qwen2_5_omni.py + - pytest -s -v tests/examples/online_serving/test_qwen2_5_omni.py -m "advanced_model" --run-level "advanced_model" agents: queue: "gpu_4_queue" # g6.12xlarge instance on AWS, has 4 L4 GPU plugins: diff --git a/tests/conftest.py b/tests/conftest.py index 267386aa3..08d26a6d1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1057,7 +1057,8 @@ def omni_server(request, run_level): Multi-stage initialization can take 10-20+ minutes. """ with _omni_server_lock: - model, stage_config_path = request.param + *port, model, stage_config_path = request.param + port = port[0] if port else None if run_level == "advanced_model": stage_config_path = modify_stage_config( stage_config_path, @@ -1070,7 +1071,8 @@ def omni_server(request, run_level): }, ) - with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"]) as server: + server_args = ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"] + with OmniServer(model, server_args, port=port) if port else OmniServer(model, server_args) as server: print("OmniServer started successfully") yield server print("OmniServer stopping...") @@ -1288,11 +1290,15 @@ class OpenAIClientHandler: responses = [] stream = request_config.get("stream", False) + modalities = request_config.get("modalities", ["text", "audio"]) if request_num == 1: # Send single request chat_completion = self.client.chat.completions.create( - model=request_config.get("model"), messages=request_config.get("messages"), stream=stream + model=request_config.get("model"), + messages=request_config.get("messages"), + stream=stream, + modalities=modalities, ) if stream: @@ -1314,6 +1320,7 @@ class OpenAIClientHandler: self.client.chat.completions.create, model=request_config.get("model"), messages=request_config.get("messages"), + modalities=modalities, stream=stream, ) futures.append(future) diff --git a/tests/e2e/online_serving/test_qwen3_omni_expansion.py b/tests/e2e/online_serving/test_qwen3_omni_expansion.py index 9e84c7559..84cc85ba1 100644 --- a/tests/e2e/online_serving/test_qwen3_omni_expansion.py +++ b/tests/e2e/online_serving/test_qwen3_omni_expansion.py @@ -4,66 +4,55 @@ E2E Online tests for Qwen3-Omni model. """ -import concurrent.futures import os os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" -import threading -import time from pathlib import Path -import openai import pytest from tests.conftest import ( - OmniServer, - convert_audio_to_text, - cosine_similarity_text, dummy_messages_from_mix_data, generate_synthetic_audio, generate_synthetic_image, + generate_synthetic_video, + modify_stage_config, ) +from tests.utils import hardware_test models = ["Qwen/Qwen3-Omni-30B-A3B-Instruct"] -# CI stage config for 2*H100-80G GPUs -stage_configs = [str(Path(__file__).parent.parent / "stage_configs" / "qwen3_omni_ci.yaml")] +AUDIO_KEY = ["water", "chirping", "crackling"] +IMAGE_KEY = ["square", "quadrate"] +VIDEO_KEY = ["sphere", "globe", "circle", "round", "ball"] + +def get_chunk_config(default_path): + path = modify_stage_config( + default_path, + updates={ + "async_chunk": True, + "stage_args": { + 0: { + "engine_args.custom_process_next_stage_input_func": "vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker_async_chunk" + }, + 1: { + "engine_args.custom_process_next_stage_input_func": "vllm_omni.model_executor.stage_input_processors.qwen3_omni.talker2code2wav_async_chunk" + }, + }, + }, + deletes={"stage_args": {2: ["custom_process_input_func"]}}, + ) + return path + + +# CI stage config for 2*H100-80G GPUs +default_path = str(Path(__file__).parent.parent / "stage_configs" / "qwen3_omni_ci.yaml") +stage_configs = [default_path, get_chunk_config(default_path)] # Create parameter combinations for model and stage config test_params = [(model, stage_config) for model in models for stage_config in stage_configs] -_omni_server_lock = threading.Lock() - - -@pytest.fixture(scope="module") -def omni_server(request): - """Start vLLM-Omni server as a subprocess with actual model weights. - Uses session scope so the server starts only once for the entire test session. - Multi-stage initialization can take 10-20+ minutes. - """ - with _omni_server_lock: - model, stage_config_path = request.param - - print(f"Starting OmniServer with model: {model}") - - with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"]) as server: - print("OmniServer started successfully") - yield server - print("OmniServer stopping...") - - print("OmniServer stopped") - - -@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", - ) - - def get_system_prompt(): return { "role": "system", @@ -83,7 +72,10 @@ def get_system_prompt(): def get_prompt(prompt_type="text_only"): prompts = { "text_only": "What is the capital of China?", - "mix": "What is recited in the audio? What is in this image? Describe the video briefly.", + "mix": "What is recited in the audio? What is in this image? What is in this video?", + "text_video": "What is in this video? ", + "text_image": "What is in this image? ", + "text_audio": "What is in this audio? ", } return prompts.get(prompt_type, prompts["text_only"]) @@ -93,196 +85,228 @@ def get_max_batch_size(size_type="few"): return batch_sizes.get(size_type, 5) -def get_deploy_config(deploy_type="TP1"): - result = { - "TP1": { - "stage_args": { - 0: { - "engine_args.gpu_memory_utilization": 0.95, - "engine_args.tensor_parallel_size": 1, - "runtime.devices": "0", - }, - 2: {"runtime.devices": "1"}, - } - } - } - return result.get(deploy_type, result["TP1"]) - - +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) -def test_text_to_text_001(client: openai.OpenAI, omni_server) -> None: - """Test processing text, generating text output via OpenAI API.""" +def test_text_to_audio_001(omni_server, openai_client) -> None: messages = dummy_messages_from_mix_data(system_prompt=get_system_prompt(), content_text=get_prompt()) - # Test single completion - start_time = time.perf_counter() - chat_completion = client.chat.completions.create( - model=omni_server.model, messages=messages, max_tokens=20, modalities=["text"] - ) - # Verify E2E - print(f"the request e2e is: {time.perf_counter() - start_time}") - # TODO: Verify the E2E latency after confirmation baseline. + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["audio"], + "key_words": {"text": ["beijing"]}, + } - # Verify only output text - assert len(chat_completion.choices) == 1, "The generated content includes more than just text." - - # Verify text output success - text_choice = chat_completion.choices[0] - assert text_choice.message.content is not None, "No text output is generated" - assert chat_completion.usage.completion_tokens <= 20, "The output length more than the requested max_tokens." - assert "beijing" in text_choice.message.content.lower(), "The output do not contain keywords." + openai_client.send_request(request_config) +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) -def test_audio_to_text_001(client: openai.OpenAI, omni_server) -> None: - """Test processing text, generating text output via OpenAI API.""" - audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(1, 1)['base64']}" - messages = dummy_messages_from_mix_data(audio_data_url=audio_data_url) - # Test single completion - start_time = time.perf_counter() - chat_completion = client.chat.completions.create(model=omni_server.model, messages=messages, modalities=["text"]) - # Verify only output text - assert len(chat_completion.choices) == 1, "The generated content includes more than just text." +def test_text_to_text_audio_001(omni_server, openai_client) -> None: + messages = dummy_messages_from_mix_data(system_prompt=get_system_prompt(), content_text=get_prompt()) - # Verify text output success - text_choice = chat_completion.choices[0] - assert text_choice.message.content is not None, "No text output is generated" - assert "water" in text_choice.message.content.lower(), "The output do not contain keywords." + request_config = { + "model": omni_server.model, + "messages": messages, + "key_words": {"text": ["beijing"]}, + } - # Verify E2E - print(f"the request e2e is: {time.perf_counter() - start_time}") - # TODO: Verify the E2E latency after confirmation baseline. + openai_client.send_request(request_config) +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) -def test_audio_to_text_audio_001(client: openai.OpenAI, omni_server) -> None: - """Test processing text, generating audio output via OpenAI API.""" - - audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(1, 5)['base64']}" - - messages = dummy_messages_from_mix_data(audio_data_url=audio_data_url) - num_concurrent_requests = get_max_batch_size() - # Test single completion - e2e_list = list() - with concurrent.futures.ThreadPoolExecutor(max_workers=num_concurrent_requests) as executor: - # Submit multiple completion requests concurrently - futures = [ - executor.submit(client.chat.completions.create, model=omni_server.model, messages=messages) - for _ in range(num_concurrent_requests) - ] - start_time = time.perf_counter() - # Wait for all requests to complete and collect results - chat_completions = list() - for future in concurrent.futures.as_completed(futures): - chat_completions.append(future.result()) - # Verify E2E - current_e2e = time.perf_counter() - start_time - print(f"the request e2e is: {current_e2e}") - # TODO: Verify the E2E latency after confirmation baseline. - e2e_list.append(current_e2e) - - print(f"the avg e2e is: {sum(e2e_list) / len(e2e_list)}") - # Verify all completions succeeded - assert len(chat_completions) == num_concurrent_requests, "Not all requests succeeded." - for chat_completion in chat_completions: - # Verify audio output success - audio_message = chat_completion.choices[1].message - audio_data = audio_message.audio.data - assert audio_data is not None, "No audio output is generated" - assert audio_message.audio.expires_at > time.time(), "The generated audio has expired." - - # Verify text output success - text_choice = chat_completion.choices[0] - text_content = text_choice.message.content - assert text_choice.message.content is not None, "No text output is generated" - - # Verify text output same as audio output - audio_content = convert_audio_to_text(audio_data) - print(f"text content is: {text_content}") - print(f"audio content is: {audio_content}") - similarity = cosine_similarity_text(audio_content, text_content) - print(f"similarity between audio and text is: {similarity}") - assert similarity > 0.9, "The audio content is not same as the text" - - -@pytest.mark.parametrize("omni_server", test_params, indirect=True) -def test_image_to_text_001(client: openai.OpenAI, omni_server) -> None: - """Test processing text, generating text output via OpenAI API.""" +def test_image_to_text_001(omni_server, openai_client) -> None: image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(224, 224)['base64']}" messages = dummy_messages_from_mix_data(image_data_url=image_data_url) - # Test single completion - start_time = time.perf_counter() - chat_completion = client.chat.completions.create( - model=omni_server.model, messages=messages, max_tokens=100, modalities=["text"] - ) - # Verify E2E - print(f"the request e2e is: {time.perf_counter() - start_time}") - # TODO: Verify the E2E latency after confirmation baseline. - # Verify only output text - assert len(chat_completion.choices) == 1, "The generated content includes more than just text." + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["text"], + "key_words": {"image": IMAGE_KEY}, + } - # Verify text output success - text_choice = chat_completion.choices[0] - text_content = text_choice.message.content - assert text_content is not None, "No text output is generated" - assert chat_completion.usage.completion_tokens <= 100, "The output length more than the requested max_tokens." - assert "square" in text_content.lower(), "The output do not contain keywords." + openai_client.send_request(request_config) +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) -def test_image_to_text_audio_001(client: openai.OpenAI, omni_server) -> None: - """Test processing text, generating audio output via OpenAI API.""" +def test_image_to_audio_001(omni_server, openai_client) -> None: + image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(224, 224)['base64']}" + messages = dummy_messages_from_mix_data(image_data_url=image_data_url) - num_concurrent_requests = get_max_batch_size() + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["audio"], + "key_words": {"image": IMAGE_KEY}, + } + openai_client.send_request(request_config) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_image_to_text_audio_001(omni_server, openai_client) -> None: image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}" messages = dummy_messages_from_mix_data(image_data_url=image_data_url) - # Test single completion - e2e_list = list() - with concurrent.futures.ThreadPoolExecutor(max_workers=num_concurrent_requests) as executor: - # Submit multiple completion requests concurrently - futures = [ - executor.submit( - client.chat.completions.create, - model=omni_server.model, - messages=messages, - ) - for _ in range(num_concurrent_requests) - ] - start_time = time.perf_counter() - # Wait for all requests to complete and collect results - chat_completions = list() - for future in concurrent.futures.as_completed(futures): - chat_completions.append(future.result()) - # Verify E2E - current_e2e = time.perf_counter() - start_time - print(f"the request e2e is: {current_e2e}") - # TODO: Verify the E2E latency after confirmation baseline. - e2e_list.append(current_e2e) + request_config = { + "model": omni_server.model, + "messages": messages, + "key_words": {"image": IMAGE_KEY}, + } - print(f"the avg e2e is: {sum(e2e_list) / len(e2e_list)}") - # Verify all completions succeeded - assert len(chat_completions) == num_concurrent_requests, "Not all requests succeeded." - for chat_completion in chat_completions: - # Verify audio output success - audio_message = chat_completion.choices[1].message - audio_data = audio_message.audio.data - assert audio_data is not None, "No audio output is generated" - assert audio_message.audio.expires_at > time.time(), "The generated audio has expired." + openai_client.send_request(request_config, request_num=get_max_batch_size()) - # Verify text output success - text_choice = chat_completion.choices[0] - text_content = text_choice.message.content - assert text_content is not None, "No text output is generated" - assert "square" in text_content.lower(), "The output do not contain keywords." - # Verify text output same as audio output - audio_content = convert_audio_to_text(audio_data) - print(f"text content is: {text_content}") - print(f"audio content is: {audio_content}") - similarity = cosine_similarity_text(audio_content, text_content) - print(f"similarity between audio and text is: {similarity}") - assert similarity > 0.9, "The audio content is not same as the text" +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_video_to_text_001(omni_server, openai_client) -> None: + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(224, 224, 300)['base64']}" + messages = dummy_messages_from_mix_data(video_data_url=video_data_url) + + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["text"], + "key_words": {"video": VIDEO_KEY}, + } + + openai_client.send_request(request_config) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_video_to_audio_001(omni_server, openai_client) -> None: + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(224, 224, 300)['base64']}" + messages = dummy_messages_from_mix_data(video_data_url=video_data_url) + + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["audio"], + "key_words": {"video": VIDEO_KEY}, + } + + openai_client.send_request(request_config) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_video_to_text_audio_001(omni_server, openai_client) -> None: + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(224, 224, 300)['base64']}" + + messages = dummy_messages_from_mix_data(video_data_url=video_data_url) + + request_config = { + "model": omni_server.model, + "messages": messages, + "key_words": {"video": VIDEO_KEY}, + } + + openai_client.send_request(request_config, request_num=get_max_batch_size()) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_text_audio_to_text_audio_001(omni_server, openai_client) -> None: + audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(5, 1)['base64']}" + messages = dummy_messages_from_mix_data( + audio_data_url=audio_data_url, system_prompt=get_system_prompt(), content_text=get_prompt("text_audio") + ) + + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["text"], + "key_words": {"audio": AUDIO_KEY}, + } + + openai_client.send_request(request_config) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_text_image_to_text_audio_001(omni_server, openai_client) -> None: + image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(224, 224)['base64']}" + + messages = dummy_messages_from_mix_data( + image_data_url=image_data_url, system_prompt=get_system_prompt(), content_text=get_prompt("text_image") + ) + + request_config = { + "model": omni_server.model, + "messages": messages, + "modalities": ["audio"], + "key_words": {"image": IMAGE_KEY}, + } + + openai_client.send_request(request_config) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_text_video_to_text_audio_001(omni_server, openai_client) -> None: + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 30)['base64']}" + + messages = dummy_messages_from_mix_data( + video_data_url=video_data_url, system_prompt=get_system_prompt(), content_text=get_prompt("text_video") + ) + + request_config = { + "model": omni_server.model, + "messages": messages, + "key_words": {"video": VIDEO_KEY}, + } + + openai_client.send_request(request_config, request_num=get_max_batch_size()) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_mix_to_text_audio_001(omni_server, openai_client) -> None: + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(224, 224, 300)['base64']}" + image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(224, 224)['base64']}" + audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(5, 1)['base64']}" + messages = dummy_messages_from_mix_data( + system_prompt=get_system_prompt(), + video_data_url=video_data_url, + image_data_url=image_data_url, + audio_data_url=audio_data_url, + content_text=get_prompt("mix"), + ) + + request_config = { + "model": omni_server.model, + "messages": messages, + "stream": True, + "key_words": {"audio": AUDIO_KEY, "image": IMAGE_KEY, "video": VIDEO_KEY}, + } + openai_client.send_request(request_config, request_num=get_max_batch_size()) diff --git a/tests/e2e/stage_configs/qwen2_5_omni_ci.yaml b/tests/e2e/stage_configs/qwen2_5_omni_ci.yaml index f8655f77e..32b1ba15b 100644 --- a/tests/e2e/stage_configs/qwen2_5_omni_ci.yaml +++ b/tests/e2e/stage_configs/qwen2_5_omni_ci.yaml @@ -13,10 +13,10 @@ stage_args: model_arch: Qwen2_5OmniForConditionalGeneration worker_type: ar scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler - max_model_len: 2400 - max_num_batched_tokens: 2400 + max_model_len: 16384 + max_num_batched_tokens: 16384 max_num_seqs: 1 - gpu_memory_utilization: 0.8 + gpu_memory_utilization: 0.9 skip_mm_profiling: true enforce_eager: true # Now we only support eager mode trust_remote_code: true @@ -43,10 +43,10 @@ stage_args: model_arch: Qwen2_5OmniForConditionalGeneration worker_type: ar scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler - max_model_len: 2400 - max_num_batched_tokens: 2400 + max_model_len: 16384 + max_num_batched_tokens: 16384 max_num_seqs: 1 - gpu_memory_utilization: 0.8 + gpu_memory_utilization: 0.9 skip_mm_profiling: true enforce_eager: true trust_remote_code: true @@ -58,7 +58,7 @@ stage_args: temperature: 0.9 top_p: 0.8 top_k: 40 - max_tokens: 128 + max_tokens: 4096 seed: 42 detokenize: True repetition_penalty: 1.05 @@ -66,7 +66,7 @@ stage_args: - stage_id: 2 runtime: process: true - devices: "0" # Example: use a different GPU than the previous stage; use "0" if single GPU + devices: "2" # Example: use a different GPU than the previous stage; use "0" if single GPU max_batch_size: 1 engine_args: model_stage: code2wav @@ -78,7 +78,7 @@ stage_args: trust_remote_code: true enable_prefix_caching: false engine_output_type: audio - max_num_batched_tokens: 4069 + max_num_batched_tokens: 100000 engine_input_source: [1] final_output: true final_output_type: audio @@ -86,7 +86,7 @@ stage_args: temperature: 0.0 top_p: 1.0 top_k: -1 - max_tokens: 128 + max_tokens: 4096 seed: 42 detokenize: True repetition_penalty: 1.1 diff --git a/tests/examples/online_serving/test_qwen2_5_omni.py b/tests/examples/online_serving/test_qwen2_5_omni.py index 5a529662b..a4199f18e 100644 --- a/tests/examples/online_serving/test_qwen2_5_omni.py +++ b/tests/examples/online_serving/test_qwen2_5_omni.py @@ -12,7 +12,8 @@ from pathlib import Path import pytest -from tests.conftest import OmniServer, convert_audio_file_to_text, cosine_similarity_text +from tests.conftest import convert_audio_file_to_text, cosine_similarity_text +from tests.utils import hardware_test models = ["Qwen/Qwen2.5-Omni-7B"] @@ -21,26 +22,7 @@ stage_configs = [str(Path(__file__).parent.parent.parent / "e2e" / "stage_config example_dir = str(Path(__file__).parent.parent.parent.parent / "examples" / "online_serving" / "qwen2_5_omni") # Create parameter combinations for model and stage config -test_params = [(model, stage_config) for model in models for stage_config in stage_configs] - - -@pytest.fixture(scope="module") -def omni_server(request): - """Start vLLM-Omni server as a subprocess with actual model weights. - Uses module scope so the server starts only once per test module. - Multi-stage initialization can take 10-20+ minutes. - """ - model, stage_config_path = request.param - - print(f"Starting OmniServer with model: {model}") - print("This may take 10-20+ minutes for initialization...") - - with OmniServer( - model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"], port=8091 - ) as server: - print("OmniServer started successfully") - yield server - print("OmniServer stopped") +test_params = [(8091, model, stage_config) for model in models for stage_config in stage_configs] def run_cmd(command): @@ -67,6 +49,9 @@ def extract_content_after_keyword(keywords, text): return matches[0] +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_send_multimodal_request_001(omni_server) -> None: command = [ @@ -99,6 +84,9 @@ def test_send_multimodal_request_001(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_send_multimodal_request_002(omni_server) -> None: command = [ @@ -131,6 +119,9 @@ def test_send_multimodal_request_002(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_send_multimodal_request_003(omni_server) -> None: command = ["bash", os.path.join(example_dir, "run_curl_multimodal_generation.sh"), "mixed_modalities"] @@ -151,6 +142,9 @@ def test_send_multimodal_request_003(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_modality_control_001(omni_server) -> None: command = [ @@ -178,6 +172,9 @@ def test_modality_control_001(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_modality_control_002(omni_server) -> None: command = [ @@ -203,6 +200,9 @@ def test_modality_control_002(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_modality_control_003(omni_server) -> None: command = [ @@ -237,6 +237,9 @@ def test_modality_control_003(omni_server) -> None: @pytest.mark.skip(reason="There is a known issue with stream error.") +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "L4", "rocm": "MI325"}, num_cards={"cuda": 4, "rocm": 2}) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_stream_001(omni_server) -> None: command = [ diff --git a/tests/examples/online_serving/test_qwen3_omni.py b/tests/examples/online_serving/test_qwen3_omni.py index 675146ff5..c7354905e 100644 --- a/tests/examples/online_serving/test_qwen3_omni.py +++ b/tests/examples/online_serving/test_qwen3_omni.py @@ -12,7 +12,8 @@ from pathlib import Path import pytest -from tests.conftest import OmniServer, convert_audio_file_to_text, cosine_similarity_text +from tests.conftest import convert_audio_file_to_text, cosine_similarity_text +from tests.utils import hardware_test models = ["Qwen/Qwen3-Omni-30B-A3B-Instruct"] @@ -21,26 +22,7 @@ stage_configs = [str(Path(__file__).parent.parent.parent / "e2e" / "stage_config example_dir = str(Path(__file__).parent.parent.parent.parent / "examples" / "online_serving" / "qwen3_omni") # Create parameter combinations for model and stage config -test_params = [(model, stage_config) for model in models for stage_config in stage_configs] - - -@pytest.fixture(scope="module") -def omni_server(request): - """Start vLLM-Omni server as a subprocess with actual model weights. - Uses module scope so the server starts only once per test module.. - Multi-stage initialization can take 10-20+ minutes. - """ - model, stage_config_path = request.param - - print(f"Starting OmniServer with model: {model}") - print("This may take 10-20+ minutes for initialization...") - - with OmniServer( - model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"], port=8091 - ) as server: - print("OmniServer started successfully") - yield server - print("OmniServer stopped") +test_params = [(8091, model, stage_config) for model in models for stage_config in stage_configs] def run_cmd(command): @@ -67,6 +49,9 @@ def extract_content_after_keyword(keywords, text): return matches[0] +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_send_multimodal_request_001(omni_server) -> None: command = [ @@ -92,6 +77,9 @@ def test_send_multimodal_request_001(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_send_multimodal_request_002(omni_server) -> None: command = [ @@ -120,6 +108,9 @@ def test_send_multimodal_request_002(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_send_multimodal_request_003(omni_server) -> None: command = ["bash", os.path.join(example_dir, "run_curl_multimodal_generation.sh"), "use_image"] @@ -134,6 +125,9 @@ def test_send_multimodal_request_003(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_modality_control_001(omni_server) -> None: command = [ @@ -155,6 +149,9 @@ def test_modality_control_001(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_modality_control_002(omni_server) -> None: command = [ @@ -175,6 +172,9 @@ def test_modality_control_002(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_modality_control_003(omni_server) -> None: command = [ @@ -202,6 +202,9 @@ def test_modality_control_003(omni_server) -> None: # TODO: Verify the E2E latency after confirmation baseline. +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_stream_001(omni_server) -> None: command = [ diff --git a/tests/perf/scripts/run_benchmark.py b/tests/perf/scripts/run_benchmark.py index 5895fe2a1..688666162 100644 --- a/tests/perf/scripts/run_benchmark.py +++ b/tests/perf/scripts/run_benchmark.py @@ -46,7 +46,8 @@ def modify_stage(default_path, updates, deletes): def create_unique_server_params(configs: list[dict[str, Any]]) -> list[tuple[str, str, str]]: - unique_params = set() + unique_params = [] + seen = set() for config in configs: test_name = config["test_name"] model = config["server_params"]["model"] @@ -55,9 +56,13 @@ def create_unique_server_params(configs: list[dict[str, Any]]) -> list[tuple[str delete = config["server_params"].get("delete", None) update = config["server_params"].get("update", None) stage_config_path = modify_stage(stage_config_path, update, delete) - unique_params.add((test_name, model, stage_config_path)) - return list(unique_params) + server_param = (test_name, model, stage_config_path) + if server_param not in seen: + seen.add(server_param) + unique_params.append(server_param) + + return unique_params def create_test_parameter_mapping(configs: list[dict[str, Any]]) -> dict[str, dict]: @@ -95,6 +100,7 @@ def omni_server(request): print(f"Starting OmniServer with test: {test_name}, model: {model}") with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"]) as server: + server.test_name = test_name print("OmniServer started successfully") yield server print("OmniServer stopping...") @@ -150,10 +156,15 @@ def get_benchmark_params_for_server(test_name: str) -> list: def create_benchmark_indices(): indices = [] - for test_name, config_data in server_to_benchmark_mapping.items(): - params_list = config_data["benchmark_params"] - for idx in range(len(params_list)): - indices.append((test_name, idx)) + seen = set() + for config in BENCHMARK_CONFIGS: + test_name = config["test_name"] + if test_name not in seen: + seen.add(test_name) + params_list = get_benchmark_params_for_server(test_name) + for idx in range(len(params_list)): + indices.append((test_name, idx)) + return indices @@ -164,6 +175,10 @@ benchmark_indices = create_benchmark_indices() def benchmark_params(request, omni_server): """Benchmark parameters fixture with proper parametrization""" test_name, param_index = request.param + + if test_name != omni_server.test_name: + pytest.skip(f"Skipping parameter for {test_name} - current server is {omni_server.test_name}") + all_params = get_benchmark_params_for_server(test_name) if not all_params: @@ -172,6 +187,10 @@ def benchmark_params(request, omni_server): if param_index >= len(all_params): raise ValueError(f"No benchmark parameters found for index {param_index} in test: {test_name}") + current = param_index + 1 + total = len(all_params) + print(f"\n Running benchmark {current}/{total} for {test_name}") + return {"test_name": test_name, "params": all_params[param_index]}