# 🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨 # This file was automatically generated from src/transformers/models/qwen3_asr/modular_qwen3_asr.py. # Do NOT edit this file manually as any edits will be overwritten by the generation of # the file from the modular. If any change should be done, please apply the change to the # modular_qwen3_asr.py file directly. One of our CI enforces this. # 🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨🚨 # Copyright 2026 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. import math from collections.abc import Callable from dataclasses import dataclass import numpy as np import torch import torch.nn.functional as F from torch import nn from ... import initialization as init from ...activations import ACT2FN from ...cache_utils import Cache from ...generation import GenerationMixin from ...modeling_layers import GenericForTokenClassification, GradientCheckpointingLayer from ...modeling_outputs import BaseModelOutputWithPast, BaseModelOutputWithPooling, ModelOutput from ...modeling_utils import ALL_ATTENTION_FUNCTIONS, PreTrainedModel from ...processing_utils import Unpack from ...utils import TransformersKwargs, auto_docstring, can_return_tuple, torch_compilable_check from ...utils.generic import is_flash_attention_requested from ...utils.output_capturing import capture_outputs from ..auto import AutoModel from .configuration_qwen3_asr import Qwen3ASRConfig, Qwen3ASREncoderConfig @auto_docstring class Qwen3ASRPreTrainedModel(PreTrainedModel): config: Qwen3ASRConfig base_model_prefix = "model" input_modalities = ("audio", "text") supports_gradient_checkpointing = True _no_split_modules = ["Qwen3DecoderLayer"] _skip_keys_device_placement = ["past_key_values"] _supports_flash_attn = True _supports_sdpa = True _can_compile_fullgraph = True _supports_attention_backend = True def _init_weights(self, module): super()._init_weights(module) if isinstance(module, SinusoidsPositionEmbedding): position_embeddings = module.compute_default_singular_positional_embedding() init.copy_(module.positional_embedding, position_embeddings) def repeat_kv(hidden_states: torch.Tensor, n_rep: int) -> torch.Tensor: """ This is the equivalent of torch.repeat_interleave(x, dim=1, repeats=n_rep). The hidden states go from (batch, num_key_value_heads, seqlen, head_dim) to (batch, num_attention_heads, seqlen, head_dim) """ batch, num_key_value_heads, slen, head_dim = hidden_states.shape if n_rep == 1: return hidden_states hidden_states = hidden_states[:, :, None, :, :].expand(batch, num_key_value_heads, n_rep, slen, head_dim) return hidden_states.reshape(batch, num_key_value_heads * n_rep, slen, head_dim) def eager_attention_forward( module: nn.Module, query: torch.Tensor, key: torch.Tensor, value: torch.Tensor, attention_mask: torch.Tensor | None, scaling: float, dropout: float = 0.0, **kwargs, ): key_states = repeat_kv(key, module.num_key_value_groups) value_states = repeat_kv(value, module.num_key_value_groups) attn_weights = torch.matmul(query, key_states.transpose(2, 3)) * scaling if attention_mask is not None: attn_weights = attn_weights + attention_mask attn_weights = nn.functional.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query.dtype) attn_weights = nn.functional.dropout(attn_weights, p=dropout, training=module.training) attn_output = torch.matmul(attn_weights, value_states) attn_output = attn_output.transpose(1, 2).contiguous() return attn_output, attn_weights class Qwen3ASRAudioAttention(nn.Module): """Multi-headed attention from 'Attention Is All You Need' paper""" def __init__(self, config): super().__init__() self.embed_dim = config.d_model self.num_heads = config.encoder_attention_heads self.dropout = config.attention_dropout self.head_dim = self.embed_dim // self.num_heads self.num_key_value_groups = 1 # needed for eager attention self.config = config if (self.head_dim * self.num_heads) != self.embed_dim: raise ValueError( f"embed_dim must be divisible by num_heads (got `embed_dim`: {self.embed_dim}" f" and `num_heads`: {self.num_heads})." ) self.scaling = self.head_dim**-0.5 self.attention_dropout = 0.0 self.is_decoder = False self.is_causal = False self.k_proj = nn.Linear(self.embed_dim, self.embed_dim, bias=True) self.v_proj = nn.Linear(self.embed_dim, self.embed_dim, bias=True) self.q_proj = nn.Linear(self.embed_dim, self.embed_dim, bias=True) self.out_proj = nn.Linear(self.embed_dim, self.embed_dim, bias=True) def forward( self, hidden_states: torch.Tensor, cu_seqlens: torch.Tensor, **kwargs, ) -> torch.Tensor: """Input shape: Batch x Time x Channel""" seq_length, _ = hidden_states.size() query_states = self.q_proj(hidden_states).reshape(seq_length, self.num_heads, -1) key_states = self.k_proj(hidden_states).reshape(seq_length, self.num_heads, -1) value_states = self.v_proj(hidden_states).reshape(seq_length, self.num_heads, -1) query_states = query_states.transpose(0, 1).unsqueeze(0) key_states = key_states.transpose(0, 1).unsqueeze(0) value_states = value_states.transpose(0, 1).unsqueeze(0) attention_interface: Callable = ALL_ATTENTION_FUNCTIONS.get_interface( self.config._attn_implementation, eager_attention_forward ) if is_flash_attention_requested(self.config): # Flash Attention: Use cu_seqlens for variable length attention max_seqlen = (cu_seqlens[1:] - cu_seqlens[:-1]).max() attn_output, _ = attention_interface( self, query_states, key_states, value_states, attention_mask=None, scaling=self.scaling, dropout=0.0 if not self.training else self.attention_dropout, cu_seq_lens_q=cu_seqlens, cu_seq_lens_k=cu_seqlens, max_length_q=max_seqlen, max_length_k=max_seqlen, is_causal=False, **kwargs, ) else: # Other implementations: Process each chunk separately lengths = cu_seqlens[1:] - cu_seqlens[:-1] splits = [ torch.split(tensor, lengths.tolist(), dim=2) for tensor in (query_states, key_states, value_states) ] attn_outputs = [ attention_interface( self, q, k, v, attention_mask=None, scaling=self.scaling, dropout=0.0 if not self.training else self.attention_dropout, is_causal=False, **kwargs, )[0] for q, k, v in zip(*splits) ] attn_output = torch.cat(attn_outputs, dim=1) attn_output = attn_output.reshape(seq_length, -1).contiguous() attn_output = self.out_proj(attn_output) return attn_output class Qwen3ASRAudioEncoderLayer(GradientCheckpointingLayer): def __init__(self, config: Qwen3ASREncoderConfig): super().__init__() self.embed_dim = config.d_model self.self_attn = Qwen3ASRAudioAttention(config) self.self_attn_layer_norm = nn.LayerNorm(self.embed_dim) self.dropout = config.dropout self.activation_fn = ACT2FN[config.activation_function] self.activation_dropout = config.activation_dropout self.fc1 = nn.Linear(self.embed_dim, config.encoder_ffn_dim) self.fc2 = nn.Linear(config.encoder_ffn_dim, self.embed_dim) self.final_layer_norm = nn.LayerNorm(self.embed_dim) def forward( self, hidden_states: torch.Tensor, cu_seqlens: torch.Tensor, **kwargs, ) -> torch.Tensor: """ Args: hidden_states (`torch.FloatTensor`): input to the layer of shape `(batch, seq_len, embed_dim)` attention_mask (`torch.FloatTensor`): attention mask of size `(batch, 1, tgt_len, src_len)` where padding elements are indicated by very large negative values. """ residual = hidden_states hidden_states = self.self_attn_layer_norm(hidden_states) hidden_states = self.self_attn( hidden_states=hidden_states, cu_seqlens=cu_seqlens, **kwargs, ) hidden_states = residual + hidden_states residual = hidden_states hidden_states = self.final_layer_norm(hidden_states) hidden_states = self.fc1(hidden_states) hidden_states = self.activation_fn(hidden_states) hidden_states = self.fc2(hidden_states) hidden_states = residual + hidden_states if hidden_states.dtype == torch.float16: clamp_value = torch.finfo(hidden_states.dtype).max - 1000 hidden_states = torch.clamp(hidden_states, min=-clamp_value, max=clamp_value) outputs = (hidden_states,) return outputs class SinusoidsPositionEmbedding(nn.Module): def __init__(self, length, channels, max_timescale=10000): super().__init__() self.length = length self.channels = channels self.max_timescale = max_timescale if channels % 2 != 0: raise ValueError("SinusoidsPositionEmbedding needs even channels input") position_embedding = self.compute_default_singular_positional_embedding() self.register_buffer("positional_embedding", position_embedding, persistent=False) def compute_default_singular_positional_embedding(self): log_timescale_increment = np.log(self.max_timescale) / (self.channels // 2 - 1) inv_timescales = torch.exp(-log_timescale_increment * torch.arange(self.channels // 2).float()) scaled_time = torch.arange(self.length)[:, np.newaxis] * inv_timescales[np.newaxis, :] return torch.cat([torch.sin(scaled_time), torch.cos(scaled_time)], dim=1) def forward(self, seqlen: int): return self.positional_embedding[:seqlen, :] def _get_feat_extract_output_lengths(input_lengths, n_window=50): """ Computes the output length of the convolutional layers and the output length of the audio encoder """ chunk_len = n_window * 2 input_lengths_leave = input_lengths % chunk_len feat_lengths = (input_lengths_leave - 1) // 2 + 1 return ((feat_lengths - 1) // 2 + 1 - 1) // 2 + 1 + (input_lengths // chunk_len) * 13 def get_audio_cu_seqlens( chunk_lengths: torch.Tensor, feature_lens: torch.Tensor, n_window_infer: int, n_window: int, kwargs: dict | None = None, ) -> torch.Tensor: """Compute cumulative sequence lengths for audio attention windowing, or pop `"cu_seqlens"` from `kwargs` if precomputed. Splits each sample's post-CNN features into inference windows and returns cumulative boundaries for flash-attention-style sequence packing. Args: chunk_lengths: `(num_chunks,)` pre-CNN chunk lengths. feature_lens: `(batch_size,)` per-sample frame counts. n_window_infer: inference window size (in raw frames). n_window: half the chunk size (in raw frames). kwargs: optional caller kwargs — if it contains `"cu_seqlens"` it is popped and returned. Returns: `(num_windows + 1,)` int32 cumulative sequence boundaries. """ if kwargs is not None and (cu_seqlens := kwargs.pop("cu_seqlens", None)) is not None: return cu_seqlens aftercnn_lens = _get_feat_extract_output_lengths(feature_lens, n_window) feature_lens_after_cnn = _get_feat_extract_output_lengths(chunk_lengths, n_window) max_len_after_cnn = feature_lens_after_cnn.max().item() n_window_ratio = n_window_infer // (n_window * 2) window_aftercnn = max_len_after_cnn * n_window_ratio cu_chunk_lens = [0] for cnn_len in aftercnn_lens: cu_chunk_lens += [window_aftercnn] * (cnn_len // window_aftercnn) remainder = cnn_len % window_aftercnn if remainder != 0: cu_chunk_lens += [remainder] return torch.tensor(cu_chunk_lens, device=feature_lens.device).cumsum(-1, dtype=torch.int32) @auto_docstring( custom_intro=""" The audio model for Qwen3 ASR without any head or projection on top. """ ) class Qwen3ASREncoder(Qwen3ASRPreTrainedModel): config: Qwen3ASREncoderConfig main_input_name = "input_features" input_modalities = "audio" _no_split_modules = ["Qwen3ASRAudioEncoderLayer"] _supports_sdpa = True _can_record_outputs = { "hidden_states": Qwen3ASRAudioEncoderLayer, "attentions": Qwen3ASRAudioAttention, } def __init__(self, config: Qwen3ASREncoderConfig): super().__init__(config) self.dropout = config.dropout embed_dim = config.d_model self.embed_scale = math.sqrt(embed_dim) if config.scale_embedding else 1.0 self.n_window = config.n_window self.positional_embedding = SinusoidsPositionEmbedding(config.max_position_embeddings, embed_dim) self.layers = nn.ModuleList([Qwen3ASRAudioEncoderLayer(config) for _ in range(config.encoder_layers)]) self.ln_post = nn.LayerNorm(config.d_model) self.gradient_checkpointing = False self.conv2d1 = nn.Conv2d(1, config.downsample_hidden_size, 3, 2, padding=1) self.conv2d2 = nn.Conv2d(config.downsample_hidden_size, config.downsample_hidden_size, 3, 2, padding=1) self.conv2d3 = nn.Conv2d(config.downsample_hidden_size, config.downsample_hidden_size, 3, 2, padding=1) self.conv_out = nn.Linear( config.downsample_hidden_size * ((((config.num_mel_bins + 1) // 2 + 1) // 2 + 1) // 2), config.d_model, bias=False, ) self.n_window_infer = self.config.n_window_infer # Initialize weights and apply final processing self.post_init() def _freeze_parameters(self): for param in self.parameters(): param.requires_grad = False self._requires_grad = False def get_input_embeddings(self) -> nn.Module: return self.conv2d1 def set_input_embeddings(self, value): self.conv2d1 = value @capture_outputs(tie_last_hidden_states=False) @auto_docstring def forward( self, input_features: torch.Tensor, input_features_mask: torch.Tensor, **kwargs, ) -> BaseModelOutputWithPooling: r""" input_features_mask (`torch.LongTensor` of shape `(batch_size, padded_feature_length)`): 1 for valid mel frames and 0 for padding. """ batch_size, num_mel_bins, padded_feature_length = input_features.shape chunk_len = self.n_window * 2 if padded_feature_length % chunk_len != 0: raise ValueError( f"Qwen3ASREncoder expects `padded_feature_length` to be a multiple of " f"`n_window * 2` ({chunk_len}), but got {padded_feature_length}." ) num_chunks = padded_feature_length // chunk_len # Compute cu_seqlens for windowed attention feature_lens = input_features_mask.sum(-1).to(torch.long) chunk_lengths = ( input_features_mask.view(batch_size, num_chunks, chunk_len).sum(dim=-1).reshape(-1).to(torch.long) ) cu_seqlens = get_audio_cu_seqlens( chunk_lengths, feature_lens, self.n_window_infer, self.n_window, kwargs=kwargs ) # Chunk and process through CNN chunked = ( input_features.view(batch_size, num_mel_bins, num_chunks, chunk_len) .permute(0, 2, 1, 3) .reshape(batch_size * num_chunks, 1, num_mel_bins, chunk_len) ) conv_out = F.gelu(self.conv2d1(chunked)) conv_out = F.gelu(self.conv2d2(conv_out)) conv_out = F.gelu(self.conv2d3(conv_out)) total_chunks, conv_channels, freq_bins, time_steps = conv_out.size() conv_out = self.conv_out( conv_out.permute(0, 3, 1, 2).contiguous().view(total_chunks, time_steps, conv_channels * freq_bins) ) conv_out += self.positional_embedding.positional_embedding[:time_steps].to(conv_out.dtype) # Select only valid (non-padding) post-CNN positions into a flat packed sequence chunk_post_cnn_lens = self._post_cnn_length( input_features_mask.view(batch_size, num_chunks, chunk_len).sum(dim=-1).reshape(-1).to(torch.long) ) valid_mask = torch.arange(time_steps, device=input_features.device) < chunk_post_cnn_lens.unsqueeze(1) valid_indices = valid_mask.flatten().nonzero().squeeze(-1) hidden_states = torch.index_select(conv_out.reshape(-1, conv_out.shape[-1]), 0, valid_indices) for encoder_layer in self.layers: layer_outputs = encoder_layer(hidden_states, cu_seqlens, **kwargs) hidden_states = layer_outputs[0] hidden_states = self.ln_post(hidden_states) return BaseModelOutputWithPooling(last_hidden_state=hidden_states) @staticmethod def _post_cnn_length(lengths: torch.Tensor) -> torch.Tensor: """Length after three (k=3, s=2, p=1) convolutions; zero-length input stays zero.""" for _ in range(3): lengths = torch.where(lengths > 0, (lengths - 1) // 2 + 1, torch.zeros_like(lengths)) return lengths class Qwen3ASRMultiModalProjector(nn.Module): def __init__(self, config: Qwen3ASRConfig): super().__init__() self.linear_1 = nn.Linear(config.audio_config.d_model, config.audio_config.d_model) self.act = ACT2FN[config.audio_config.activation_function] self.linear_2 = nn.Linear(config.audio_config.d_model, config.audio_config.output_dim) def forward(self, audio_features): hidden_states = self.linear_1(audio_features) hidden_states = self.act(hidden_states) hidden_states = self.linear_2(hidden_states) return hidden_states @dataclass class Qwen3ASRModelOutputWithPast(BaseModelOutputWithPast): r""" audio_hidden_states (`torch.FloatTensor`, *optional*): Projected audio hidden states. """ audio_hidden_states: torch.FloatTensor | None = None @auto_docstring( custom_intro=""" The Qwen3ASR model (fine-tuned Whisper encoder, multi-modal projector, Qwen2 language model), without a language modeling head. """ ) class Qwen3ASRModel(Qwen3ASRPreTrainedModel): _tp_plan = None _pp_plan = None _keep_in_fp32_modules_strict = None def __init__(self, config): super().__init__(config) self.audio_tower = AutoModel.from_config(config.audio_config) self.language_model = AutoModel.from_config(config.text_config) self.multi_modal_projector = Qwen3ASRMultiModalProjector(config) self.post_init() @can_return_tuple @auto_docstring( custom_intro="This method is used to get the audio embeddings from input features (a log mel spectrogram)." ) def get_audio_features( self, input_features: torch.FloatTensor, input_features_mask: torch.LongTensor, **kwargs: Unpack[TransformersKwargs], ) -> tuple | BaseModelOutputWithPooling: r""" input_features_mask (`torch.LongTensor` of shape `(batch_size, padded_feature_length)`): 1 for valid mel frames and 0 for padding. """ audio_output = self.audio_tower( input_features=input_features, input_features_mask=input_features_mask, **kwargs, ) audio_output.pooler_output = self.multi_modal_projector(audio_output.last_hidden_state) return audio_output def get_placeholder_mask( self, input_ids: torch.LongTensor, inputs_embeds: torch.FloatTensor, audio_features: torch.FloatTensor ): """ Obtains multimodal placeholder mask from `input_ids` or `inputs_embeds`, and checks that the placeholder token count is equal to the length of multimodal features. If the lengths are different, an error is raised. """ if input_ids is None: special_audio_mask = inputs_embeds == self.get_input_embeddings()( torch.tensor(self.config.audio_token_id, dtype=torch.long, device=inputs_embeds.device) ) special_audio_mask = special_audio_mask.all(-1) else: special_audio_mask = input_ids == self.config.audio_token_id n_audio_tokens = special_audio_mask.sum() n_audio_features = audio_features.shape[0] special_audio_mask = special_audio_mask.unsqueeze(-1).expand_as(inputs_embeds).to(inputs_embeds.device) torch_compilable_check( inputs_embeds[special_audio_mask].numel() == audio_features.numel(), f"Audio features and audio tokens do not match, tokens: {n_audio_tokens}, features: {n_audio_features}", ) return special_audio_mask @can_return_tuple @auto_docstring def forward( self, input_ids: torch.LongTensor | None = None, input_features: torch.FloatTensor | None = None, input_features_mask: torch.Tensor | 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, **kwargs: Unpack[TransformersKwargs], ) -> tuple | Qwen3ASRModelOutputWithPast: r""" input_features_mask (`torch.Tensor` of shape `(batch_size, feature_sequence_length)`): Mask to avoid performing attention on padding feature indices. """ if inputs_embeds is None: inputs_embeds = self.get_input_embeddings()(input_ids) audio_embeds = None if input_features is not None and input_ids is not None: audio_embeds = self.get_audio_features(input_features, input_features_mask, return_dict=True).pooler_output # replace text-audio token placeholders with audio embeddings special_audio_mask = self.get_placeholder_mask( input_ids, inputs_embeds=inputs_embeds, audio_features=audio_embeds ) inputs_embeds = inputs_embeds.masked_scatter(special_audio_mask, audio_embeds.to(inputs_embeds.device)) outputs = self.language_model( inputs_embeds=inputs_embeds, attention_mask=attention_mask, position_ids=position_ids, past_key_values=past_key_values, use_cache=use_cache, **kwargs, ) return Qwen3ASRModelOutputWithPast( last_hidden_state=outputs.last_hidden_state, past_key_values=outputs.past_key_values, hidden_states=outputs.hidden_states, attentions=outputs.attentions, audio_hidden_states=audio_embeds, ) @auto_docstring( custom_intro=""" Base class for Qwen3ASR causal language model (or autoregressive) outputs. """ ) @dataclass class Qwen3ASRCausalLMOutputWithPast(ModelOutput): r""" loss (`torch.FloatTensor` of shape `(1,)`, *optional*, returned when `labels` is provided): Language modeling loss (for next-token prediction). logits (`torch.FloatTensor` of shape `(batch_size, sequence_length, config.vocab_size)`): Prediction scores of the language modeling head. past_key_values (`Cache`, *optional*, returned when `use_cache=True` is passed or when `config.use_cache=True`): It is a [`~cache_utils.Cache`] instance. audio_hidden_states (`torch.FloatTensor`, *optional*): Hidden states of the audio encoder after projection. """ loss: torch.FloatTensor | None = None logits: torch.FloatTensor | None = None past_key_values: Cache | None = None hidden_states: tuple[torch.FloatTensor] | None = None attentions: tuple[torch.FloatTensor] | None = None audio_hidden_states: torch.FloatTensor | None = None @auto_docstring( custom_intro=""" The Qwen3ASR model which consists of an audio encoder and a language model. """ ) class Qwen3ASRForConditionalGeneration(Qwen3ASRPreTrainedModel, GenerationMixin): _tied_weights_keys = {"lm_head.weight": "model.language_model.embed_tokens.weight"} def __init__(self, config): super().__init__(config) self.model = Qwen3ASRModel(config) self.lm_head = nn.Linear(config.text_config.hidden_size, config.text_config.vocab_size, bias=False) self.post_init() def get_audio_features(self, input_features, input_features_mask, **kwargs): return self.model.get_audio_features(input_features, input_features_mask, **kwargs) @can_return_tuple @auto_docstring def forward( self, input_ids: torch.LongTensor | None = None, input_features: torch.FloatTensor | None = None, input_features_mask: torch.Tensor | 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, labels: torch.LongTensor | None = None, use_cache: bool | None = None, logits_to_keep: int | torch.Tensor = 0, **kwargs: Unpack[TransformersKwargs], ) -> tuple | Qwen3ASRCausalLMOutputWithPast: r""" input_features_mask (`torch.Tensor` of shape `(batch_size, feature_sequence_length)`): Mask to avoid performing attention on padding feature indices. labels (`torch.LongTensor` of shape `(batch_size, sequence_length)`, *optional*): Labels for computing the masked language modeling loss. Example: ```python >>> from transformers import Qwen3ASRForConditionalGeneration, AutoProcessor >>> model_id = "Qwen/Qwen3-ASR-1.7B-hf" >>> processor = AutoProcessor.from_pretrained(model_id) >>> model = Qwen3ASRForConditionalGeneration.from_pretrained(model_id, device_map="auto") ```""" outputs = self.model( input_ids=input_ids, input_features=input_features, input_features_mask=input_features_mask, attention_mask=attention_mask, position_ids=position_ids, past_key_values=past_key_values, inputs_embeds=inputs_embeds, use_cache=use_cache, **kwargs, ) hidden_states = outputs.last_hidden_state slice_indices = slice(-logits_to_keep, None) if isinstance(logits_to_keep, int) else logits_to_keep logits = self.lm_head(hidden_states[:, slice_indices, :]) loss = None if labels is not None: loss = self.loss_function( logits=logits, labels=labels, vocab_size=self.config.text_config.vocab_size, **kwargs ) return Qwen3ASRCausalLMOutputWithPast( loss=loss, logits=logits, past_key_values=outputs.past_key_values, hidden_states=outputs.hidden_states, attentions=outputs.attentions, audio_hidden_states=outputs.audio_hidden_states, ) def prepare_inputs_for_generation(self, *args, is_first_iteration: bool = False, **kwargs): input_features = kwargs.pop("input_features", None) input_features_mask = kwargs.pop("input_features_mask", None) model_inputs = super().prepare_inputs_for_generation(*args, **kwargs) if is_first_iteration or not model_inputs.get("use_cache", False): if input_features is not None: model_inputs["input_features"] = input_features if input_features_mask is not None: model_inputs["input_features_mask"] = input_features_mask return model_inputs @auto_docstring( custom_intro=""" The Qwen3 ASR model with a token classification head for timestamp prediction (forced alignment). """ ) class Qwen3ASRForTokenClassification(GenericForTokenClassification, Qwen3ASRPreTrainedModel): pass __all__ = [ "Qwen3ASREncoder", "Qwen3ASRForConditionalGeneration", "Qwen3ASRModel", "Qwen3ASRPreTrainedModel", "Qwen3ASRForTokenClassification", ]