anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into dev/rebase-v0.16.0

Browse Source
This commit is contained in:
tzhouam committed 2026-02-15 01:24:23 +00:00
2 parents 3051fb10b5 a42b7487b0
commit 93eeb1a4c8
8 files changed
+432 -376
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.buildkite/pipeline.yml
+105 -105
View File
@@ -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
.buildkite/test-nightly.yaml
+3 -3
View File
@@ -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:
tests/conftest.py
+10 -3
View File
@@ -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)
tests/e2e/online_serving/test_qwen3_omni_expansion.py
+230 -206
View File
@@ -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())
tests/e2e/stage_configs/qwen2_5_omni_ci.yaml
+10 -10
View File
@@ -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
tests/examples/online_serving/test_qwen2_5_omni.py
+24 -21
View File
@@ -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 = [
tests/examples/online_serving/test_qwen3_omni.py
+24 -21
View File
@@ -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 = [
tests/perf/scripts/run_benchmark.py
+26 -7
View File
@@ -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]}