[OpenVINO] Support Youtu-VL-4B-Instruct with task image-text-to-text

docs/source/openvino/models.mdx

@@ -174,6 +174,7 @@ Here is the list of the supported architectures :
 - XLM
 - XLM-RoBERTa
 - XVERSE
+- Youtu_vl
 - Zamba2
 
 ## [Diffusers](https://huggingface.co/docs/diffusers/index)

optimum/exporters/openvino/model_configs.py

@@ -173,6 +173,8 @@
     SpeechT5ModelPatcher,
     VideoChatFlashQwenVisionEmbeddingModelPatcher,
     XverseModelPatcher,
+    YoutuVLLanguageModelPatcher,
+    YoutuVLVisionEmbMergerPatcher,
     Zamba2ModelPatcher,
     _get_model_attribute,
 )
@@ -326,6 +328,14 @@ def init_model_configs():
             "transformers",
             "AutoModel",
         )
+    TasksManager._CUSTOM_CLASSES[("pt", "llama4", "image-text-to-text")] = (
+        "transformers",
+        "AutoModelForImageTextToText",
+    )
+    TasksManager._CUSTOM_CLASSES[("pt", "youtu_vl", "image-text-to-text")] = (
+        "transformers",
+        "AutoModelForCausalLM",
+    )
 
     if is_diffusers_available() and "fill" not in TasksManager._DIFFUSERS_TASKS_TO_MODEL_LOADERS:
         TasksManager._DIFFUSERS_TASKS_TO_MODEL_LOADERS["fill"] = "FluxFillPipeline"
