diff --git a/tests/e2e/online_serving/test_qwen3_omni_expansion.py b/tests/e2e/online_serving/test_qwen3_omni_expansion.py index db3d7a8d1..25ba4f284 100644 --- a/tests/e2e/online_serving/test_qwen3_omni_expansion.py +++ b/tests/e2e/online_serving/test_qwen3_omni_expansion.py @@ -410,6 +410,44 @@ def test_audio_in_video_002(omni_server, openai_client) -> None: openai_client.send_omni_request(request_config, request_num=get_max_batch_size()) +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_audio_in_video_default_loader_sampling_regression(omni_server, openai_client) -> None: + """ + Regression: ``use_audio_in_video`` with default video loader sampling (no + ``media_io_kwargs.video.fps``) after a prior request that pinned fps. + + Before the fix the second request could partial-hit mm-cache (audio cached, + video re-processed without audio) and fail with ``StopIteration`` in + ``replace_multimodal_special_tokens``. + """ + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(224, 224, 128, embed_audio=True)['base64']}" + messages = dummy_messages_from_mix_data( + system_prompt=get_system_prompt(), + video_data_url=video_data_url, + content_text="Reply with one word: OK.", + ) + + base_config = { + "model": omni_server.model, + "messages": messages, + "stream": False, + "use_audio_in_video": True, + "modalities": ["text"], + } + + # Prime mm-cache with explicit loader fps (historical failure mode). + openai_client.send_omni_request( + { + **base_config, + "extra_body": {"media_io_kwargs": {"video": {"fps": 1, "num_frames": 128}}}, + } + ) + + # Default loader sampling: must succeed without explicit media_io fps. + openai_client.send_omni_request(dict(base_config)) + + @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @pytest.mark.parametrize("omni_server", test_params, indirect=True) def test_one_word_prompt_001(omni_server, openai_client) -> None: diff --git a/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni.py b/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni.py index 4bd48c043..7b2c2558c 100644 --- a/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni.py +++ b/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni.py @@ -17,7 +17,6 @@ from vllm.logger import init_logger from vllm.model_executor.models.interfaces import SupportsMRoPE, SupportsMultiModal, SupportsPP from vllm.model_executor.models.qwen2_5_omni_thinker import ( Qwen2_5OmniConditionalGenerationMixin, - Qwen2_5OmniThinkerProcessingInfo, ) from vllm.model_executor.models.utils import init_vllm_registered_model, maybe_prefix from vllm.model_executor.models.vision import ( @@ -39,6 +38,7 @@ from vllm_omni.model_executor.models.output_templates import OmniOutput from vllm_omni.model_executor.models.qwen2_5_omni.qwen2_5_omni_thinker import ( Qwen2_5OmniThinkerDummyInputsBuilder, Qwen2_5OmniThinkerMultiModalProcessor, + Qwen2_5OmniThinkerProcessingInfo, ) from vllm_omni.model_executor.models.utils import add_prefix_to_loaded_weights, split_list_into_ranges from vllm_omni.platforms import current_omni_platform diff --git a/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py b/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py index 5d58a63d4..d0df1c2ea 100644 --- a/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py +++ b/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py @@ -4,6 +4,7 @@ from collections.abc import Iterable, Iterator, Mapping from typing import Any import numpy as np +import PIL.Image as PILImage import torch from torch import nn from transformers.models.qwen2_5_omni.configuration_qwen2_5_omni import ( @@ -26,16 +27,21 @@ from vllm.model_executor.models.module_mapping import MultiModelKeys from vllm.model_executor.models.qwen2_5_omni_thinker import ( Qwen2_5OmniAudioFeatureInputs, Qwen2_5OmniThinkerDummyInputsBuilder, - Qwen2_5OmniThinkerProcessingInfo, check_interleaved_audio_video, merge_interleaved_embeddings, ) from vllm.model_executor.models.qwen2_5_omni_thinker import ( Qwen2_5OmniConditionalGenerationMixin as Qwen2_5OmniConditionalGenerationMixinBase, ) +from vllm.model_executor.models.qwen2_5_omni_thinker import ( + Qwen2_5OmniThinkerMultiModalDataParser as _Qwen2_5OmniThinkerMultiModalDataParserBase, +) from vllm.model_executor.models.qwen2_5_omni_thinker import ( Qwen2_5OmniThinkerMultiModalProcessor as _Qwen2_5OmniThinkerMultiModalProcessorBase, ) +from vllm.model_executor.models.qwen2_5_omni_thinker import ( + Qwen2_5OmniThinkerProcessingInfo as _Qwen2_5OmniThinkerProcessingInfoBase, +) from vllm.model_executor.models.qwen2_5_vl import ( Qwen2_5_VisionTransformer, Qwen2_5_VLImageEmbeddingInputs, @@ -56,12 +62,13 @@ from vllm.multimodal.inputs import ( MultiModalFeatureSpec, MultiModalKwargsItems, ) -from vllm.multimodal.parse import MultiModalDataItems +from vllm.multimodal.parse import MultiModalDataItems, VideoProcessorItems from vllm.multimodal.processing.processor import ( MultiModalPromptUpdates, PlaceholderFeaturesInfo, ) from vllm.sequence import IntermediateTensors +from vllm.utils.collection_utils import is_list_of from vllm_omni.quantization.component_config import ( resolve_encoder_quant_config, @@ -74,11 +81,143 @@ except (ImportError, ModuleNotFoundError): logger = init_logger(__name__) +def _presampled_videos_hf_kwargs( + mm_data: Mapping[str, object], + mm_kwargs: Mapping[str, object], +) -> Mapping[str, object]: + """Adjust HF video kwargs for videos pre-sampled by vLLM's video loader. + + When ``video_metadata`` is present (emitted by + ``Qwen2_5OmniVideoProcessorItems``), the frames were already sampled + according to ``media_io_kwargs``, so the HF processor must not re-sample + them, and it needs the sampled fps (instead of its default) to compute + ``video_second_per_grid`` correctly. Otherwise the audio/video temporal + alignment breaks under ``use_audio_in_video=True``. + """ + video_metadata = mm_data.get("video_metadata") + if not video_metadata: + return mm_kwargs + + mm_kwargs = dict(mm_kwargs) + videos_kwargs = dict(mm_kwargs.get("videos_kwargs") or {}) + videos_kwargs["do_sample_frames"] = False + + def _compute_sampled_video_fps(metadata) -> float | None: + duration = getattr(metadata, "duration", None) + indices = getattr(metadata, "frames_indices", None) + if not duration or not indices or duration <= 0: + return None + return len(indices) / float(duration) + + # An explicit user-provided fps takes precedence. + if "fps" not in videos_kwargs: + fps_values = [_compute_sampled_video_fps(m) for m in video_metadata] + # HF accepts a single fps per call and uses it for every video's + # video_second_per_grid. + known_fps = [fps for fps in fps_values if fps is not None] + unique_fps = set(known_fps) + if len(unique_fps) == 1: + videos_kwargs["fps"] = known_fps[0] + elif len(unique_fps) > 1: + logger.warning( + f"Mixed sampled FPS {sorted(unique_fps)} in one request; HF accepts a single fps, using {known_fps[0]}." + ) + videos_kwargs["fps"] = known_fps[0] + + mm_kwargs["videos_kwargs"] = videos_kwargs + return mm_kwargs + + +class Qwen2_5OmniVideoProcessorItems(VideoProcessorItems): + """Video items that carry the loader's ``(frames, metadata)`` tuples. + + The tuples are kept in ``data`` so that mm hashing and cache-miss + re-parsing retain the metadata; ``get_processor_data`` unpacks them into + the ``videos`` + ``video_metadata`` arguments of the HF processor. + """ + + def get_processor_data(self) -> Mapping[str, object]: + from transformers.video_utils import VideoMetadata + + videos = [frames for frames, _ in self.data] + video_metadata = [ + VideoMetadata(**{k: v for k, v in metadata.items() if k != "do_sample_frames"}) for _, metadata in self.data + ] + return {"videos": videos, "video_metadata": video_metadata} + + +class Qwen2_5OmniThinkerMultiModalDataParser(_Qwen2_5OmniThinkerMultiModalDataParserBase): + def _parse_video_data(self, data): + if data is None or isinstance(data, dict) or self.is_embeddings(data): + return super()._parse_video_data(data) + + # Normalize to a list of per-video items (same as the base parser). + if (is_list_of(data, PILImage.Image) and len(data) > 0) or ( + isinstance(data, (np.ndarray, torch.Tensor)) and data.ndim == 4 + ): + data_items = [data] + elif isinstance(data, (np.ndarray, torch.Tensor)): + data_items = [elem for elem in data] + elif isinstance(data, tuple) and len(data) == 2: + data_items = [data] + else: + data_items = data + + videos_with_metadata = [self._get_video_with_metadata(item) for item in data_items] + + # Metadata is optional: videos fetched by vLLM's media connector + # arrive as (frames, metadata) tuples, while plain arrays (e.g. + # offline `multi_modal_data` or dummy profiling inputs) carry no + # metadata and parse as before. + if any(metadata is None for _, metadata in videos_with_metadata): + return super()._parse_video_data(data) + + return Qwen2_5OmniVideoProcessorItems( + videos_with_metadata, + metadata=[metadata for _, metadata in videos_with_metadata], + ) + + +class Qwen2_5OmniThinkerProcessingInfo(_Qwen2_5OmniThinkerProcessingInfoBase): + def get_data_parser(self): + feature_extractor = self.get_feature_extractor() + + return Qwen2_5OmniThinkerMultiModalDataParser( + spatial_merge_size=self.get_hf_config().vision_config.spatial_merge_size, + target_sr=feature_extractor.sampling_rate, + target_channels=self.get_target_channels(), + expected_hidden_size=self._get_expected_hidden_size(), + ) + + class Qwen2_5OmniThinkerMultiModalProcessor( _Qwen2_5OmniThinkerMultiModalProcessorBase, ): """Override to fix use_audio_in_video detection when mm cache returns None.""" + def _cached_apply_hf_processor(self, inputs, timing_ctx): + # If use_audio_in_video, process video and audio together; otherwise, + # skip cache to avoid errors. + # Prevents partial cache hits from causing processing failures. + if inputs.hf_processor_mm_kwargs.get("use_audio_in_video"): + return self._apply_hf_processor(inputs, timing_ctx) + return super()._cached_apply_hf_processor(inputs, timing_ctx) + + def _call_hf_processor( + self, + prompt: str, + mm_data: Mapping[str, object], + mm_kwargs: Mapping[str, object], + tok_kwargs: Mapping[str, object], + ): + mm_kwargs = _presampled_videos_hf_kwargs(mm_data, mm_kwargs) + return super()._call_hf_processor( + prompt=prompt, + mm_data=mm_data, + mm_kwargs=mm_kwargs, + tok_kwargs=tok_kwargs, + ) + def _maybe_apply_prompt_updates( self, mm_items: MultiModalDataItems, diff --git a/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_thinker.py b/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_thinker.py index 0d7f69062..011496f86 100644 --- a/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_thinker.py +++ b/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_thinker.py @@ -74,7 +74,6 @@ from vllm.model_executor.models.module_mapping import MultiModelKeys from vllm.model_executor.models.qwen2_5_omni_thinker import ( Qwen2_5OmniAudioFeatureInputs, Qwen2_5OmniThinkerDummyInputsBuilder, - Qwen2_5OmniThinkerMultiModalProcessor, check_interleaved_audio_video, merge_interleaved_embeddings, ) @@ -120,6 +119,9 @@ from vllm.transformers_utils.processor import cached_processor_from_config from vllm_omni.model_executor.models.qwen2_5_omni.qwen2_5_omni_thinker import ( Qwen2_5OmniConditionalGenerationMixin, + Qwen2_5OmniThinkerMultiModalDataParser, + Qwen2_5OmniThinkerMultiModalProcessor, + _presampled_videos_hf_kwargs, ) from vllm_omni.quantization.component_config import ( PRE_QUANTIZED_METHODS, @@ -621,6 +623,19 @@ class Qwen3OmniMoeThinkerProcessingInfo(Qwen2AudioProcessingInfo, Qwen2_5_VLProc assert isinstance(feature_extractor, WhisperFeatureExtractor) return feature_extractor + def get_data_parser(self): + feature_extractor = self.get_feature_extractor() + + # Install the Omni parser, which keeps video metadata from vLLM's + # video loader (see Qwen2_5OmniVideoProcessorItems); the parser + # inherited from Qwen2AudioProcessingInfo would drop it. + return Qwen2_5OmniThinkerMultiModalDataParser( + spatial_merge_size=self.get_hf_config().vision_config.spatial_merge_size, + target_sr=feature_extractor.sampling_rate, + target_channels=1, + expected_hidden_size=self._get_expected_hidden_size(), + ) + def get_supported_mm_limits(self) -> Mapping[str, int | None]: return {"audio": None, "image": None, "video": None} @@ -671,8 +686,11 @@ class Qwen3OmniMoeThinkerMultiModalProcessor( tok_kwargs: Mapping[str, object], ) -> BatchFeature: mm_data = dict(mm_data) + mm_kwargs = dict(mm_kwargs) audios = mm_data.pop("audios", []) + mm_kwargs = dict(_presampled_videos_hf_kwargs(mm_data, mm_kwargs)) + def pad_to_hop_length(x: np.ndarray, hop_length: int) -> np.ndarray: length = x.shape[-1] if length % hop_length != 0: @@ -927,9 +945,13 @@ class Qwen3OmniMoeThinkerMultiModalProcessor( audio_in_video_item_idx += 1 - second_per_grid_ts = hf_processor_mm_kwargs.get("second_per_grid_ts", None) - if second_per_grid_ts: - video_second_per_grid_t = second_per_grid_ts[item_idx] + # Prefer the HF-computed value (temporal_patch_size / sampled fps) + # over request kwargs; + second_per_grid_ts = out_mm_data.get("second_per_grid_ts") + if second_per_grid_ts is None: + second_per_grid_ts = hf_processor_mm_kwargs.get("second_per_grid_ts", None) + if second_per_grid_ts is not None: + video_second_per_grid_t = float(second_per_grid_ts[item_idx]) else: video_second_per_grid_t = 2.0