[Test] Add full test for Qwen3-Omni-30B-A3B-Instruct for image and audio single modal (#827)

Signed-off-by: wangyu31577 <wangyu31577@hundsun.com>
Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>
Co-authored-by: wangyu31577 <wangyu31577@hundsun.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
This commit is contained in:
authored and GitHub committed 2026-01-21 13:04:46 +08:00
1 parent d2a72f380f
commit 334d306d9a
4 files changed
+617 -22

No files matched your search

+4 -1
View File
@@ -52,7 +52,10 @@ dev = [
"mypy==1.11.1",
"pre-commit==4.0.1",
"openai-whisper>=20250625",
"psutil>=7.2.0"
"psutil>=7.2.0",
"soundfile>=0.13.1",
"imageio[ffmpeg]>=0.6.0",
"opencv-python>=4.12.0.88"
]
docs = [
+398 -17
View File
@@ -1,11 +1,15 @@
import base64
import datetime
import io
import math
import os
import random
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
# Set CPU device for CI environments without GPU
if "VLLM_TARGET_DEVICE" not in os.environ:
os.environ["VLLM_TARGET_DEVICE"] = "cpu"
import base64
import socket
import subprocess
import sys
@@ -13,10 +17,10 @@ import time
from pathlib import Path
from typing import Any
import numpy as np
import psutil
import pytest
import torch
import whisper
import yaml
from vllm.logger import init_logger
from vllm.utils.network_utils import get_open_port
@@ -119,27 +123,397 @@ def dummy_messages_from_mix_data(
return messages
def cosine_similarity_text(s1, s2):
"""
Calculate cosine similarity between two text strings.
Notes:
------
- Higher score means more similar texts
- Score of 1.0 means identical word composition (bag-of-words)
- Score of 0.0 means completely different vocabulary
"""
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.metrics.pairwise import cosine_similarity
def generate_synthetic_audio(
duration: int, # seconds
num_channels: int, # 1:Mono,2:Stereo 5:5.1 surround sound
sample_rate: int = 48000, # Default use 48000Hz.
save_to_file: bool = False,
) -> dict[str, Any]:
""" "Generate synthetic audio with rain."""
import soundfile as sf
vectorizer = CountVectorizer().fit_transform([s1, s2])
vectors = vectorizer.toarray()
return cosine_similarity([vectors[0]], [vectors[1]])[0][0]
# Initialize audio data array
num_samples = int(sample_rate * duration)
audio_data = np.zeros((num_samples, num_channels), dtype=np.float32)
# Configure parameters based on rain intensity
drop_density = 10 # Number of raindrops per second
drop_volume = 0.15 # Volume of individual raindrops
background_volume = 0.02 # Volume of background rain noise
# Pink noise sounds more natural than white noise for rain
white_noise = np.random.randn(num_samples)
pink_noise = np.convolve(white_noise, np.ones(8) / 8, mode="same")
pink_noise = pink_noise / np.max(np.abs(pink_noise)) if np.max(np.abs(pink_noise)) > 0 else pink_noise
bg_noise = pink_noise * background_volume
# Add background noise to all channels
for ch in range(num_channels):
audio_data[:, ch] += bg_noise
# Total number of raindrops = density × duration × channels for stereo effect
total_drops = int(drop_density * duration * num_channels)
for _ in range(total_drops):
# Random timing for raindrop
drop_time = random.uniform(0, duration)
# Random duration of raindrop sound (0.01-0.05 seconds)
drop_duration = random.uniform(0.01, 0.05)
# Random frequency gives variation in raindrop pitch
drop_freq = random.uniform(500, 5000) # Hz
# Random channel selection for stereo positioning
channel = random.randint(0, num_channels - 1)
# Calculate sample positions for this raindrop
start_sample = int(drop_time * sample_rate)
drop_samples = int(drop_duration * sample_rate)
end_sample = min(start_sample + drop_samples, num_samples)
if start_sample < end_sample:
# Generate the raindrop sound
num_drop_samples = end_sample - start_sample
t = np.arange(num_drop_samples) / sample_rate
# Basic sine wave for raindrop sound
drop_sound = drop_volume * np.sin(2 * math.pi * drop_freq * t)
# Apply envelope for natural attack and decay
envelope = np.ones(num_drop_samples)
attack_samples = int(num_drop_samples * 0.1) # 10% of samples for attack
decay_samples = num_drop_samples - attack_samples
if attack_samples > 0:
# Linear attack: volume increases from 0 to 1
envelope[:attack_samples] = np.linspace(0, 1, attack_samples)
if decay_samples > 0:
# Exponential decay for natural sound fade
decay = np.exp(-8 * t[attack_samples:] / drop_duration)
envelope[attack_samples:] = decay
# Apply envelope to raindrop sound
drop_sound *= envelope
# Add raindrop sound to selected channel
audio_data[start_sample:end_sample, channel] += drop_sound
# Step 3: Add simple reverb effect for realism
# Reverb simulates sound reflections in environment
if duration > 2:
# Single delay reverb (100ms delay)
delay_samples = int(0.1 * sample_rate)
if delay_samples < num_samples:
for ch in range(num_channels):
delayed = np.zeros(num_samples)
delayed[delay_samples:] = audio_data[:-delay_samples, ch] * 0.3
audio_data[:, ch] += delayed
# Step 4: Normalize audio to prevent clipping
# Find maximum amplitude and scale to 80% of maximum volume
max_amp = np.max(np.abs(audio_data))
if max_amp > 0:
audio_data = audio_data / max_amp * 0.8
# Handle file saving
audio_bytes = None
if save_to_file:
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = f"audio_{num_channels}ch_{timestamp}.wav"
try:
sf.write(output_path, audio_data, sample_rate, format="WAV", subtype="PCM_16")
print(f"Audio saved: {output_path}")
with open(output_path, "rb") as f:
audio_bytes = f.read()
except Exception as e:
print(f"Save failed: {e}")
save_to_file = False
# If not saving or save failed, create in memory
if not save_to_file or audio_bytes is None:
buffer = io.BytesIO()
sf.write(buffer, audio_data, sample_rate, format="WAV", subtype="PCM_16")
buffer.seek(0)
audio_bytes = buffer.read()
# Return result
base64_audio = base64.b64encode(audio_bytes).decode("utf-8")
result = {
"base64": base64_audio,
}
if save_to_file and output_path:
result["file_path"] = output_path
return result
def generate_synthetic_video(width: int, height: int, num_frames: int, save_to_file: bool = False) -> str:
"""Generate synthetic video with bouncing balls and return base64 string."""
import cv2
import imageio
# Create random balls
num_balls = random.randint(3, 8)
balls = []
for _ in range(num_balls):
radius = min(width, height) // 8
if radius < 1:
raise ValueError(f"Video dimensions ({width}x{height}) are too small for synthetic video generation")
x = random.randint(radius, width - radius)
y = random.randint(radius, height - radius)
speed = random.uniform(3.0, 8.0)
angle = random.uniform(0, 2 * math.pi)
vx = speed * math.cos(angle)
vy = speed * math.sin(angle)
# OpenCV uses BGR format, but imageio expects RGB
# We'll create in BGR first, then convert to RGB later
color_bgr = (random.randint(50, 255), random.randint(50, 255), random.randint(50, 255))
balls.append({"x": x, "y": y, "vx": vx, "vy": vy, "radius": radius, "color_bgr": color_bgr})
# Generate video frames
video_frames = []
for frame_idx in range(num_frames):
# Create black background (BGR format)
frame_bgr = np.zeros((height, width, 3), dtype=np.uint8)
for ball in balls:
# Update position
ball["x"] += ball["vx"]
ball["y"] += ball["vy"]
# Boundary collision detection
if ball["x"] - ball["radius"] <= 0 or ball["x"] + ball["radius"] >= width:
ball["vx"] = -ball["vx"]
ball["x"] = max(ball["radius"], min(width - ball["radius"], ball["x"]))
if ball["y"] - ball["radius"] <= 0 or ball["y"] + ball["radius"] >= height:
ball["vy"] = -ball["vy"]
ball["y"] = max(ball["radius"], min(height - ball["radius"], ball["y"]))
# Use cv2 to draw circle
x, y = int(ball["x"]), int(ball["y"])
radius = ball["radius"]
# Draw solid circle (main circle)
cv2.circle(frame_bgr, (x, y), radius, ball["color_bgr"], -1)
# Add simple 3D effect: draw a brighter center
if radius > 3: # Only add highlight when radius is large enough
highlight_radius = max(1, radius // 2)
highlight_x = max(highlight_radius, min(x - radius // 4, width - highlight_radius))
highlight_y = max(highlight_radius, min(y - radius // 4, height - highlight_radius))
# Create highlight color (brighter)
highlight_color = tuple(min(c + 40, 255) for c in ball["color_bgr"])
cv2.circle(frame_bgr, (highlight_x, highlight_y), highlight_radius, highlight_color, -1)
# Convert BGR to RGB for imageio
frame_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
video_frames.append(frame_rgb)
video_bytes = None
saved_file_path = None
buffer = io.BytesIO()
writer_kwargs = {
"format": "mp4",
"fps": 30,
"codec": "libx264",
"quality": 7,
"pixelformat": "yuv420p",
"macro_block_size": 16,
"ffmpeg_params": [
"-preset",
"medium",
"-crf",
"23",
"-movflags",
"+faststart",
"-pix_fmt",
"yuv420p",
"-vf",
f"scale={width}:{height}",
],
}
if save_to_file:
import datetime
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = f"video_{width}x{height}_{timestamp}.mp4"
try:
with imageio.get_writer(output_path, **writer_kwargs) as writer:
for frame in video_frames:
writer.append_data(frame)
saved_file_path = output_path
print(f"Video saved to: {saved_file_path}")
with open(output_path, "rb") as f:
video_bytes = f.read()
except Exception as e:
print(f"Warning: Failed to save video to file {output_path}: {e}")
save_to_file = False
if not save_to_file or video_bytes is None:
with imageio.get_writer(buffer, **writer_kwargs) as writer:
for frame in video_frames:
writer.append_data(frame)
buffer.seek(0)
video_bytes = buffer.read()
base64_video = base64.b64encode(video_bytes).decode("utf-8")
result = {
"base64": base64_video,
}
if save_to_file and saved_file_path:
result["file_path"] = saved_file_path
return result
def generate_synthetic_image(width: int, height: int, save_to_file: bool = False) -> Any:
"""Generate synthetic image with randomly colored squares and return base64 string."""
from PIL import Image, ImageDraw
# Create white background
image = Image.new("RGB", (width, height), (255, 255, 255))
draw = ImageDraw.Draw(image)
# Generate random number of squares
num_squares = random.randint(3, 8)
for _ in range(num_squares):
# Random square size
square_size = random.randint(min(width, height) // 8, min(width, height) // 4)
# Random position
x = random.randint(0, width - square_size - 1)
y = random.randint(0, height - square_size - 1)
# Random color
color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255))
# Random border width
border_width = random.randint(1, 5)
# Draw square
draw.rectangle([x, y, x + square_size, y + square_size], fill=color, outline=(0, 0, 0), width=border_width)
# Handle file saving
image_bytes = None
saved_file_path = None
if save_to_file:
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = f"image_{width}x{height}_{timestamp}.jpg"
try:
# Save image to file
image.save(output_path, format="JPEG", quality=85, optimize=True)
saved_file_path = output_path
print(f"Image saved to: {saved_file_path}")
# Read file for base64 encoding
with open(output_path, "rb") as f:
image_bytes = f.read()
except Exception as e:
print(f"Warning: Failed to save image to file {output_path}: {e}")
save_to_file = False
# If not saving or save failed, create in memory
if not save_to_file or image_bytes is None:
buffer = io.BytesIO()
image.save(buffer, format="JPEG", quality=85, optimize=True)
buffer.seek(0)
image_bytes = buffer.read()
# Generate base64
base64_image = base64.b64encode(image_bytes).decode("utf-8")
# Return result
result = {
"base64": base64_image,
}
if save_to_file and saved_file_path:
result["file_path"] = saved_file_path
return result
def preprocess_text(text):
import re
word_to_num = {
"zero": "0",
"one": "1",
"two": "2",
"three": "3",
"four": "4",
"five": "5",
"six": "6",
"seven": "7",
"eight": "8",
"nine": "9",
"ten": "10",
}
for word, num in word_to_num.items():
pattern = r"\b" + re.escape(word) + r"\b"
text = re.sub(pattern, num, text, flags=re.IGNORECASE)
text = re.sub(r"[^\w\s]", "", text)
text = re.sub(r"\s+", " ", text)
return text.lower().strip()
def cosine_similarity_text(text1, text2, n: int = 3):
from collections import Counter
if not text1 or not text2:
return 0.0
text1 = preprocess_text(text1)
text2 = preprocess_text(text2)
ngrams1 = [text1[i : i + n] for i in range(len(text1) - n + 1)]
ngrams2 = [text2[i : i + n] for i in range(len(text2) - n + 1)]
counter1 = Counter(ngrams1)
counter2 = Counter(ngrams2)
all_ngrams = set(counter1.keys()) | set(counter2.keys())
vec1 = [counter1.get(ng, 0) for ng in all_ngrams]
vec2 = [counter2.get(ng, 0) for ng in all_ngrams]
dot_product = sum(a * b for a, b in zip(vec1, vec2))
norm1 = sum(a * a for a in vec1) ** 0.5
norm2 = sum(b * b for b in vec2) ** 0.5
if norm1 == 0 or norm2 == 0:
return 0.0
return dot_product / (norm1 * norm2)
def convert_audio_to_text(audio_data):
"""
Convert base64 encoded audio data to text using speech recognition.
"""
import whisper
audio_data = base64.b64decode(audio_data)
output_path = f"./test_{int(time.time())}"
@@ -147,8 +521,15 @@ def convert_audio_to_text(audio_data):
audio_file.write(audio_data)
print(f"audio data is saved: {output_path}")
model = whisper.load_model("base")
text = model.transcribe(output_path)["text"]
text = model.transcribe(
output_path,
temperature=0.0,
word_timestamps=True,
condition_on_previous_text=False,
initial_prompt="Please transcribe with proper word spacing.",
)["text"]
if text:
return text
else:
@@ -19,6 +19,8 @@ from tests.conftest import (
convert_audio_to_text,
cosine_similarity_text,
dummy_messages_from_mix_data,
generate_synthetic_audio,
generate_synthetic_image,
modify_stage_config,
)
@@ -68,6 +70,20 @@ def get_max_batch_size(size_type="few"):
return batch_sizes.get(size_type, 5)
def get_deploy_config(deploy_type="TP1"):
result = {
"TP1": {
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.parametrize("test_config", test_params)
def test_text_to_text_001(test_config: tuple[str, str]) -> None:
"""Test processing text, generating text output via OpenAI API."""
@@ -156,3 +172,198 @@ def test_text_to_text_audio_001(test_config: tuple[str, str]) -> None:
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."""
model, stage_config_path = test_config
deploy_config = get_deploy_config()
deploy_config[0]["default_sampling_params.ignore_eos"] = True
stage_config_path = modify_stage_config(stage_config_path, deploy_config)
with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "90"]) as server:
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
api_client = client(server)
start_time = time.perf_counter()
chat_completion = api_client.chat.completions.create(
model=server.model, messages=messages, max_tokens=200, modalities=["text"]
)
# 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 == 200, (
"The output length differs from the requested max_tokens."
)
# Verify E2E
print(f"the request e2e is: {time.perf_counter() - start_time}")
# TODO: Verify the E2E latency after confirmation baseline.
@pytest.mark.parametrize("test_config", test_params)
def test_audio_to_text_audio_001(test_config: tuple[str, str]) -> None:
"""Test processing text, generating 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:
audio_data_url = []
for _ in range(5):
audio_data_url.append(f"data:audio/wav;base64,{generate_synthetic_audio(1, 5)['base64']}")
messages = dummy_messages_from_mix_data(audio_data_url=audio_data_url)
# 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"
# 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("test_config", test_params)
def test_image_to_text_001(test_config: tuple[str, str]) -> None:
"""Test processing text, generating text output via OpenAI API."""
model, stage_config_path = test_config
deploy_config = get_deploy_config()
stage_config_path = modify_stage_config(stage_config_path, deploy_config)
with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "90"]) as server:
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
api_client = client(server)
start_time = time.perf_counter()
chat_completion = api_client.chat.completions.create(
model=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."
# 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."
@pytest.mark.parametrize("test_config", test_params)
def test_image_to_text_audio_001(test_config: tuple[str, str]) -> None:
"""Test processing text, generating audio output via OpenAI API."""
model, stage_config_path = test_config
num_concurrent_requests = 5
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:
image_data_url = []
for _ in range(4):
image_data_url.append(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
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_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"
+4 -4
View File
@@ -15,7 +15,7 @@ stage_args:
worker_cls: vllm_omni.worker.gpu_ar_worker.GPUARWorker
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.6
enforce_eager: true
enforce_eager: false
trust_remote_code: true
engine_output_type: latent # Output hidden states for talker
distributed_executor_backend: "mp"
@@ -46,7 +46,7 @@ stage_args:
worker_cls: vllm_omni.worker.gpu_ar_worker.GPUARWorker
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.3
enforce_eager: true
enforce_eager: false
trust_remote_code: true
engine_output_type: latent # Output codec codes for code2wav
enable_prefix_caching: false
@@ -58,7 +58,7 @@ stage_args:
default_sampling_params:
temperature: 0.9
top_k: 50
max_tokens: 100
max_tokens: 1000
seed: 42
detokenize: False
repetition_penalty: 1.05
@@ -90,7 +90,7 @@ stage_args:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 200
max_tokens: 2000
seed: 42
detokenize: True
repetition_penalty: 1.1