@@ -3861,6 +3871,201 @@ def __init__(
             self._normalized_config = self.NORMALIZED_CONFIG_CLASS(self._config)
 
 
+@register_in_tasks_manager(
+    "youtu_vl", *["text-generation", "text-generation-with-past"], library_name="transformers"
+)
+class YoutuVLTextOpenVINOConfig(MiniCPM3OpenVINOConfig):
+    # The Youtu language model is a DeepSeek-V2/V3 style decoder that uses multi-head
+    # latent attention (MLA). The key/value cache layout therefore matches MiniCPM3 /
+    # DeepSeek (k_head_dim = qk_nope_head_dim + qk_rope_head_dim, v_head_dim). The model
+    # is shipped as remote code, so unlike the in-library deepseek/minicpm3 configs we do
+    # not constrain the maximum transformers version.
+    MIN_TRANSFORMERS_VERSION = "4.49.0"
+    MAX_TRANSFORMERS_VERSION = "999.9.9"
+    _MODEL_PATCHER = YoutuVLLanguageModelPatcher
+
+
+class YoutuVLVisionEmbeddingsDummyInputGenerator(DummyVisionInputGenerator):
+    # The Youtu-VL vision tower follows the Qwen2.5-VL windowed-attention design, so the
+    # vision part is split into a patch-embedding model and a transformer/merger model.
+    # Data-dependent operations (window indices, cu_seqlens based masks) are computed at
+    # runtime and fed into the merger as explicit inputs.
+    SUPPORTED_INPUT_NAMES = (
+        "pixel_values",
+        "hidden_states",
+        "attention_mask",
+        "window_attention_mask",
+        "window_index",
+        "rotary_pos_emb",
+    )
+
+    def __init__(
+        self,
+        task: str,
+        normalized_config: NormalizedVisionConfig,
+        batch_size: int = 1,
+        **kwargs,
+    ):
+        self.task = task
+        self.normalized_config = normalized_config
+        vision_config = normalized_config.config
+        self.batch_size = 1
+        self.patch_size = vision_config.patch_size
+        self.num_channels = vision_config.num_channels
+        self.hidden_size = vision_config.hidden_size
+        self.num_heads = vision_config.num_attention_heads
+        self.spatial_merge_size = 2
+        self.spatial_merge_unit = self.spatial_merge_size * self.spatial_merge_size
+        # square grid of patches used for tracing; the corresponding axes are dynamic
+        self.spatial_dim = 8
+        self.num_patches = self.spatial_dim * self.spatial_dim
+        # flattened patch dimension (num_channels * patch_size * patch_size)
+        self.embed_dim = self.num_channels * self.patch_size * self.patch_size
+        # rotary embedding dimension used by the vision transformer
+        self.rotary_dim = self.hidden_size // self.num_heads // 2
+
+    def generate(self, input_name: str, framework: str = "pt", int_dtype: str = "int64", float_dtype: str = "fp32"):
+        if input_name == "pixel_values":
+            shape = [self.batch_size, self.num_patches, self.embed_dim]
+            return self.random_float_tensor(shape, framework=framework, dtype=float_dtype)
+        if input_name == "hidden_states":
+            shape = [self.num_patches, self.hidden_size]
+            return self.random_float_tensor(shape, framework=framework, dtype=float_dtype)
+        if input_name in ["attention_mask", "window_attention_mask"]:
+            return self.random_mask_tensor(
+                [1, self.num_patches, self.num_patches], framework=framework, dtype=float_dtype
+            )
+        if input_name == "rotary_pos_emb":
+            shape = [self.num_patches, self.rotary_dim]
+            return self.random_float_tensor(shape, framework=framework, dtype=float_dtype)
+        if input_name == "window_index":
+            length = self.num_patches // self.spatial_merge_unit
+            return self.random_int_tensor([length], max_value=length, framework=framework, dtype=int_dtype)
+        raise ValueError(f"Unsupported input name {input_name}")
+
+
+@register_in_tasks_manager("youtu_vl", *["image-text-to-text"], library_name="transformers")
+class YoutuVLOpenVINOConfig(BaseVLMOpenVINOConfig):
+    SUPPORTED_BEHAVIORS = [
+        model_type.value for model_type in QwenVLConfigBehavior if model_type != QwenVLConfigBehavior.VISION_EMBEDDINGS_POS
+    ]
+    NORMALIZED_CONFIG_CLASS = NormalizedVisionConfig
+    DUMMY_INPUT_GENERATOR_CLASSES = (YoutuVLVisionEmbeddingsDummyInputGenerator,)
+    MIN_TRANSFORMERS_VERSION = "4.49.0"
+
+    def __init__(
+        self,
+        config: "PretrainedConfig",
+        task: str = "feature-extraction",
+        int_dtype: str = "int64",
+        float_dtype: str = "fp32",
+        behavior: QwenVLConfigBehavior = QwenVLConfigBehavior.VISION_EMBEDDINGS,
+        preprocessors: Optional[List[Any]] = None,
+        **kwargs,
+    ):
+        super().__init__(
+            config=config,
+            task=task,
+            int_dtype=int_dtype,
+            float_dtype=float_dtype,
+            preprocessors=preprocessors,
+        )
+        self._behavior = behavior
+        self._orig_config = config
+        if (
+            self._behavior in [QwenVLConfigBehavior.VISION_EMBEDDINGS, QwenVLConfigBehavior.VISION_EMBEDDINGS_MERGER]
+            and hasattr(config, "vision_config")
+        ):
+            self._config = config.vision_config
+            self._normalized_config = self.NORMALIZED_CONFIG_CLASS(self._config)
+
+    @staticmethod
+    def get_model_for_behavior(model, behavior: Union[str, QwenVLConfigBehavior]):
+        if isinstance(behavior, str) and not isinstance(behavior, QwenVLConfigBehavior):
+            behavior = QwenVLConfigBehavior(behavior)
+
+        if behavior == QwenVLConfigBehavior.LANGUAGE:
+            return model
+
+        if behavior == QwenVLConfigBehavior.VISION_EMBEDDINGS:
+            # patch embedding (Siglip2VisionEmbeddingsWoPos)
+            vision_embeddings = model.siglip2.vision_model.embeddings
+            vision_embeddings.config = model.config.vision_config
+            return vision_embeddings
+
+        if behavior == QwenVLConfigBehavior.VISION_EMBEDDINGS_MERGER:
+            # the encoder + post layernorm + merger are exported together; they are
+            # spread across several submodules of the top-level model, so we export the
+            # whole model and rely on the patcher to wire the right forward.
+            model.config.vision_config.fullatt_block_indexes = model.config.vision_config.fullatt_block_indexes
+            return model
+
+        if behavior == QwenVLConfigBehavior.TEXT_EMBEDDINGS:
+            text_embedding = model.get_input_embeddings()
+            text_embedding.config = model.config
+            return text_embedding
+
+    def with_behavior(
+        self,
+        behavior: Union[str, QwenVLConfigBehavior],
+    ):
+        if isinstance(behavior, str) and not isinstance(behavior, QwenVLConfigBehavior):
+            behavior = QwenVLConfigBehavior(behavior)
+
+        if behavior == QwenVLConfigBehavior.TEXT_EMBEDDINGS:
+            return get_vlm_text_embeddings_config("youtu_vl", self._orig_config, self.int_dtype, self.float_dtype)
+
+        if behavior == QwenVLConfigBehavior.LANGUAGE:
+            return get_vlm_text_generation_config(
+                "youtu_vl",
+                self._orig_config,
+                self.int_dtype,
+                self.float_dtype,
+                model_patcher=YoutuVLLanguageModelPatcher,
+            )
+
+        if behavior in [QwenVLConfigBehavior.VISION_EMBEDDINGS, QwenVLConfigBehavior.VISION_EMBEDDINGS_MERGER]:
+            return self.__class__(
+                self._orig_config,
+                task=self.task,
+                int_dtype=self.int_dtype,
+                float_dtype=self.float_dtype,
+                behavior=behavior,
+                preprocessors=self._preprocessors,
+            )
+
+    def patch_model_for_export(self, model: PreTrainedModel, model_kwargs: Optional[Dict[str, Any]] = None):
+        model_kwargs = model_kwargs or {}
+        if self._behavior == QwenVLConfigBehavior.VISION_EMBEDDINGS_MERGER:
+            return YoutuVLVisionEmbMergerPatcher(self, model, model_kwargs)
+        if self._behavior == QwenVLConfigBehavior.VISION_EMBEDDINGS:
+            return ModelPatcher(self, model, model_kwargs=model_kwargs)
+        return super().patch_model_for_export(model, model_kwargs)
+
+    @property
+    def inputs(self) -> Dict[str, Dict[int, str]]:
+        if self._behavior == QwenVLConfigBehavior.VISION_EMBEDDINGS:
+            return {"pixel_values": {0: "batch_size", 1: "num_patches"}}
+        if self._behavior == QwenVLConfigBehavior.VISION_EMBEDDINGS_MERGER:
+            return {
+                "hidden_states": {0: "sequence_length"},
+                "attention_mask": {1: "sequence_length", 2: "sequence_length"},
+                "window_attention_mask": {1: "sequence_length", 2: "sequence_length"},
+                "window_index": {0: "unit_sequence_length"},
+                "rotary_pos_emb": {0: "sequence_length"},
+            }
+        return {}
+
+    @property
+    def outputs(self) -> Dict[str, Dict[int, str]]:
+        if self._behavior in [
+            QwenVLConfigBehavior.VISION_EMBEDDINGS,
+            QwenVLConfigBehavior.VISION_EMBEDDINGS_MERGER,
+        ]:
+            return {"last_hidden_state": {0: "seq_len"}}
+        return {}
+
+
 @register_in_tasks_manager("gemma3", *["image-text-to-text"], library_name="transformers")
 class Gemma3OpenVINOConfig(BaseVLMOpenVINOConfig):
     MIN_TRANSFORMERS_VERSION = "4.50.0"

optimum/exporters/openvino/model_patcher.py

@@ -10512,3 +10512,126 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_value, traceback):
         super().__exit__(exc_type, exc_value, traceback)
         self._model.forward = self._model._orig_forward
