[Test] Add precision test cases for Qwen3-Omni-30B-A3B-Instruct in CI (#828)
Signed-off-by: wangyu31577 <wangyu31577@hundsun.com> Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com> Signed-off-by: wangyu <53896905+yenuo26@users.noreply.github.com> Co-authored-by: wangyu31577 <wangyu31577@hundsun.com> Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
This commit is contained in:
8 files changed
+266
-370
No files matched your search
+6
-29
@@ -52,6 +52,10 @@ def default_vllm_config():
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clean_gpu_memory_between_tests():
|
||||
clean_gpu_memory()
|
||||
|
||||
|
||||
def clean_gpu_memory():
|
||||
if os.getenv("VLLM_TEST_CLEAN_GPU_MEMORY", "0") != "1":
|
||||
yield
|
||||
return
|
||||
@@ -528,7 +532,6 @@ def convert_audio_to_text(audio_data):
|
||||
temperature=0.0,
|
||||
word_timestamps=True,
|
||||
condition_on_previous_text=False,
|
||||
initial_prompt="Please transcribe with proper word spacing.",
|
||||
)["text"]
|
||||
if text:
|
||||
return text
|
||||
@@ -709,31 +712,5 @@ class OmniServer:
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if self.proc:
|
||||
try:
|
||||
parent = psutil.Process(self.proc.pid)
|
||||
children = parent.children(recursive=True)
|
||||
for child in children:
|
||||
try:
|
||||
child.terminate()
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
|
||||
gone, still_alive = psutil.wait_procs(children, timeout=10)
|
||||
|
||||
for child in still_alive:
|
||||
try:
|
||||
child.kill()
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
|
||||
try:
|
||||
parent.terminate()
|
||||
parent.wait(timeout=10)
|
||||
except (psutil.NoSuchProcess, psutil.TimeoutExpired):
|
||||
try:
|
||||
parent.kill()
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
self._kill_process_tree(self.proc.pid)
|
||||
clean_gpu_memory()
|
||||
@@ -10,6 +10,7 @@ import pytest
|
||||
from vllm.distributed.parallel_state import cleanup_dist_env_and_memory
|
||||
from vllm.sampling_params import SamplingParams
|
||||
|
||||
from tests.conftest import clean_gpu_memory
|
||||
from vllm_omni.entrypoints.omni import Omni
|
||||
from vllm_omni.outputs import OmniRequestOutput
|
||||
|
||||
@@ -335,6 +336,7 @@ class OmniRunner:
|
||||
self.close()
|
||||
del self.omni
|
||||
cleanup_dist_env_and_memory()
|
||||
clean_gpu_memory()
|
||||
|
||||
def close(self):
|
||||
"""Close and cleanup the Omni instance."""
|
||||
|
||||
@@ -40,42 +40,49 @@ def test_cache_dit(model_name: str):
|
||||
"residual_diff_threshold": 0.24,
|
||||
"max_continuous_cached_steps": 3,
|
||||
}
|
||||
m = None
|
||||
try:
|
||||
m = Omni(
|
||||
model=model_name,
|
||||
cache_backend="cache_dit",
|
||||
cache_config=cache_config,
|
||||
)
|
||||
|
||||
m = Omni(
|
||||
model=model_name,
|
||||
cache_backend="cache_dit",
|
||||
cache_config=cache_config,
|
||||
)
|
||||
# Use minimal settings for fast testing
|
||||
height = 256
|
||||
width = 256
|
||||
num_inference_steps = 4 # Minimal steps for fast test
|
||||
|
||||
# Use minimal settings for fast testing
|
||||
height = 256
|
||||
width = 256
|
||||
num_inference_steps = 4 # Minimal steps for fast test
|
||||
outputs = m.generate(
|
||||
"a photo of a cat sitting on a laptop keyboard",
|
||||
height=height,
|
||||
width=width,
|
||||
num_inference_steps=num_inference_steps,
|
||||
guidance_scale=0.0,
|
||||
generator=torch.Generator("cuda").manual_seed(42),
|
||||
num_outputs_per_prompt=1, # Single output for speed
|
||||
)
|
||||
# Extract images from request_output[0]['images']
|
||||
first_output = outputs[0]
|
||||
assert first_output.final_output_type == "image"
|
||||
if not hasattr(first_output, "request_output") or not first_output.request_output:
|
||||
raise ValueError("No request_output found in OmniRequestOutput")
|
||||
|
||||
outputs = m.generate(
|
||||
"a photo of a cat sitting on a laptop keyboard",
|
||||
height=height,
|
||||
width=width,
|
||||
num_inference_steps=num_inference_steps,
|
||||
guidance_scale=0.0,
|
||||
generator=torch.Generator("cuda").manual_seed(42),
|
||||
num_outputs_per_prompt=1, # Single output for speed
|
||||
)
|
||||
# Extract images from request_output[0]['images']
|
||||
first_output = outputs[0]
|
||||
assert first_output.final_output_type == "image"
|
||||
if not hasattr(first_output, "request_output") or not first_output.request_output:
|
||||
raise ValueError("No request_output found in OmniRequestOutput")
|
||||
req_out = first_output.request_output[0]
|
||||
if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"):
|
||||
raise ValueError("Invalid request_output structure or missing 'images' key")
|
||||
|
||||
req_out = first_output.request_output[0]
|
||||
if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"):
|
||||
raise ValueError("Invalid request_output structure or missing 'images' key")
|
||||
images = req_out.images
|
||||
|
||||
images = req_out.images
|
||||
|
||||
# Verify generation succeeded
|
||||
assert images is not None
|
||||
assert len(images) == 1
|
||||
# Check image size
|
||||
assert images[0].width == width
|
||||
assert images[0].height == height
|
||||
# Verify generation succeeded
|
||||
assert images is not None
|
||||
assert len(images) == 1
|
||||
# Check image size
|
||||
assert images[0].width == width
|
||||
assert images[0].height == height
|
||||
except Exception as e:
|
||||
print(f"Test failed with error: {e}")
|
||||
raise
|
||||
finally:
|
||||
if m is not None and hasattr(m, "close"):
|
||||
m.close()
|
||||
@@ -34,33 +34,41 @@ elif is_rocm():
|
||||
|
||||
@pytest.mark.parametrize("model_name", models)
|
||||
def test_diffusion_model(model_name: str):
|
||||
m = Omni(model=model_name)
|
||||
# high resolution may cause OOM on L4
|
||||
height = 256
|
||||
width = 256
|
||||
outputs = m.generate(
|
||||
"a photo of a cat sitting on a laptop keyboard",
|
||||
height=height,
|
||||
width=width,
|
||||
num_inference_steps=2,
|
||||
guidance_scale=0.0,
|
||||
generator=torch.Generator("cuda").manual_seed(42),
|
||||
num_outputs_per_prompt=2,
|
||||
)
|
||||
# Extract images from request_output[0]['images']
|
||||
first_output = outputs[0]
|
||||
assert first_output.final_output_type == "image"
|
||||
if not hasattr(first_output, "request_output") or not first_output.request_output:
|
||||
raise ValueError("No request_output found in OmniRequestOutput")
|
||||
m = None
|
||||
try:
|
||||
m = Omni(model=model_name)
|
||||
# high resolution may cause OOM on L4
|
||||
height = 256
|
||||
width = 256
|
||||
outputs = m.generate(
|
||||
"a photo of a cat sitting on a laptop keyboard",
|
||||
height=height,
|
||||
width=width,
|
||||
num_inference_steps=2,
|
||||
guidance_scale=0.0,
|
||||
generator=torch.Generator("cuda").manual_seed(42),
|
||||
num_outputs_per_prompt=2,
|
||||
)
|
||||
# Extract images from request_output[0]['images']
|
||||
first_output = outputs[0]
|
||||
assert first_output.final_output_type == "image"
|
||||
if not hasattr(first_output, "request_output") or not first_output.request_output:
|
||||
raise ValueError("No request_output found in OmniRequestOutput")
|
||||
|
||||
req_out = first_output.request_output[0]
|
||||
if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"):
|
||||
raise ValueError("Invalid request_output structure or missing 'images' key")
|
||||
req_out = first_output.request_output[0]
|
||||
if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"):
|
||||
raise ValueError("Invalid request_output structure or missing 'images' key")
|
||||
|
||||
images = req_out.images
|
||||
images = req_out.images
|
||||
|
||||
assert len(images) == 2
|
||||
# check image size
|
||||
assert images[0].width == width
|
||||
assert images[0].height == height
|
||||
images[0].save("image_output.png")
|
||||
assert len(images) == 2
|
||||
# check image size
|
||||
assert images[0].width == width
|
||||
assert images[0].height == height
|
||||
images[0].save("image_output.png")
|
||||
except Exception as e:
|
||||
print(f"Test failed with error: {e}")
|
||||
raise
|
||||
finally:
|
||||
if m is not None and hasattr(m, "close"):
|
||||
m.close()
|
||||
@@ -36,42 +36,49 @@ def test_teacache(model_name: str):
|
||||
cache_config = {
|
||||
"rel_l1_thresh": 0.2, # Default threshold
|
||||
}
|
||||
m = None
|
||||
try:
|
||||
m = Omni(
|
||||
model=model_name,
|
||||
cache_backend="tea_cache",
|
||||
cache_config=cache_config,
|
||||
)
|
||||
|
||||
m = Omni(
|
||||
model=model_name,
|
||||
cache_backend="tea_cache",
|
||||
cache_config=cache_config,
|
||||
)
|
||||
# Use minimal settings for fast testing
|
||||
height = 256
|
||||
width = 256
|
||||
num_inference_steps = 4 # Minimal steps for fast test
|
||||
|
||||
# Use minimal settings for fast testing
|
||||
height = 256
|
||||
width = 256
|
||||
num_inference_steps = 4 # Minimal steps for fast test
|
||||
outputs = m.generate(
|
||||
"a photo of a cat sitting on a laptop keyboard",
|
||||
height=height,
|
||||
width=width,
|
||||
num_inference_steps=num_inference_steps,
|
||||
guidance_scale=0.0,
|
||||
generator=torch.Generator("cuda").manual_seed(42),
|
||||
num_outputs_per_prompt=1, # Single output for speed
|
||||
)
|
||||
# Extract images from request_output[0]['images']
|
||||
first_output = outputs[0]
|
||||
assert first_output.final_output_type == "image"
|
||||
if not hasattr(first_output, "request_output") or not first_output.request_output:
|
||||
raise ValueError("No request_output found in OmniRequestOutput")
|
||||
|
||||
outputs = m.generate(
|
||||
"a photo of a cat sitting on a laptop keyboard",
|
||||
height=height,
|
||||
width=width,
|
||||
num_inference_steps=num_inference_steps,
|
||||
guidance_scale=0.0,
|
||||
generator=torch.Generator("cuda").manual_seed(42),
|
||||
num_outputs_per_prompt=1, # Single output for speed
|
||||
)
|
||||
# Extract images from request_output[0]['images']
|
||||
first_output = outputs[0]
|
||||
assert first_output.final_output_type == "image"
|
||||
if not hasattr(first_output, "request_output") or not first_output.request_output:
|
||||
raise ValueError("No request_output found in OmniRequestOutput")
|
||||
req_out = first_output.request_output[0]
|
||||
if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"):
|
||||
raise ValueError("Invalid request_output structure or missing 'images' key")
|
||||
|
||||
req_out = first_output.request_output[0]
|
||||
if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"):
|
||||
raise ValueError("Invalid request_output structure or missing 'images' key")
|
||||
images = req_out.images
|
||||
|
||||
images = req_out.images
|
||||
|
||||
# Verify generation succeeded
|
||||
assert images is not None
|
||||
assert len(images) == 1
|
||||
# Check image size
|
||||
assert images[0].width == width
|
||||
assert images[0].height == height
|
||||
# Verify generation succeeded
|
||||
assert images is not None
|
||||
assert len(images) == 1
|
||||
# Check image size
|
||||
assert images[0].width == width
|
||||
assert images[0].height == height
|
||||
except Exception as e:
|
||||
print(f"Test failed with error: {e}")
|
||||
raise
|
||||
finally:
|
||||
if m is not None and hasattr(m, "close"):
|
||||
m.close()
|
||||
@@ -8,21 +8,22 @@ import os
|
||||
|
||||
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
|
||||
|
||||
import base64
|
||||
import concurrent.futures
|
||||
import ctypes
|
||||
import signal
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import openai
|
||||
import pytest
|
||||
from vllm.assets.video import VideoAsset
|
||||
from vllm.utils.network_utils import get_open_port
|
||||
|
||||
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,
|
||||
)
|
||||
from vllm_omni.utils import is_rocm
|
||||
|
||||
models = ["Qwen/Qwen3-Omni-30B-A3B-Instruct"]
|
||||
@@ -32,114 +33,27 @@ if is_rocm():
|
||||
# ROCm stage config optimized for MI325 GPU
|
||||
stage_configs = [str(Path(__file__).parent / "stage_configs" / "rocm" / "qwen3_omni_ci.yaml")]
|
||||
else:
|
||||
stage_configs = [str(Path(__file__).parent / "stage_configs" / "qwen3_omni_ci.yaml")]
|
||||
stage_configs = [str(Path(__file__).parent.parent / "stage_configs" / "qwen3_omni_ci.yaml")]
|
||||
|
||||
# Create parameter combinations for model and stage config
|
||||
test_params = [(model, stage_config) for model in models for stage_config in stage_configs]
|
||||
|
||||
|
||||
class OmniServer:
|
||||
"""Omniserver for vLLM-Omni tests."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
model: str,
|
||||
serve_args: list[str],
|
||||
*,
|
||||
env_dict: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
self.model = model
|
||||
self.serve_args = serve_args
|
||||
self.env_dict = env_dict
|
||||
self.proc: subprocess.Popen | None = None
|
||||
self.host = "127.0.0.1"
|
||||
self.port = get_open_port()
|
||||
|
||||
def _start_server(self) -> None:
|
||||
"""Start the vLLM-Omni server subprocess."""
|
||||
env = os.environ.copy()
|
||||
env["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
|
||||
if self.env_dict is not None:
|
||||
env.update(self.env_dict)
|
||||
|
||||
cmd = [
|
||||
sys.executable,
|
||||
"-m",
|
||||
"vllm_omni.entrypoints.cli.main",
|
||||
"serve",
|
||||
self.model,
|
||||
"--omni",
|
||||
"--host",
|
||||
self.host,
|
||||
"--port",
|
||||
str(self.port),
|
||||
] + self.serve_args
|
||||
|
||||
# Helper to ensure child process dies when parent dies
|
||||
libc = ctypes.CDLL("libc.so.6")
|
||||
|
||||
def preexec_fn():
|
||||
# Ensure the child process receives SIGTERM when the parent (this test runner) dies.
|
||||
# This prevents orphaned processes if the test is killed unexpectedly.
|
||||
PR_SET_PDEATHSIG = 1
|
||||
libc.prctl(PR_SET_PDEATHSIG, signal.SIGTERM)
|
||||
|
||||
print(f"Launching OmniServer with: {' '.join(cmd)}")
|
||||
self.proc = subprocess.Popen(
|
||||
cmd,
|
||||
env=env,
|
||||
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), # Set working directory to vllm-omni root
|
||||
start_new_session=True,
|
||||
preexec_fn=preexec_fn,
|
||||
)
|
||||
|
||||
# Wait for server to be ready
|
||||
max_wait = 600 # 10 minutes
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < max_wait:
|
||||
try:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(1)
|
||||
result = sock.connect_ex((self.host, self.port))
|
||||
if result == 0:
|
||||
print(f"Server ready on {self.host}:{self.port}")
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(2)
|
||||
|
||||
raise RuntimeError(f"Server failed to start within {max_wait} seconds")
|
||||
|
||||
def __enter__(self):
|
||||
self._start_server()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if self.proc:
|
||||
try:
|
||||
os.killpg(self.proc.pid, signal.SIGTERM)
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
|
||||
try:
|
||||
self.proc.wait(timeout=30)
|
||||
except subprocess.TimeoutExpired:
|
||||
try:
|
||||
os.killpg(self.proc.pid, signal.SIGKILL)
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
self.proc.wait()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@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 for all tests.
|
||||
Uses session scope so the server starts only once for the entire test session.
|
||||
Multi-stage initialization can take 10-20+ minutes.
|
||||
"""
|
||||
model, stage_config_path = request.param
|
||||
with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "90"]) as server:
|
||||
|
||||
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"]) as server:
|
||||
print("OmniServer started successfully")
|
||||
yield server
|
||||
print("OmniServer stopped")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -151,15 +65,6 @@ def client(omni_server):
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def base64_encoded_video() -> str:
|
||||
"""Base64 encoded video for testing."""
|
||||
video = VideoAsset(name="baby_reading", num_frames=4)
|
||||
with open(video.video_path, "rb") as f:
|
||||
content = f.read()
|
||||
return base64.b64encode(content).decode("utf-8")
|
||||
|
||||
|
||||
def get_system_prompt():
|
||||
return {
|
||||
"role": "system",
|
||||
@@ -193,71 +98,50 @@ def dummy_messages_from_video_data(
|
||||
]
|
||||
|
||||
|
||||
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.",
|
||||
}
|
||||
return prompts.get(prompt_type, prompts["text_only"])
|
||||
|
||||
|
||||
def get_max_batch_size(size_type="few"):
|
||||
batch_sizes = {"few": 5, "medium": 100, "large": 256}
|
||||
return batch_sizes.get(size_type, 5)
|
||||
|
||||
|
||||
@pytest.mark.skipif(is_rocm(), reason="Test skipped on AMD environment due to known output issues")
|
||||
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
|
||||
def test_video_to_audio_concurrent(
|
||||
client: openai.OpenAI,
|
||||
omni_server,
|
||||
base64_encoded_video: str,
|
||||
) -> None:
|
||||
"""Test processing video with multiple concurrent completions, generating audio output via OpenAI API."""
|
||||
# Create data URL for the base64 encoded video
|
||||
video_data_url = f"data:video/mp4;base64,{base64_encoded_video}"
|
||||
def test_mix_to_text_audio_001(client: openai.OpenAI, omni_server) -> None:
|
||||
"""
|
||||
Test multi-modal input processing and text/audio output generation via OpenAI API.
|
||||
Deploy Setting: default yaml
|
||||
Input Modal: text + audio + video + image
|
||||
Output Modal: text + audio
|
||||
Input Setting: stream=True
|
||||
Datasets: single request
|
||||
"""
|
||||
|
||||
messages = dummy_messages_from_video_data(video_data_url)
|
||||
|
||||
# Test multiple concurrent completions
|
||||
num_concurrent_requests = 5
|
||||
|
||||
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)
|
||||
]
|
||||
|
||||
# Wait for all requests to complete and collect results
|
||||
chat_completions = [future.result() for future in concurrent.futures.as_completed(futures)]
|
||||
|
||||
# Verify all completions succeeded
|
||||
assert len(chat_completions) == num_concurrent_requests
|
||||
|
||||
for chat_completion in chat_completions:
|
||||
assert len(chat_completion.choices) == 2 # 1 for text output, 1 for audio output
|
||||
|
||||
# Verify text output
|
||||
text_choice = chat_completion.choices[0]
|
||||
assert text_choice.finish_reason == "length"
|
||||
|
||||
# Verify we got a response
|
||||
text_message = text_choice.message
|
||||
assert text_message.content is not None and len(text_message.content) >= 10
|
||||
assert text_message.role == "assistant"
|
||||
|
||||
# Verify audio output
|
||||
audio_choice = chat_completion.choices[1]
|
||||
assert audio_choice.finish_reason == "stop"
|
||||
audio_message = audio_choice.message
|
||||
|
||||
# Check if audio was generated
|
||||
if hasattr(audio_message, "audio") and audio_message.audio:
|
||||
assert audio_message.audio.data is not None
|
||||
assert len(audio_message.audio.data) > 0
|
||||
|
||||
# Test streaming completion
|
||||
chat_completion = client.chat.completions.create(
|
||||
model=omni_server.model,
|
||||
messages=messages,
|
||||
stream=True,
|
||||
# Test single completion
|
||||
e2e_list = list()
|
||||
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"),
|
||||
)
|
||||
|
||||
# Collect text and audio data from stream
|
||||
# Test single completion
|
||||
start_time = time.perf_counter()
|
||||
chat_completion = client.chat.completions.create(model=omni_server.model, messages=messages, stream=True)
|
||||
|
||||
text_content = ""
|
||||
audio_data = None
|
||||
|
||||
for chunk in chat_completion:
|
||||
for choice in chunk.choices:
|
||||
if hasattr(choice, "delta"):
|
||||
@@ -268,18 +152,92 @@ def test_video_to_audio_concurrent(
|
||||
modality = getattr(chunk, "modality", None)
|
||||
|
||||
if modality == "audio" and content:
|
||||
# Audio chunk - decode base64 content
|
||||
# Audio chunk - content
|
||||
if audio_data is None:
|
||||
audio_data = base64.b64decode(content)
|
||||
audio_data = content
|
||||
else:
|
||||
audio_data += base64.b64decode(content)
|
||||
audio_data += content
|
||||
elif modality == "text" and content:
|
||||
# Text chunk - accumulate text content
|
||||
text_content += content if content else ""
|
||||
|
||||
# Verify text output
|
||||
assert text_content is not None and len(text_content) >= 2
|
||||
# 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)
|
||||
|
||||
# Verify audio output
|
||||
assert audio_data is not None
|
||||
assert len(audio_data) > 0
|
||||
print(f"the avg e2e is: {sum(e2e_list) / len(e2e_list)}")
|
||||
# Verify all completions succeeded
|
||||
assert audio_data is not None, "No audio output is generated"
|
||||
|
||||
# Verify text output success
|
||||
assert text_content is not None and len(text_content) >= 2, "No text output is generated"
|
||||
assert any(
|
||||
keyword in text_content.lower() for keyword in ["square", "quadrate", "sphere", "globe", "circle", "round"]
|
||||
), "The output does not contain any of the 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.lower(), text_content.lower())
|
||||
print(f"similarity is: {similarity}")
|
||||
assert similarity > 0.9, "The audio content is not same as the text"
|
||||
|
||||
|
||||
@pytest.mark.skipif(is_rocm(), reason="Test skipped on AMD environment due to known output issues")
|
||||
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
|
||||
def test_text_to_text_audio_001(client: openai.OpenAI, omni_server) -> None:
|
||||
"""
|
||||
Test text input processing and text/audio output generation via OpenAI API.
|
||||
Deploy Setting: default yaml
|
||||
Input Modal: text
|
||||
Output Modal: text + audio
|
||||
Datasets: few requests
|
||||
"""
|
||||
|
||||
num_concurrent_requests = get_max_batch_size()
|
||||
messages = dummy_messages_from_mix_data(system_prompt=get_system_prompt(), content_text=get_prompt())
|
||||
|
||||
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"
|
||||
assert "beijing" in text_choice.message.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.lower(), text_content.lower())
|
||||
print(f"similarity is: {similarity}")
|
||||
assert similarity > 0.9, "The audio content is not same as the text"
|
||||
@@ -111,69 +111,6 @@ def test_text_to_text_001(test_config: tuple[str, str]) -> None:
|
||||
assert "beijing" in text_choice.message.content.lower(), "The output do not contain keywords."
|
||||
|
||||
|
||||
@pytest.mark.parametrize("test_config", test_params)
|
||||
def test_text_to_text_audio_001(test_config: tuple[str, str]) -> None:
|
||||
"""Test processing text, generating text and audio output via OpenAI API."""
|
||||
|
||||
model, stage_config_path = test_config
|
||||
num_concurrent_requests = get_max_batch_size()
|
||||
stage_config_path = modify_stage_config(
|
||||
stage_config_path,
|
||||
{
|
||||
0: {"runtime.max_batch_size": num_concurrent_requests},
|
||||
1: {"runtime.max_batch_size": num_concurrent_requests},
|
||||
},
|
||||
)
|
||||
with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "90"]) as server:
|
||||
messages = dummy_messages_from_mix_data(
|
||||
system_prompt=get_system_prompt(), content_text="What is the capital of China?"
|
||||
)
|
||||
|
||||
# Test single completion
|
||||
api_client = client(server)
|
||||
e2e_list = list()
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=num_concurrent_requests) as executor:
|
||||
# Submit multiple completion requests concurrently
|
||||
futures = [
|
||||
executor.submit(api_client.chat.completions.create, model=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"
|
||||
assert "beijing" in text_choice.message.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}")
|
||||
assert cosine_similarity_text(audio_content.lower(), text_content.lower()) > 0.9, (
|
||||
"The audio content is not same as the text"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("test_config", test_params)
|
||||
def test_audio_to_text_001(test_config: tuple[str, str]) -> None:
|
||||
"""Test processing text, generating text output via OpenAI API."""
|
||||
|
||||
@@ -8,7 +8,7 @@ stage_args:
|
||||
- stage_id: 0
|
||||
runtime:
|
||||
devices: "0,1"
|
||||
max_batch_size: 1
|
||||
max_batch_size: 5
|
||||
engine_args:
|
||||
model_stage: thinker
|
||||
model_arch: Qwen3OmniMoeForConditionalGeneration
|
||||
@@ -39,7 +39,7 @@ stage_args:
|
||||
- stage_id: 1
|
||||
runtime:
|
||||
devices: "1"
|
||||
max_batch_size: 1
|
||||
max_batch_size: 5
|
||||
engine_args:
|
||||
model_stage: talker
|
||||
model_arch: Qwen3OmniMoeForConditionalGeneration
|
||||
|
||||
Reference in new issue
Block a user