# Copyright 2025 The Qwen Team and The HuggingFace Inc. team. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """PyTorch Qwen3.5 model.""" from typing import Optional import torch import torch.nn.functional as F from torch import nn from ... import initialization as init from ...cache_utils import Cache from ...masking_utils import create_causal_mask from ...modeling_layers import GradientCheckpointingLayer from ...modeling_outputs import BaseModelOutputWithPast, BaseModelOutputWithPooling from ...modeling_rope_utils import RopeParameters from ...modeling_utils import PreTrainedModel from ...processing_utils import Unpack from ...utils import TransformersKwargs, auto_docstring, can_return_tuple, logging from ...utils.generic import merge_with_config_defaults from ...utils.output_capturing import capture_outputs from ..qwen3.modeling_qwen3 import Qwen3ForCausalLM from ..qwen3_next.configuration_qwen3_next import Qwen3NextConfig from ..qwen3_next.modeling_qwen3_next import ( Qwen3NextAttention, Qwen3NextDynamicCache, Qwen3NextGatedDeltaNet, Qwen3NextMLP, Qwen3NextModel, Qwen3NextPreTrainedModel, Qwen3NextRMSNorm, apply_mask_to_padding_states, ) from ..qwen3_vl.configuration_qwen3_vl import Qwen3VLConfig, Qwen3VLVisionConfig from ..qwen3_vl.modeling_qwen3_vl import ( Qwen3VLForConditionalGeneration, Qwen3VLModel, Qwen3VLModelOutputWithPast, Qwen3VLTextRotaryEmbedding, Qwen3VLVisionModel, Qwen3VLVisionRotaryEmbedding, ) logger = logging.get_logger(__name__) class Qwen3_5TextConfig(Qwen3NextConfig): r""" This is the configuration class to store the configuration of a [`Qwen3_5TextModel`]. It is used to instantiate a Qwen3_5 model according to the specified arguments, defining the model architecture. Instantiating a configuration with the defaults will yield a similar configuration to that of Qwen3.5-9B-Instruct [Qwen/Qwen3.5-9B-Instruct](https://huggingface.co/Qwen/Qwen3.5-9B-Instruct). Configuration objects inherit from [`PreTrainedConfig`] and can be used to control the model outputs. Read the documentation from [`PreTrainedConfig`] for more information. Args: vocab_size (`int`, *optional*, defaults to 248320): Vocabulary size of the model. Defines the number of different tokens that can be represented by the `inputs_ids`. hidden_size (`int`, *optional*, defaults to 4096): Dimension of the hidden representations. intermediate_size (`int`, *optional*, defaults to 12288): Dimension of the MLP representations. num_hidden_layers (`int`, *optional*, defaults to 32): Number of hidden layers in the Transformer encoder. num_attention_heads (`int`, *optional*, defaults to 16): Number of attention heads for each attention layer in the Transformer encoder. num_key_value_heads (`int`, *optional*, defaults to 4): This is the number of key_value heads that should be used to implement Grouped Query Attention. If `num_key_value_heads=num_attention_heads`, the model will use Multi Head Attention (MHA), if `num_key_value_heads=1` the model will use Multi Query Attention (MQA) otherwise GQA is used. When converting a multi-head checkpoint to a GQA checkpoint, each group key and value head should be constructed by meanpooling all the original heads within that group. For more details checkout [this paper](https://arxiv.org/pdf/2305.13245.pdf). If it is not specified, will default to `32`. hidden_act (`str`, *optional*, defaults to `"silu"`): The non-linear activation function in the decoder. max_position_embeddings (`int`, *optional*, defaults to 32768): The maximum sequence length that this model might ever be used with. initializer_range (`float`, *optional*, defaults to 0.02): The standard deviation of the truncated_normal_initializer for initializing all weight matrices. rms_norm_eps (`float`, *optional*, defaults to 1e-06): The epsilon used by the rms normalization layers. use_cache (`bool`, *optional*, defaults to `True`): Whether or not the model should return the last key/values attentions (not used by all models). Only relevant if `config.is_decoder=True`. tie_word_embeddings (`bool`, *optional*, defaults to `False`): Whether the model's input and output word embeddings should be tied. rope_parameters (`RopeParameters`, *optional*): Dictionary containing the configuration parameters for the RoPE embeddings. The dictionary should contain a value for `rope_theta` and optionally parameters used for scaling in case you want to use RoPE with longer `max_position_embeddings`. attention_bias (`bool`, *optional*, defaults to `False`): Whether to use a bias in the query, key, value and output projection layers during self-attention. attention_dropout (`float`, *optional*, defaults to 0.0): The dropout ratio for the attention probabilities. head_dim (`int`, *optional*, defaults to 256): Projection weights dimension in multi-head attention. linear_conv_kernel_dim (`int`, *optional*, defaults to 4): Kernel size of the convolution used in linear attention layers. linear_key_head_dim (`int`, *optional*, defaults to 128): Dimension of each key head in linear attention. linear_value_head_dim (`int`, *optional*, defaults to 128): Dimension of each value head in linear attention. linear_num_key_heads (`int`, *optional*, defaults to 16): Number of key heads used in linear attention layers. linear_num_value_heads (`int`, *optional*, defaults to 32): Number of value heads used in linear attention layers. layer_types (`list[str]`, *optional*): Types of each layer (attention or linear). pad_token_id (`int`, *optional*): Padding token id. bos_token_id (`int`, *optional*): Beginning of stream token id. eos_token_id (`int`, *optional*): End of stream token id. ```python >>> from transformers import Qwen3_5TextModel, Qwen3_5TextConfig >>> # Initializing a Qwen3.5 style configuration >>> configuration = Qwen3_5TextConfig() >>> # Initializing a model from the Qwen3.5-9B style configuration >>> model = Qwen3_5TextModel(configuration) >>> # Accessing the model configuration >>> configuration = model.config ``` """ model_type = "qwen3_5_text" base_config_key = "text_config" base_model_tp_plan = { "layers.*.self_attn.q_proj": "colwise", "layers.*.self_attn.k_proj": "colwise", "layers.*.self_attn.v_proj": "colwise", "layers.*.self_attn.o_proj": "rowwise", "layers.*.mlp.gate_proj": "colwise", "layers.*.mlp.up_proj": "colwise", "layers.*.mlp.down_proj": "rowwise", } def __init__( self, vocab_size=248320, hidden_size=4096, intermediate_size=12288, num_hidden_layers=32, num_attention_heads=16, num_key_value_heads=4, hidden_act="silu", max_position_embeddings=32768, initializer_range=0.02, rms_norm_eps=1e-6, use_cache=True, tie_word_embeddings=False, rope_parameters: RopeParameters | dict[str, RopeParameters] | None = None, attention_bias=False, attention_dropout=0.0, head_dim=256, linear_conv_kernel_dim=4, linear_key_head_dim=128, linear_value_head_dim=128, linear_num_key_heads=16, linear_num_value_heads=32, layer_types=None, pad_token_id: int | None = None, bos_token_id: int | None = None, eos_token_id: int | None = None, **kwargs, ): kwargs["ignore_keys_at_rope_validation"] = {"mrope_section", "mrope_interleaved"} super().__init__( tie_word_embeddings=tie_word_embeddings, **kwargs, ) del self.decoder_sparse_step del self.norm_topk_prob del self.mlp_only_layers del self.moe_intermediate_size del self.shared_expert_intermediate_size del self.num_experts_per_tok del self.num_experts del self.output_router_logits del self.router_aux_loss_coef class Qwen3_5VisionConfig(Qwen3VLVisionConfig): model_type = "qwen3_5" def __init__( self, depth=27, hidden_size=1152, hidden_act="gelu_pytorch_tanh", intermediate_size=4304, num_heads=16, in_channels=3, patch_size=16, spatial_merge_size=2, temporal_patch_size=2, out_hidden_size=3584, num_position_embeddings=2304, initializer_range=0.02, **kwargs, ): super().__init__(**kwargs) del self.deepstack_visual_indexes class Qwen3_5Config(Qwen3VLConfig): r""" This is the configuration class to store the configuration of a [`Qwen3_5Model`]. It is used to instantiate a Qwen3.5 model according to the specified arguments, defining the model architecture. Instantiating a configuration with the defaults will yield a similar configuration to that of Qwen3.5-9B-Instruct [Qwen/Qwen3.5-9B-Instruct](https://huggingface.co/Qwen/Qwen3.5-9B-Instruct). Configuration objects inherit from [`PreTrainedConfig`] and can be used to control the model outputs. Read the documentation from [`PreTrainedConfig`] for more information. Args: text_config (`Union[PreTrainedConfig, dict]`, *optional*, defaults to `Qwen3_5TextConfig`): The config object or dictionary of the text backbone. vision_config (`Union[PreTrainedConfig, dict]`, *optional*, defaults to `Qwen3_5VisionConfig`): The config object or dictionary of the vision backbone. image_token_id (`int`, *optional*, defaults to 248056): The image token index to encode the image prompt. video_token_id (`int`, *optional*, defaults to 248057): The video token index to encode the image prompt. vision_start_token_id (`int`, *optional*, defaults to 248053): The start token index to encode the image prompt. vision_end_token_id (`int`, *optional*, defaults to 248054): The end token index to encode the image prompt. tie_word_embeddings (`bool`, *optional*, defaults to `False`): Whether to tie the word embeddings. ```python >>> from transformers import Qwen3_5ForConditionalGeneration, Qwen3_5Config >>> # Initializing a Qwen3.5 style configuration >>> configuration = Qwen3_5Config() >>> # Initializing a model from the Qwen3.5-9B style configuration >>> model = Qwen3_5ForConditionalGeneration(configuration) >>> # Accessing the model configuration >>> configuration = model.config ```""" model_type = "qwen3_5" sub_configs = {"vision_config": Qwen3_5VisionConfig, "text_config": Qwen3_5TextConfig} def __init__( self, text_config=None, vision_config=None, image_token_id=248056, video_token_id=248057, vision_start_token_id=248053, vision_end_token_id=248054, tie_word_embeddings=False, **kwargs, ): super().__init__( text_config=text_config, vision_config=vision_config, image_token_id=image_token_id, video_token_id=video_token_id, vision_start_token_id=vision_start_token_id, vision_end_token_id=vision_end_token_id, tie_word_embeddings=tie_word_embeddings, **kwargs, ) class Qwen3_5DynamicCache(Qwen3NextDynamicCache): pass class Qwen3_5VisionRotaryEmbedding(Qwen3VLVisionRotaryEmbedding): pass class Qwen3_5TextRotaryEmbedding(Qwen3VLTextRotaryEmbedding): def __init__(self, config: Qwen3_5TextConfig, device=None): super().__init__() self.mrope_section = config.rope_parameters.get("mrope_section", [11, 11, 10]) def compute_default_rope_parameters( config: Qwen3_5TextConfig | None = None, device: Optional["torch.device"] = None, seq_len: int | None = None, ) -> tuple["torch.Tensor", float]: base = config.rope_parameters["rope_theta"] partial_rotary_factor = config.rope_parameters.get("partial_rotary_factor", 1.0) head_dim = getattr(config, "head_dim", None) or config.hidden_size // config.num_attention_heads dim = int(head_dim * partial_rotary_factor) attention_factor = 1.0 # Unused in this type of RoPE # Compute the inverse frequencies inv_freq = 1.0 / ( base ** (torch.arange(0, dim, 2, dtype=torch.int64).to(device=device, dtype=torch.float) / dim) ) return inv_freq, attention_factor class Qwen3_5GatedDeltaNet(Qwen3NextGatedDeltaNet): def __init__(self, config: Qwen3_5Config, layer_idx: int): super().__init__(config, layer_idx) del projection_size_qkvz # noqa: F821 del projection_size_ba # noqa: F821 del self.in_proj_qkvz del self.in_proj_ba self.in_proj_qkv = nn.Linear(self.hidden_size, self.key_dim * 2 + self.value_dim, bias=False) self.in_proj_z = nn.Linear(self.hidden_size, self.value_dim, bias=False) self.in_proj_b = nn.Linear(self.hidden_size, self.num_v_heads, bias=False) self.in_proj_a = nn.Linear(self.hidden_size, self.num_v_heads, bias=False) def fix_query_key_value_ordering(self): raise AttributeError("Not needed for Qwen3.5 Series") def forward( self, hidden_states: torch.Tensor, cache_params: Qwen3_5DynamicCache | None = None, cache_position: torch.LongTensor | None = None, attention_mask: torch.Tensor | None = None, ): hidden_states = apply_mask_to_padding_states(hidden_states, attention_mask) # Set up dimensions for reshapes later batch_size, seq_len, _ = hidden_states.shape use_precomputed_states = ( cache_params is not None and cache_params.has_previous_state and seq_len == 1 and cache_position is not None ) # getting projected states from cache if it exists if cache_params is not None: conv_state = cache_params.conv_states[self.layer_idx] recurrent_state = cache_params.recurrent_states[self.layer_idx] mixed_qkv = self.in_proj_qkv(hidden_states) mixed_qkv = mixed_qkv.transpose(1, 2) z = self.in_proj_z(hidden_states) z = z.reshape(batch_size, seq_len, -1, self.head_v_dim) b = self.in_proj_b(hidden_states) a = self.in_proj_a(hidden_states) if use_precomputed_states: # 2. Convolution sequence transformation # NOTE: the conv state is updated in `causal_conv1d_update` mixed_qkv = self.causal_conv1d_update( mixed_qkv, conv_state, self.conv1d.weight.squeeze(1), self.conv1d.bias, self.activation, ) else: if cache_params is not None: conv_state = F.pad(mixed_qkv, (self.conv_kernel_size - mixed_qkv.shape[-1], 0)) cache_params.conv_states[self.layer_idx] = conv_state if self.causal_conv1d_fn is not None: mixed_qkv = self.causal_conv1d_fn( x=mixed_qkv, weight=self.conv1d.weight.squeeze(1), bias=self.conv1d.bias, activation=self.activation, seq_idx=None, ) else: mixed_qkv = F.silu(self.conv1d(mixed_qkv)[:, :, :seq_len]) mixed_qkv = mixed_qkv.transpose(1, 2) query, key, value = torch.split( mixed_qkv, [ self.key_dim, self.key_dim, self.value_dim, ], dim=-1, ) query = query.reshape(batch_size, seq_len, -1, self.head_k_dim) key = key.reshape(batch_size, seq_len, -1, self.head_k_dim) value = value.reshape(batch_size, seq_len, -1, self.head_v_dim) beta = b.sigmoid() # If the model is loaded in fp16, without the .float() here, A might be -inf g = -self.A_log.float().exp() * F.softplus(a.float() + self.dt_bias) if self.num_v_heads // self.num_k_heads > 1: query = query.repeat_interleave(self.num_v_heads // self.num_k_heads, dim=2) key = key.repeat_interleave(self.num_v_heads // self.num_k_heads, dim=2) if not use_precomputed_states: core_attn_out, last_recurrent_state = self.chunk_gated_delta_rule( query, key, value, g=g, beta=beta, initial_state=None, output_final_state=cache_params is not None, use_qk_l2norm_in_kernel=True, ) else: core_attn_out, last_recurrent_state = self.recurrent_gated_delta_rule( query, key, value, g=g, beta=beta, initial_state=recurrent_state, output_final_state=cache_params is not None, use_qk_l2norm_in_kernel=True, ) # Update cache if cache_params is not None: cache_params.recurrent_states[self.layer_idx] = last_recurrent_state # reshape input data into 2D tensor core_attn_out = core_attn_out.reshape(-1, self.head_v_dim) z = z.reshape(-1, self.head_v_dim) core_attn_out = self.norm(core_attn_out, z) core_attn_out = core_attn_out.reshape(batch_size, seq_len, -1) output = self.out_proj(core_attn_out) return output class Qwen3_5Attention(Qwen3NextAttention): pass class Qwen3_5MLP(Qwen3NextMLP): def __init__(self, config: Qwen3_5Config, intermediate_size: int): super().__init__(config, intermediate_size) self.intermediate_size = intermediate_size class Qwen3_5RMSNorm(Qwen3NextRMSNorm): pass class Qwen3_5DecoderLayer(GradientCheckpointingLayer): def __init__(self, config: Qwen3_5TextConfig, layer_idx: int): super().__init__() self.hidden_size = config.hidden_size self.layer_type = config.layer_types[layer_idx] if self.layer_type == "linear_attention": self.linear_attn = Qwen3_5GatedDeltaNet(config, layer_idx) elif self.layer_type == "full_attention": self.self_attn = Qwen3_5Attention(config, layer_idx) self.mlp = Qwen3_5MLP(config, config.intermediate_size) self.input_layernorm = Qwen3_5RMSNorm(config.hidden_size, eps=config.rms_norm_eps) self.post_attention_layernorm = Qwen3_5RMSNorm(config.hidden_size, eps=config.rms_norm_eps) def forward( self, hidden_states: torch.Tensor, position_embeddings: tuple[torch.Tensor, torch.Tensor], attention_mask: torch.Tensor | None = None, position_ids: torch.LongTensor | None = None, past_key_values: Cache | None = None, cache_position: torch.LongTensor | None = None, **kwargs: Unpack[TransformersKwargs], ) -> torch.FloatTensor: residual = hidden_states hidden_states = self.input_layernorm(hidden_states) # Token Mixer if self.layer_type == "linear_attention": hidden_states = self.linear_attn( hidden_states=hidden_states, cache_params=past_key_values, cache_position=cache_position, attention_mask=attention_mask, ) elif self.layer_type == "full_attention": # Self Attention hidden_states, _ = self.self_attn( hidden_states=hidden_states, attention_mask=attention_mask, position_ids=position_ids, past_key_values=past_key_values, cache_position=cache_position, position_embeddings=position_embeddings, **kwargs, ) hidden_states = residual + hidden_states # Fully Connected residual = hidden_states hidden_states = self.post_attention_layernorm(hidden_states) hidden_states = self.mlp(hidden_states) hidden_states = residual + hidden_states return hidden_states class Qwen3_5PreTrainedModel(Qwen3NextPreTrainedModel): config: Qwen3_5Config _no_split_modules = ["Qwen3_5DecoderLayer", "Qwen3_5VisionBlock"] _can_record_outputs = { "hidden_states": Qwen3_5DecoderLayer, "attentions": Qwen3_5Attention, } @torch.no_grad() def _init_weights(self, module): PreTrainedModel._init_weights(self, module) if isinstance(module, Qwen3_5GatedDeltaNet): init.ones_(module.dt_bias) init.copy_(module.A_log, torch.empty_like(module.A_log).uniform_(0, 16).log_()) # We initialize with 0s to be 1 centered as the RMSNorm here does (1 + weight) elif isinstance(module, Qwen3_5RMSNorm): init.zeros_(module.weight) elif isinstance(module, Qwen3_5VisionRotaryEmbedding): inv_freq = 1.0 / (module.theta ** (torch.arange(0, module.dim, 2, dtype=torch.float) / module.dim)) init.copy_(module.inv_freq, inv_freq) class Qwen3_5VisionModel(Qwen3VLVisionModel): config: Qwen3_5VisionConfig _no_split_modules = ["Qwen3_5VisionBlock"] def __init__(self, config, *inputs, **kwargs) -> None: super().__init__(config, *inputs, **kwargs) del self.deepstack_visual_indexes del self.deepstack_merger_list @merge_with_config_defaults @capture_outputs def forward(self, hidden_states: torch.Tensor, grid_thw: torch.Tensor, **kwargs) -> torch.Tensor: """ Args: hidden_states (`torch.Tensor` of shape `(seq_len, hidden_size)`): The final hidden states of the model. grid_thw (`torch.Tensor` of shape `(num_images_or_videos, 3)`): The temporal, height and width of feature shape of each image in LLM. Returns: `torch.Tensor`: hidden_states. """ hidden_states = self.patch_embed(hidden_states) pos_embeds = self.fast_pos_embed_interpolate(grid_thw) hidden_states = hidden_states + pos_embeds rotary_pos_emb = self.rot_pos_emb(grid_thw) seq_len, _ = hidden_states.size() hidden_states = hidden_states.reshape(seq_len, -1) 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()) cu_seqlens = torch.repeat_interleave(grid_thw[:, 1] * grid_thw[:, 2], grid_thw[:, 0]).cumsum( dim=0, # Select dtype based on the following factors: # - FA2 requires that cu_seqlens_q must have dtype int32 # - torch.onnx.export requires that cu_seqlens_q must have same dtype as grid_thw # See https://github.com/huggingface/transformers/pull/34852 for more information dtype=grid_thw.dtype if torch.jit.is_tracing() else torch.int32, ) cu_seqlens = F.pad(cu_seqlens, (1, 0), value=0) for blk in self.blocks: hidden_states = blk( hidden_states, cu_seqlens=cu_seqlens, position_embeddings=position_embeddings, **kwargs, ) merged_hidden_states = self.merger(hidden_states) return BaseModelOutputWithPooling( last_hidden_state=hidden_states, pooler_output=merged_hidden_states, ) class Qwen3_5ModelOutputWithPast(Qwen3VLModelOutputWithPast): pass class Qwen3_5TextModel(Qwen3NextModel): def __init__(self, config: Qwen3_5TextConfig): super().__init__(config) self.rotary_emb = Qwen3_5TextRotaryEmbedding(config=config) def forward( self, input_ids: torch.LongTensor | None = None, attention_mask: torch.Tensor | None = None, position_ids: torch.LongTensor | None = None, past_key_values: Cache | None = None, inputs_embeds: torch.FloatTensor | None = None, use_cache: bool | None = None, cache_position: torch.LongTensor | None = None, **kwargs: Unpack[TransformersKwargs], ) -> BaseModelOutputWithPast: if (input_ids is None) ^ (inputs_embeds is not None): raise ValueError("You must specify exactly one of input_ids or inputs_embeds") if inputs_embeds is None: inputs_embeds = self.embed_tokens(input_ids) if use_cache and past_key_values is None: past_key_values = Qwen3_5DynamicCache(config=self.config) if cache_position is None: past_seen_tokens = past_key_values.get_seq_length() if past_key_values is not None else 0 cache_position = torch.arange( past_seen_tokens, past_seen_tokens + inputs_embeds.shape[1], device=inputs_embeds.device ) # mrope: the hard coded `3` is for temporal, height and width. if position_ids is None: position_ids = cache_position.view(1, 1, -1).expand(3, inputs_embeds.shape[0], -1) elif position_ids.ndim == 2: position_ids = position_ids[None, ...].expand(3, position_ids.shape[0], -1) if position_ids.ndim == 3 and position_ids.shape[0] == 4: text_position_ids = position_ids[0] position_ids = position_ids[1:] else: text_position_ids = position_ids[0] causal_mask = create_causal_mask( config=self.config, inputs_embeds=inputs_embeds, attention_mask=attention_mask, cache_position=cache_position, past_key_values=past_key_values, position_ids=text_position_ids, ) linear_attn_mask = self._update_linear_attn_mask(attention_mask, cache_position) hidden_states = inputs_embeds position_embeddings = self.rotary_emb(hidden_states, position_ids) for layer_idx, decoder_layer in enumerate(self.layers[: self.config.num_hidden_layers]): layer_mask = linear_attn_mask if decoder_layer.layer_type == "linear_attention" else causal_mask hidden_states = decoder_layer( hidden_states, position_embeddings=position_embeddings, attention_mask=layer_mask, position_ids=position_ids, past_key_values=past_key_values, use_cache=use_cache, cache_position=cache_position, **kwargs, ) hidden_states = self.norm(hidden_states) return Qwen3_5ModelOutputWithPast( last_hidden_state=hidden_states, past_key_values=past_key_values, ) class Qwen3_5Model(Qwen3VLModel): def get_video_features( self, **super_kwargs, ) -> tuple | BaseModelOutputWithPooling: # Same implementation as for images return super().get_video_features(**super_kwargs) def get_image_features( self, pixel_values: torch.FloatTensor, image_grid_thw: torch.LongTensor | None = None, **kwargs: Unpack[TransformersKwargs], ) -> tuple | BaseModelOutputWithPooling: pixel_values = pixel_values.type(self.visual.dtype) vision_output: BaseModelOutputWithPooling = self.visual( pixel_values, grid_thw=image_grid_thw, return_dict=True, **kwargs ) image_embeds = vision_output.pooler_output split_sizes = (image_grid_thw.prod(-1) // self.visual.spatial_merge_size**2).tolist() image_embeds = torch.split(image_embeds, split_sizes) vision_output.pooler_output = image_embeds return vision_output @auto_docstring @can_return_tuple def forward( self, input_ids: torch.LongTensor = None, attention_mask: torch.Tensor | None = None, position_ids: torch.LongTensor | None = None, past_key_values: Cache | None = None, inputs_embeds: torch.FloatTensor | None = None, pixel_values: torch.Tensor | None = None, pixel_values_videos: torch.FloatTensor | None = None, image_grid_thw: torch.LongTensor | None = None, video_grid_thw: torch.LongTensor | None = None, cache_position: torch.LongTensor | None = None, **kwargs: Unpack[TransformersKwargs], ) -> tuple | Qwen3_5ModelOutputWithPast: r""" image_grid_thw (`torch.LongTensor` of shape `(num_images, 3)`, *optional*): The temporal, height and width of feature shape of each image in LLM. video_grid_thw (`torch.LongTensor` of shape `(num_videos, 3)`, *optional*): The temporal, height and width of feature shape of each video in LLM. """ if (input_ids is None) ^ (inputs_embeds is not None): raise ValueError("You must specify exactly one of input_ids or inputs_embeds") if inputs_embeds is None: inputs_embeds = self.get_input_embeddings()(input_ids) if pixel_values is not None: image_outputs: BaseModelOutputWithPooling = self.get_image_features( pixel_values, image_grid_thw, return_dict=True ) image_embeds = image_outputs.pooler_output image_embeds = torch.cat(image_embeds, dim=0).to(inputs_embeds.device, inputs_embeds.dtype) image_mask, _ = self.get_placeholder_mask( input_ids, inputs_embeds=inputs_embeds, image_features=image_embeds ) inputs_embeds = inputs_embeds.masked_scatter(image_mask, image_embeds) if pixel_values_videos is not None: video_outputs: BaseModelOutputWithPooling = self.get_video_features( pixel_values_videos, video_grid_thw, return_dict=True ) video_embeds = video_outputs.pooler_output video_embeds = torch.cat(video_embeds, dim=0).to(inputs_embeds.device, inputs_embeds.dtype) _, video_mask = self.get_placeholder_mask( input_ids, inputs_embeds=inputs_embeds, video_features=video_embeds ) inputs_embeds = inputs_embeds.masked_scatter(video_mask, video_embeds) if position_ids is None: position_ids = self.compute_3d_position_ids( input_ids=input_ids, image_grid_thw=image_grid_thw, video_grid_thw=video_grid_thw, inputs_embeds=inputs_embeds, attention_mask=attention_mask, past_key_values=past_key_values, ) outputs = self.language_model( input_ids=None, position_ids=position_ids, attention_mask=attention_mask, past_key_values=past_key_values, inputs_embeds=inputs_embeds, cache_position=cache_position, **kwargs, ) return Qwen3_5ModelOutputWithPast( **outputs, rope_deltas=self.rope_deltas, ) class Qwen3_5ForCausalLM(Qwen3ForCausalLM): config: Qwen3_5TextConfig _keys_to_ignore_on_load_unexpected = [r"^mtp.*", r"^model.visual.*"] def __init__(self, config): super().__init__(config) self.model = Qwen3_5TextModel(config) class Qwen3_5ForConditionalGeneration(Qwen3VLForConditionalGeneration): def get_video_features( self, **super_kwargs, ) -> tuple | BaseModelOutputWithPooling: return super().get_video_features(**super_kwargs) def get_image_features( self, **super_kwargs, ) -> tuple | BaseModelOutputWithPooling: return super().get_image_features(**super_kwargs) __all__ = [ "Qwen3_5Config", "Qwen3_5TextConfig", "Qwen3_5VisionModel", "Qwen3_5TextModel", "Qwen3_5Model", "Qwen3_5ForCausalLM", "Qwen3_5ForConditionalGeneration", "Qwen3_5PreTrainedModel", ]