+
+def _youtu_vl_vision_rotate_half(x):
+    x1 = x[..., : x.shape[-1] // 2]
+    x2 = x[..., x.shape[-1] // 2 :]
+    return torch.cat((-x2, x1), dim=-1)
+
+def _youtu_vl_apply_rotary_pos_emb_vision(q, k, cos, sin):
+    orig_q_dtype = q.dtype
+    orig_k_dtype = k.dtype
+    q, k = q.float(), k.float()
+    cos, sin = cos.unsqueeze(-2).float(), sin.unsqueeze(-2).float()
+    q_embed = (q * cos) + (_youtu_vl_vision_rotate_half(q) * sin)
+    k_embed = (k * cos) + (_youtu_vl_vision_rotate_half(k) * sin)
+    return q_embed.to(orig_q_dtype), k_embed.to(orig_k_dtype)
+
+def _youtu_vl_vision_sdpa_attn_forward(self, hidden_states, attention_mask, position_embeddings):
+    seq_length = hidden_states.shape[0]
+    q = self.q_proj(hidden_states).view(seq_length, self.num_heads, self.head_dim)
+    k = self.k_proj(hidden_states).view(seq_length, self.num_heads, self.head_dim)
+    v = self.v_proj(hidden_states).view(seq_length, self.num_heads, self.head_dim)
+    cos, sin = position_embeddings
+    q, k = _youtu_vl_apply_rotary_pos_emb_vision(q, k, cos, sin)
+    q = q.transpose(0, 1).unsqueeze(0)
+    k = k.transpose(0, 1).unsqueeze(0)
+    v = v.transpose(0, 1).unsqueeze(0)
+    attn_output = torch.nn.functional.scaled_dot_product_attention(q, k, v, attn_mask=attention_mask)
+    attn_output = attn_output.squeeze(0).transpose(0, 1).reshape(seq_length, -1).to(hidden_states.dtype)
+    return self.out_proj(attn_output), None
+
+def _youtu_vl_vision_block_forward(self, hidden_states, attention_mask, position_embeddings):
+    residual = hidden_states
+    hidden_states = self.layer_norm1(hidden_states)
+    hidden_states, _ = self.self_attn(
+        hidden_states=hidden_states, attention_mask=attention_mask, position_embeddings=position_embeddings
+    )
+    hidden_states = residual + hidden_states
+
+    residual = hidden_states
+    hidden_states = self.layer_norm2(hidden_states)
+    hidden_states = self.mlp(hidden_states)
+    hidden_states = residual + hidden_states
+    return (hidden_states,)
+
+class YoutuVLVisionEmbMergerPatcher(ModelPatcher):
+    def __init__(
+        self,
+        config: "OnnxConfig",
+        model: "PreTrainedModel",
+        model_kwargs: Dict[str, Any] = None,
+    ):
+        model.__orig_forward = model.forward
+
+        # Modified from the Youtu-VL Siglip2 windowed vision encoder. The data-dependent
+        # window indices / cu_seqlens-based attention masks are computed at runtime and fed
+        # as explicit inputs (attention_mask, window_attention_mask, window_index,
+        # rotary_pos_emb), mirroring the Qwen2.5-VL OpenVINO export.
+        def image_embed_forward(
+            self,
+            hidden_states: torch.Tensor,
+            attention_mask: torch.Tensor,
+            window_attention_mask: torch.Tensor,
+            window_index: torch.Tensor,
+            rotary_pos_emb: torch.Tensor,
+        ) -> torch.Tensor:
+            encoder = self.siglip2.vision_model.encoder
+            post_layernorm = self.siglip2.vision_model.post_layernorm
+            spatial_merge_unit = encoder.spatial_merge_unit
+            fullatt_block_indexes = self.config.vision_config.fullatt_block_indexes
+            num_layers = len(encoder.layers)
+
+            seq_len = hidden_states.shape[0]
+            hidden_states = hidden_states.reshape(seq_len // spatial_merge_unit, spatial_merge_unit, -1)
+            hidden_states = hidden_states[window_index, :, :]
+            hidden_states = hidden_states.reshape(seq_len, -1)
+            rotary_pos_emb = rotary_pos_emb.reshape(seq_len // spatial_merge_unit, spatial_merge_unit, -1)
+            rotary_pos_emb = rotary_pos_emb[window_index, :, :]
+            rotary_pos_emb = rotary_pos_emb.reshape(seq_len, -1)
+            emb = torch.cat((rotary_pos_emb, rotary_pos_emb), dim=-1)
+            position_embeddings = (emb.cos(), emb.sin())
+
+            for layer_num, blk in enumerate(encoder.layers):
+                if (1 + layer_num) % 8 == 0 or layer_num == num_layers - 1 or layer_num in fullatt_block_indexes:
+                    attention_mask_now = attention_mask
+                else:
+                    attention_mask_now = window_attention_mask
+                hidden_states = blk(
+                    hidden_states, attention_mask=attention_mask_now, position_embeddings=position_embeddings
+                )[0]
+
+            # restore the original (non-windowed) ordering before the patch merger
+            hidden_states = hidden_states.reshape(seq_len // spatial_merge_unit, spatial_merge_unit, -1)
+            reverse_indices = torch.argsort(window_index)
+            hidden_states = hidden_states[reverse_indices, :, :]
+            hidden_states = hidden_states.reshape(seq_len, -1)
+
+            hidden_states = post_layernorm(hidden_states)
+            hidden_states = self.merger(hidden_states, None)
+            return hidden_states
+
+        model.forward = types.MethodType(image_embed_forward, model)
+        super().__init__(config, model, model_kwargs)
+
+    def __enter__(self):
+        super().__enter__()
+        for block in self._model.siglip2.vision_model.encoder.layers:
+            block._orig_forward = block.forward
+            block.forward = types.MethodType(_youtu_vl_vision_block_forward, block)
+            block.self_attn._orig_forward = block.self_attn.forward
+            block.self_attn.forward = types.MethodType(_youtu_vl_vision_sdpa_attn_forward, block.self_attn)
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        super().__exit__(exc_type, exc_value, traceback)
+        self._model.forward = self._model.__orig_forward
+        for block in self._model.siglip2.vision_model.encoder.layers:
+            block.forward = block._orig_forward
+            block.self_attn.forward = block.self_attn._orig_forward
+
+class YoutuVLLanguageModelPatcher(OVDecoderModelPatcher):
+    # The Youtu language model uses DeepSeek-style multi-head latent attention (MLA)
+    # with rotary embeddings computed inside the base model and SDPA in the attention
+    # block. These patterns are already torchscript/OpenVINO friendly, so no extra
+    # attention patching is required on top of the generic decoder patcher.
+    pass

optimum/exporters/openvino/utils.py

@@ -320,6 +320,7 @@ def _get_kokoro_submodels(model):
     "internvl_chat",
     "maira2",
     "minicpmv",
+    "youtu_vl",
     "phi3_v",
     "qwen2_vl",
     "qwen2_5_vl",