Spaces:
Build error
Build error
| # coding=utf-8 | |
| # Copyright 2022 The OpenAI Authors 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. | |
| """TensorFlow Whisper model.""" | |
| from __future__ import annotations | |
| import math | |
| import random | |
| from typing import Dict, List, Optional, Tuple, Union | |
| import numpy as np | |
| import tensorflow as tf | |
| from ...activations_tf import get_tf_activation | |
| from ...generation.configuration_utils import GenerationConfig | |
| from ...generation.tf_logits_process import TFLogitsProcessorList | |
| from ...modeling_tf_outputs import ( | |
| TFBaseModelOutput, | |
| TFBaseModelOutputWithPastAndCrossAttentions, | |
| TFSeq2SeqLMOutput, | |
| TFSeq2SeqModelOutput, | |
| ) | |
| from ...modeling_tf_utils import ( | |
| TFCausalLanguageModelingLoss, | |
| TFModelInputType, | |
| TFPreTrainedModel, | |
| keras, | |
| keras_serializable, | |
| unpack_inputs, | |
| ) | |
| from ...tf_utils import check_embeddings_within_bounds, shape_list, stable_softmax | |
| from ...utils import add_start_docstrings, add_start_docstrings_to_model_forward, logging, replace_return_docstrings | |
| from .configuration_whisper import WhisperConfig | |
| from .tokenization_whisper import TASK_IDS, TO_LANGUAGE_CODE | |
| logger = logging.get_logger(__name__) | |
| _CONFIG_FOR_DOC = "WhisperConfig" | |
| LARGE_NEGATIVE = -1e8 | |
| def sinusoidal_embedding_init(shape, dtype=tf.float32) -> tf.Tensor: | |
| """Returns sinusoids for positional embedding""" | |
| length, channels = shape | |
| if channels % 2 != 0: | |
| raise ValueError( | |
| f"Number of channels has to be divisible by 2 for sinusoidal positional embeddings, got {channels} channels." | |
| ) | |
| log_timescale_increment = math.log(10000) / (channels // 2 - 1) | |
| inv_timescales = tf.exp(-log_timescale_increment * tf.range(channels // 2, dtype=tf.float32)) | |
| scaled_time = tf.reshape(tf.range(length, dtype=tf.float32), (-1, 1)) * tf.reshape(inv_timescales, (1, -1)) | |
| return tf.cast(tf.concat([tf.sin(scaled_time), tf.cos(scaled_time)], axis=1), dtype) | |
| # Copied from transformers.models.bart.modeling_tf_bart.shift_tokens_right | |
| def shift_tokens_right(input_ids: tf.Tensor, pad_token_id: int, decoder_start_token_id: int): | |
| pad_token_id = tf.cast(pad_token_id, input_ids.dtype) | |
| decoder_start_token_id = tf.cast(decoder_start_token_id, input_ids.dtype) | |
| start_tokens = tf.fill( | |
| (shape_list(input_ids)[0], 1), tf.convert_to_tensor(decoder_start_token_id, input_ids.dtype) | |
| ) | |
| shifted_input_ids = tf.concat([start_tokens, input_ids[:, :-1]], -1) | |
| # replace possible -100 values in labels by `pad_token_id` | |
| shifted_input_ids = tf.where( | |
| shifted_input_ids == -100, | |
| tf.fill(shape_list(shifted_input_ids), tf.convert_to_tensor(pad_token_id, input_ids.dtype)), | |
| shifted_input_ids, | |
| ) | |
| # "Verify that `labels` has only positive values and -100" | |
| assert_gte0 = tf.debugging.assert_greater_equal(shifted_input_ids, tf.constant(0, dtype=input_ids.dtype)) | |
| # Make sure the assertion op is called by wrapping the result in an identity no-op | |
| with tf.control_dependencies([assert_gte0]): | |
| shifted_input_ids = tf.identity(shifted_input_ids) | |
| return shifted_input_ids | |
| # Copied from transformers.models.bart.modeling_tf_bart._make_causal_mask | |
| def _make_causal_mask(input_ids_shape: tf.TensorShape, past_key_values_length: int = 0): | |
| """ | |
| Make causal mask used for bi-directional self-attention. | |
| """ | |
| bsz = input_ids_shape[0] | |
| tgt_len = input_ids_shape[1] | |
| mask = tf.ones((tgt_len, tgt_len)) * LARGE_NEGATIVE | |
| mask_cond = tf.range(shape_list(mask)[-1]) | |
| mask = tf.where(mask_cond < tf.reshape(mask_cond + 1, (shape_list(mask)[-1], 1)), 0.0, mask) | |
| if past_key_values_length > 0: | |
| mask = tf.concat([tf.zeros((tgt_len, past_key_values_length)), mask], axis=-1) | |
| return tf.tile(mask[None, None, :, :], (bsz, 1, 1, 1)) | |
| # Copied from transformers.models.bart.modeling_tf_bart._expand_mask | |
| def _expand_mask(mask: tf.Tensor, tgt_len: Optional[int] = None): | |
| """ | |
| Expands attention_mask from `[bsz, seq_len]` to `[bsz, 1, tgt_seq_len, src_seq_len]`. | |
| """ | |
| src_len = shape_list(mask)[1] | |
| tgt_len = tgt_len if tgt_len is not None else src_len | |
| one_cst = tf.constant(1.0) | |
| mask = tf.cast(mask, dtype=one_cst.dtype) | |
| expanded_mask = tf.tile(mask[:, None, None, :], (1, 1, tgt_len, 1)) | |
| return (one_cst - expanded_mask) * LARGE_NEGATIVE | |
| class TFWhisperPositionalEmbedding(keras.layers.Layer): | |
| def __init__( | |
| self, | |
| num_positions: int, | |
| embedding_dim: int, | |
| padding_idx: Optional[int] = None, | |
| embedding_initializer=None, | |
| **kwargs, | |
| ): | |
| super().__init__(**kwargs) | |
| self.num_positions = num_positions | |
| self.embedding_dim = embedding_dim | |
| self.padding_idx = padding_idx | |
| self.embedding_initializer = keras.initializers.get(embedding_initializer) | |
| def build(self, input_shape): | |
| self.weight = self.add_weight( | |
| name="weight", | |
| shape=[self.num_positions, self.embedding_dim], | |
| initializer=self.embedding_initializer, | |
| trainable=True, | |
| ) | |
| super().build(input_shape) | |
| def call(self, input_ids, past_key_values_length=0): | |
| past_key_values_length = tf.cast(past_key_values_length, tf.int32) | |
| gather_indices = tf.range(tf.shape(input_ids)[1], delta=1) + past_key_values_length | |
| return tf.gather(self.weight, gather_indices) | |
| class TFWhisperAttention(keras.layers.Layer): | |
| """Multi-headed attention from 'Attention Is All You Need' paper""" | |
| def __init__( | |
| self, | |
| embed_dim: int, | |
| num_heads: int, | |
| dropout: float = 0.0, | |
| is_decoder: bool = False, | |
| bias: bool = True, | |
| **kwargs, | |
| ): | |
| super().__init__(**kwargs) | |
| self.embed_dim = embed_dim | |
| self.num_heads = num_heads | |
| self.dropout = keras.layers.Dropout(dropout) | |
| self.head_dim = embed_dim // num_heads | |
| if (self.head_dim * 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`: {num_heads})." | |
| ) | |
| self.scaling = self.head_dim**-0.5 | |
| self.is_decoder = is_decoder | |
| self.k_proj = keras.layers.Dense(embed_dim, use_bias=False, name="k_proj") | |
| self.v_proj = keras.layers.Dense(embed_dim, use_bias=bias, name="v_proj") | |
| self.q_proj = keras.layers.Dense(embed_dim, use_bias=bias, name="q_proj") | |
| self.out_proj = keras.layers.Dense(embed_dim, use_bias=bias, name="out_proj") | |
| # Copied from transformers.models.bart.modeling_tf_bart.TFBartAttention._shape with BART->whisper | |
| def _shape(self, tensor: tf.Tensor, seq_len: int, bsz: int): | |
| return tf.transpose(tf.reshape(tensor, (bsz, seq_len, self.num_heads, self.head_dim)), (0, 2, 1, 3)) | |
| # Copied from transformers.models.bart.modeling_tf_bart.TFBartAttention.call with BART->whisper | |
| def call( | |
| self, | |
| hidden_states: tf.Tensor, | |
| key_value_states: tf.Tensor | None = None, | |
| past_key_value: Tuple[Tuple[tf.Tensor]] | None = None, | |
| attention_mask: tf.Tensor | None = None, | |
| layer_head_mask: tf.Tensor | None = None, | |
| training: Optional[bool] = False, | |
| ) -> Tuple[tf.Tensor, tf.Tensor | None]: | |
| """Input shape: Batch x Time x Channel""" | |
| # if key_value_states are provided this layer is used as a cross-attention layer | |
| # for the decoder | |
| is_cross_attention = key_value_states is not None | |
| bsz, tgt_len, embed_dim = shape_list(hidden_states) | |
| # get query proj | |
| query_states = self.q_proj(hidden_states) * self.scaling | |
| # get key, value proj | |
| if is_cross_attention and past_key_value is not None: | |
| # reuse k,v, cross_attentions | |
| key_states = past_key_value[0] | |
| value_states = past_key_value[1] | |
| elif is_cross_attention: | |
| # cross_attentions | |
| key_states = self._shape(self.k_proj(key_value_states), -1, bsz) | |
| value_states = self._shape(self.v_proj(key_value_states), -1, bsz) | |
| elif past_key_value is not None: | |
| # reuse k, v, self_attention | |
| key_states = self._shape(self.k_proj(hidden_states), -1, bsz) | |
| value_states = self._shape(self.v_proj(hidden_states), -1, bsz) | |
| key_states = tf.concat([past_key_value[0], key_states], axis=2) | |
| value_states = tf.concat([past_key_value[1], value_states], axis=2) | |
| else: | |
| # self_attention | |
| key_states = self._shape(self.k_proj(hidden_states), -1, bsz) | |
| value_states = self._shape(self.v_proj(hidden_states), -1, bsz) | |
| if self.is_decoder: | |
| # if cross_attention save Tuple(tf.Tensor, tf.Tensor) of all cross attention key/value_states. | |
| # Further calls to cross_attention layer can then reuse all cross-attention | |
| # key/value_states (first "if" case) | |
| # if uni-directional self-attention (decoder) save Tuple(tf.Tensor, tf.Tensor) of | |
| # all previous decoder key/value_states. Further calls to uni-directional self-attention | |
| # can concat previous decoder key/value_states to current projected key/value_states (third "elif" case) | |
| # if encoder bi-directional self-attention `past_key_value` is always `None` | |
| past_key_value = (key_states, value_states) | |
| proj_shape = (bsz * self.num_heads, -1, self.head_dim) | |
| query_states = tf.reshape(self._shape(query_states, tgt_len, bsz), proj_shape) | |
| key_states = tf.reshape(key_states, proj_shape) | |
| value_states = tf.reshape(value_states, proj_shape) | |
| src_len = shape_list(key_states)[1] | |
| attn_weights = tf.matmul(query_states, key_states, transpose_b=True) | |
| tf.debugging.assert_equal( | |
| shape_list(attn_weights), | |
| [bsz * self.num_heads, tgt_len, src_len], | |
| message=( | |
| f"Attention weights should be of size {(bsz * self.num_heads, tgt_len, src_len)}, but is" | |
| f" {shape_list(attn_weights)}" | |
| ), | |
| ) | |
| if attention_mask is not None: | |
| tf.debugging.assert_equal( | |
| shape_list(attention_mask), | |
| [bsz, 1, tgt_len, src_len], | |
| message=( | |
| f"Attention mask should be of size {(bsz, 1, tgt_len, src_len)}, but is" | |
| f" {shape_list(attention_mask)}" | |
| ), | |
| ) | |
| attention_mask = tf.cast(attention_mask, dtype=attn_weights.dtype) | |
| attn_weights = tf.reshape(attn_weights, (bsz, self.num_heads, tgt_len, src_len)) + attention_mask | |
| attn_weights = tf.reshape(attn_weights, (bsz * self.num_heads, tgt_len, src_len)) | |
| attn_weights = stable_softmax(attn_weights, axis=-1) | |
| if layer_head_mask is not None: | |
| tf.debugging.assert_equal( | |
| shape_list(layer_head_mask), | |
| [self.num_heads], | |
| message=( | |
| f"Head mask for a single layer should be of size {(self.num_heads)}, but is" | |
| f" {shape_list(layer_head_mask)}" | |
| ), | |
| ) | |
| attn_weights = tf.reshape(layer_head_mask, (1, -1, 1, 1)) * tf.reshape( | |
| attn_weights, (bsz, self.num_heads, tgt_len, src_len) | |
| ) | |
| attn_weights = tf.reshape(attn_weights, (bsz * self.num_heads, tgt_len, src_len)) | |
| attn_probs = self.dropout(attn_weights, training=training) | |
| attn_output = tf.matmul(attn_probs, value_states) | |
| tf.debugging.assert_equal( | |
| shape_list(attn_output), | |
| [bsz * self.num_heads, tgt_len, self.head_dim], | |
| message=( | |
| f"`attn_output` should be of size {(bsz, self.num_heads, tgt_len, self.head_dim)}, but is" | |
| f" {shape_list(attn_output)}" | |
| ), | |
| ) | |
| attn_output = tf.transpose( | |
| tf.reshape(attn_output, (bsz, self.num_heads, tgt_len, self.head_dim)), (0, 2, 1, 3) | |
| ) | |
| attn_output = tf.reshape(attn_output, (bsz, tgt_len, embed_dim)) | |
| attn_output = self.out_proj(attn_output) | |
| attn_weights: tf.Tensor = tf.reshape(attn_weights, (bsz, self.num_heads, tgt_len, src_len)) | |
| return attn_output, attn_weights, past_key_value | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "k_proj", None) is not None: | |
| with tf.name_scope(self.k_proj.name): | |
| self.k_proj.build([None, None, self.embed_dim]) | |
| if getattr(self, "v_proj", None) is not None: | |
| with tf.name_scope(self.v_proj.name): | |
| self.v_proj.build([None, None, self.embed_dim]) | |
| if getattr(self, "q_proj", None) is not None: | |
| with tf.name_scope(self.q_proj.name): | |
| self.q_proj.build([None, None, self.embed_dim]) | |
| if getattr(self, "out_proj", None) is not None: | |
| with tf.name_scope(self.out_proj.name): | |
| self.out_proj.build([None, None, self.embed_dim]) | |
| # Copied from transformers.models.speech_to_text.modeling_tf_speech_to_text.TFSpeech2TextEncoderLayer with Speech2Text->Whisper | |
| class TFWhisperEncoderLayer(keras.layers.Layer): | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(**kwargs) | |
| self.embed_dim = config.d_model | |
| self.self_attn = TFWhisperAttention( | |
| self.embed_dim, config.encoder_attention_heads, dropout=config.attention_dropout, name="self_attn" | |
| ) | |
| self.self_attn_layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="self_attn_layer_norm") | |
| self.dropout = keras.layers.Dropout(config.dropout) | |
| self.activation_fn = get_tf_activation(config.activation_function) | |
| self.activation_dropout = keras.layers.Dropout(config.activation_dropout) | |
| self.fc1 = keras.layers.Dense(config.encoder_ffn_dim, name="fc1") | |
| self.fc2 = keras.layers.Dense(self.embed_dim, name="fc2") | |
| self.final_layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="final_layer_norm") | |
| self.config = config | |
| def call( | |
| self, hidden_states: tf.Tensor, attention_mask: tf.Tensor, layer_head_mask: tf.Tensor, training: bool = False | |
| ): | |
| """ | |
| Args: | |
| hidden_states (`tf.Tensor`): input to the layer of shape `(batch, seq_len, embed_dim)` | |
| attention_mask (`tf.Tensor`): attention mask of size | |
| `(batch, 1, tgt_len, src_len)` where padding elements are indicated by very large negative values. | |
| layer_head_mask (`tf.Tensor`): mask for attention heads in a given layer of size | |
| `(encoder_attention_heads,)` | |
| """ | |
| residual = hidden_states | |
| hidden_states = self.self_attn_layer_norm(hidden_states) | |
| hidden_states, self_attn_weights, _ = self.self_attn( | |
| hidden_states=hidden_states, | |
| attention_mask=attention_mask, | |
| layer_head_mask=layer_head_mask, | |
| training=training, | |
| ) | |
| tf.debugging.assert_equal( | |
| shape_list(hidden_states), | |
| shape_list(residual), | |
| message=f"Self attn modified the shape of query {shape_list(residual)} to {shape_list(hidden_states)}", | |
| ) | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| hidden_states = residual + hidden_states | |
| residual = hidden_states | |
| hidden_states = self.final_layer_norm(hidden_states) | |
| hidden_states = self.activation_fn(self.fc1(hidden_states)) | |
| hidden_states = self.activation_dropout(hidden_states, training=training) | |
| hidden_states = self.fc2(hidden_states) | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| hidden_states = residual + hidden_states | |
| return hidden_states, self_attn_weights | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "self_attn", None) is not None: | |
| with tf.name_scope(self.self_attn.name): | |
| self.self_attn.build(None) | |
| if getattr(self, "self_attn_layer_norm", None) is not None: | |
| with tf.name_scope(self.self_attn_layer_norm.name): | |
| self.self_attn_layer_norm.build([None, None, self.embed_dim]) | |
| if getattr(self, "fc1", None) is not None: | |
| with tf.name_scope(self.fc1.name): | |
| self.fc1.build([None, None, self.embed_dim]) | |
| if getattr(self, "fc2", None) is not None: | |
| with tf.name_scope(self.fc2.name): | |
| self.fc2.build([None, None, self.config.encoder_ffn_dim]) | |
| if getattr(self, "final_layer_norm", None) is not None: | |
| with tf.name_scope(self.final_layer_norm.name): | |
| self.final_layer_norm.build([None, None, self.embed_dim]) | |
| # Copied from transformers.models.speech_to_text.modeling_tf_speech_to_text.TFSpeech2TextDecoderLayer with Speech2Text->Whisper | |
| class TFWhisperDecoderLayer(keras.layers.Layer): | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(**kwargs) | |
| self.embed_dim = config.d_model | |
| self.self_attn = TFWhisperAttention( | |
| embed_dim=self.embed_dim, | |
| num_heads=config.decoder_attention_heads, | |
| dropout=config.attention_dropout, | |
| name="self_attn", | |
| is_decoder=True, | |
| ) | |
| self.dropout = keras.layers.Dropout(config.dropout) | |
| self.activation_fn = get_tf_activation(config.activation_function) | |
| self.activation_dropout = keras.layers.Dropout(config.activation_dropout) | |
| self.self_attn_layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="self_attn_layer_norm") | |
| self.encoder_attn = TFWhisperAttention( | |
| self.embed_dim, | |
| config.decoder_attention_heads, | |
| dropout=config.attention_dropout, | |
| name="encoder_attn", | |
| is_decoder=True, | |
| ) | |
| self.encoder_attn_layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="encoder_attn_layer_norm") | |
| self.fc1 = keras.layers.Dense(config.decoder_ffn_dim, name="fc1") | |
| self.fc2 = keras.layers.Dense(self.embed_dim, name="fc2") | |
| self.final_layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="final_layer_norm") | |
| self.config = config | |
| def call( | |
| self, | |
| hidden_states, | |
| attention_mask: tf.Tensor | None = None, | |
| encoder_hidden_states: tf.Tensor | None = None, | |
| encoder_attention_mask: tf.Tensor | None = None, | |
| layer_head_mask: tf.Tensor | None = None, | |
| cross_attn_layer_head_mask: tf.Tensor | None = None, | |
| past_key_value: Tuple[tf.Tensor] | None = None, | |
| training=False, | |
| ) -> Tuple[tf.Tensor, tf.Tensor, Tuple[Tuple[tf.Tensor]]]: | |
| """ | |
| Args: | |
| hidden_states (`tf.Tensor`): input to the layer of shape `(batch, seq_len, embed_dim)` | |
| attention_mask (`tf.Tensor`): attention mask of size | |
| `(batch, 1, tgt_len, src_len)` where padding elements are indicated by very large negative values. | |
| encoder_hidden_states (`tf.Tensor`): | |
| cross attention input to the layer of shape `(batch, seq_len, embed_dim)` | |
| encoder_attention_mask (`tf.Tensor`): encoder attention mask of size | |
| `(batch, 1, tgt_len, src_len)` where padding elements are indicated by very large negative values. | |
| layer_head_mask (`tf.Tensor`): mask for attention heads in a given layer of size | |
| `(decoder_attention_heads,)` | |
| cross_attn_layer_head_mask (`tf.Tensor`): mask for heads of the cross-attention module. | |
| `(decoder_attention_heads,)` | |
| past_key_value (`Tuple(tf.Tensor)`): cached past key and value projection states | |
| """ | |
| residual = hidden_states | |
| hidden_states = self.self_attn_layer_norm(hidden_states) | |
| # Self Attention | |
| # decoder uni-directional self-attention cached key/values tuple is at positions 1,2 | |
| self_attn_past_key_value = past_key_value[:2] if past_key_value is not None else None | |
| # add present self-attn cache to positions 1,2 of present_key_value tuple | |
| hidden_states, self_attn_weights, present_key_value = self.self_attn( | |
| hidden_states=hidden_states, | |
| past_key_value=self_attn_past_key_value, | |
| attention_mask=attention_mask, | |
| layer_head_mask=layer_head_mask, | |
| training=training, | |
| ) | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| hidden_states = residual + hidden_states | |
| # Cross-Attention Block | |
| cross_attn_present_key_value = None | |
| cross_attn_weights = None | |
| if encoder_hidden_states is not None: | |
| residual = hidden_states | |
| hidden_states = self.encoder_attn_layer_norm(hidden_states) | |
| # cross_attn cached key/values tuple is at positions 3,4 of present_key_value tuple | |
| cross_attn_past_key_value = past_key_value[-2:] if past_key_value is not None else None | |
| hidden_states, cross_attn_weights, cross_attn_present_key_value = self.encoder_attn( | |
| hidden_states=hidden_states, | |
| key_value_states=encoder_hidden_states, | |
| attention_mask=encoder_attention_mask, | |
| layer_head_mask=cross_attn_layer_head_mask, | |
| past_key_value=cross_attn_past_key_value, | |
| training=training, | |
| ) | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| hidden_states = residual + hidden_states | |
| # add cross-attn to positions 3,4 of present_key_value tuple | |
| present_key_value = present_key_value + cross_attn_present_key_value | |
| # Fully Connected | |
| residual = hidden_states | |
| hidden_states = self.final_layer_norm(hidden_states) | |
| hidden_states = self.activation_fn(self.fc1(hidden_states)) | |
| hidden_states = self.activation_dropout(hidden_states, training=training) | |
| hidden_states = self.fc2(hidden_states) | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| hidden_states = residual + hidden_states | |
| return ( | |
| hidden_states, | |
| self_attn_weights, | |
| cross_attn_weights, | |
| present_key_value, | |
| ) | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "self_attn", None) is not None: | |
| with tf.name_scope(self.self_attn.name): | |
| self.self_attn.build(None) | |
| if getattr(self, "self_attn_layer_norm", None) is not None: | |
| with tf.name_scope(self.self_attn_layer_norm.name): | |
| self.self_attn_layer_norm.build([None, None, self.embed_dim]) | |
| if getattr(self, "encoder_attn", None) is not None: | |
| with tf.name_scope(self.encoder_attn.name): | |
| self.encoder_attn.build(None) | |
| if getattr(self, "encoder_attn_layer_norm", None) is not None: | |
| with tf.name_scope(self.encoder_attn_layer_norm.name): | |
| self.encoder_attn_layer_norm.build([None, None, self.embed_dim]) | |
| if getattr(self, "fc1", None) is not None: | |
| with tf.name_scope(self.fc1.name): | |
| self.fc1.build([None, None, self.embed_dim]) | |
| if getattr(self, "fc2", None) is not None: | |
| with tf.name_scope(self.fc2.name): | |
| self.fc2.build([None, None, self.config.decoder_ffn_dim]) | |
| if getattr(self, "final_layer_norm", None) is not None: | |
| with tf.name_scope(self.final_layer_norm.name): | |
| self.final_layer_norm.build([None, None, self.embed_dim]) | |
| class TFWhisperPreTrainedModel(TFPreTrainedModel): | |
| config_class = WhisperConfig | |
| base_model_prefix = "model" | |
| main_input_name = "input_features" | |
| def _get_feat_extract_output_lengths(self, input_lengths: tf.Tensor) -> int: | |
| """ | |
| Computes the output length of the convolutional layers | |
| """ | |
| input_lengths = (input_lengths - 1) // 2 + 1 | |
| return input_lengths | |
| def dummy_inputs(self) -> Dict[str, tf.Tensor]: | |
| """ | |
| Dummy inputs to build the network. | |
| Returns: | |
| `Dict[str, tf.Tensor]`: The dummy inputs. | |
| """ | |
| return { | |
| self.main_input_name: tf.random.uniform( | |
| [1, self.config.num_mel_bins, self.config.max_source_positions * 2 - 1], dtype=tf.float32 | |
| ), | |
| "decoder_input_ids": tf.constant([[1, 3]], dtype=tf.int32), | |
| } | |
| def input_signature(self): | |
| return { | |
| "input_features": tf.TensorSpec((None, self.config.num_mel_bins, None), tf.float32, name="input_features"), | |
| "decoder_input_ids": tf.TensorSpec((None, None), tf.int32, name="decoder_input_ids"), | |
| "decoder_attention_mask": tf.TensorSpec((None, None), tf.int32, name="decoder_attention_mask"), | |
| } | |
| WHISPER_START_DOCSTRING = r""" | |
| This model inherits from [`TFPreTrainedModel`]. Check the superclass documentation for the generic methods the | |
| library implements for all its model (such as downloading or saving, resizing the input embeddings, pruning heads | |
| etc.) | |
| This model is also a [keras.Model](https://www.tensorflow.org/api_docs/python/tf/keras/Model) subclass. Use it | |
| as a regular TF 2.0 Keras Model and refer to the TF 2.0 documentation for all matter related to general usage and | |
| behavior. | |
| Parameters: | |
| config ([`WhisperConfig`]): | |
| Model configuration class with all the parameters of the model. Initializing with a config file does not | |
| load the weights associated with the model, only the configuration. Check out the | |
| [`~TFPreTrainedModel.from_pretrained`] method to load the model weights. | |
| """ | |
| WHISPER_INPUTS_DOCSTRING = r""" | |
| Args: | |
| input_features (`tf.Tensor` of shape `(batch_size, feature_size, sequence_length)`): | |
| Float values of fbank features extracted from the raw speech waveform. Raw speech waveform can be obtained | |
| by loading a `.flac` or `.wav` audio file into an array of type `List[float]` or a `numpy.ndarray`, *e.g.* | |
| via the soundfile library (`pip install soundfile`). To prepare the array into `input_features`, the | |
| [`AutoFeatureExtractor`] should be used for extracting the fbank features, padding and conversion into a | |
| tensor of type `tf.Tensor`. See [`~WhisperFeatureExtractor.__call__`] | |
| decoder_input_ids (`tf.Tensor` of shape `(batch_size, target_sequence_length)`, *optional*): | |
| Indices of decoder input sequence tokens in the vocabulary. | |
| Indices can be obtained using [`SpeechToTextTokenizer`]. See [`PreTrainedTokenizer.encode`] and | |
| [`PreTrainedTokenizer.__call__`] for details. | |
| [What are decoder input IDs?](../glossary#decoder-input-ids) | |
| SpeechToText uses the `eos_token_id` as the starting token for `decoder_input_ids` generation. If | |
| `past_key_values` is used, optionally only the last `decoder_input_ids` have to be input (see | |
| `past_key_values`). | |
| decoder_attention_mask (`tf.Tensor` of shape `(batch_size, target_sequence_length)`, *optional*): | |
| Default behavior: generate a tensor that ignores pad tokens in `decoder_input_ids`. Causal mask will also | |
| be used by default. | |
| If you want to change padding behavior, you should read | |
| [`modeling_whisper._prepare_decoder_attention_mask`] and modify to your needs. See diagram 1 in [the | |
| paper](https://arxiv.org/abs/1910.13461) for more information on the default strategy. | |
| head_mask (`tf.Tensor` of shape `(encoder_layers, encoder_attention_heads)`, *optional*): | |
| Mask to nullify selected heads of the attention modules in the encoder. Mask values selected in `[0, 1]`: | |
| - 1 indicates the head is **not masked**, | |
| - 0 indicates the head is **masked**. | |
| decoder_head_mask (`tf.Tensor` of shape `(decoder_layers, decoder_attention_heads)`, *optional*): | |
| Mask to nullify selected heads of the attention modules in the decoder. Mask values selected in `[0, 1]`: | |
| - 1 indicates the head is **not masked**, | |
| - 0 indicates the head is **masked**. | |
| cross_attn_head_mask (`tf.Tensor` of shape `(decoder_layers, decoder_attention_heads)`, *optional*): | |
| Mask to nullify selected heads of the cross-attention modules. Mask values selected in `[0, 1]`: | |
| - 1 indicates the head is **not masked**, | |
| - 0 indicates the head is **masked**. | |
| encoder_outputs (`tuple(tuple(tf.Tensor)`, *optional*): | |
| Tuple consists of (`last_hidden_state`, *optional*: `hidden_states`, *optional*: `attentions`) | |
| `last_hidden_state` of shape `(batch_size, sequence_length, hidden_size)`, *optional*) is a sequence of | |
| hidden-states at the output of the last layer of the encoder. Used in the cross-attention of the decoder. | |
| past_key_values (`tuple(tuple(tf.Tensor))`, *optional*, returned when `use_cache=True` is passed or when `config.use_cache=True`): | |
| Tuple of `tuple(tf.Tensor)` of length `config.n_layers`, with each tuple having 2 tensors of shape | |
| `(batch_size, num_heads, sequence_length, embed_size_per_head)`) and 2 additional tensors of shape | |
| `(batch_size, num_heads, encoder_sequence_length, embed_size_per_head)`. | |
| Contains pre-computed hidden-states (key and values in the self-attention blocks and in the cross-attention | |
| blocks) that can be used (see `past_key_values` input) to speed up sequential decoding. | |
| If `past_key_values` are used, the user can optionally input only the last `decoder_input_ids` (those that | |
| don't have their past key value states given to this model) of shape `(batch_size, 1)` instead of all | |
| `decoder_input_ids` of shape `(batch_size, sequence_length)`. | |
| decoder_inputs_embeds (`tf.Tensor` of shape `(batch_size, target_sequence_length, hidden_size)`, *optional*): | |
| Optionally, instead of passing `decoder_input_ids` you can choose to directly pass an embedded | |
| representation. If `past_key_values` is used, optionally only the last `decoder_inputs_embeds` have to be | |
| input (see `past_key_values`). This is useful if you want more control over how to convert | |
| `decoder_input_ids` indices into associated vectors than the model's internal embedding lookup matrix. | |
| use_cache (`bool`, *optional*): | |
| If set to `True`, `past_key_values` key value states are returned and can be used to speed up decoding (see | |
| `past_key_values`). | |
| output_attentions (`bool`, *optional*): | |
| Whether or not to return the attentions tensors of all attention layers. See `attentions` under returned | |
| tensors for more detail. | |
| output_hidden_states (`bool`, *optional*): | |
| Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for | |
| more detail. | |
| return_dict (`bool`, *optional*): | |
| Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
| """ | |
| class TFWhisperEncoder(keras.layers.Layer): | |
| config_class = WhisperConfig | |
| """ | |
| Transformer encoder consisting of *config.encoder_layers* self attention layers. Each layer is a | |
| [`TFWhisperEncoderLayer`]. | |
| Args: | |
| config: WhisperConfig | |
| embed_tokens (TFWhisperEmbedding): output embedding | |
| """ | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(**kwargs) | |
| self.config = config | |
| self.layerdrop = config.encoder_layerdrop | |
| self.embed_dim = config.d_model | |
| self.num_mel_bins = config.num_mel_bins | |
| self.padding_idx = config.pad_token_id | |
| self.max_source_positions = config.max_source_positions | |
| self.embed_scale = math.sqrt(self.embed_dim) if config.scale_embedding else 1.0 | |
| # Padding is added in call() to match the PyTorch implementation | |
| self.conv1 = keras.layers.Conv1D(self.embed_dim, kernel_size=3, strides=1, padding="valid", name="conv1") | |
| self.conv2 = keras.layers.Conv1D(self.embed_dim, kernel_size=3, strides=2, padding="valid", name="conv2") | |
| self.embed_positions = TFWhisperPositionalEmbedding( | |
| num_positions=self.max_source_positions, | |
| embedding_dim=self.embed_dim, | |
| embedding_initializer=sinusoidal_embedding_init, | |
| name="embed_positions", | |
| ) | |
| self.embed_positions.trainable = False | |
| self.encoder_layers = [TFWhisperEncoderLayer(config, name=f"layers.{i}") for i in range(config.encoder_layers)] | |
| self.layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="layer_norm") | |
| self.dropout = keras.layers.Dropout(config.dropout) | |
| def call( | |
| self, | |
| input_features=None, | |
| head_mask=None, | |
| output_attentions=None, | |
| output_hidden_states=None, | |
| return_dict=None, | |
| training=False, | |
| ): | |
| r""" | |
| Args: | |
| input_features (`tf.Tensor` of shape `(batch_size, feature_size, sequence_length)`): | |
| Float values of fbank features extracted from the raw speech waveform. Raw speech waveform can be | |
| obtained by loading a `.flac` or `.wav` audio file into an array of type `List[float]` or a | |
| `numpy.ndarray`, *e.g.* via the soundfile library (`pip install soundfile`). To prepare the array into | |
| `input_features`, the [`AutoFeatureExtractor`] should be used for extracting the fbank features, | |
| padding and conversion into a tensor of type `tf.Tensor`. See [`~WhisperFeatureExtractor.__call__`] | |
| head_mask (`tf.Tensor` of shape `(encoder_layers, encoder_attention_heads)`, *optional*): | |
| Mask to nullify selected heads of the attention modules. Mask values selected in `[0, 1]`: | |
| - 1 indicates the head is **not masked**, | |
| - 0 indicates the head is **masked**. | |
| output_attentions (`bool`, *optional*): | |
| Whether or not to return the attentions tensors of all attention layers. See `attentions` under | |
| returned tensors for more detail. | |
| output_hidden_states (`bool`, *optional*): | |
| Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors | |
| for more detail. | |
| return_dict (`bool`, *optional*): | |
| Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
| """ | |
| output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
| output_hidden_states = ( | |
| output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
| ) | |
| return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
| # TF 2.0 layers can't use channels first format when running on CPU. | |
| input_features = tf.transpose(input_features, perm=(0, 2, 1)) | |
| input_features = tf.pad(input_features, [[0, 0], [1, 1], [0, 0]]) | |
| inputs_embeds = keras.activations.gelu(self.conv1(input_features)) | |
| inputs_embeds = tf.pad(inputs_embeds, [[0, 0], [1, 1], [0, 0]]) | |
| inputs_embeds = keras.activations.gelu(self.conv2(inputs_embeds)) | |
| inputs_embeds = tf.transpose(inputs_embeds, perm=(0, 1, 2)) | |
| embed_pos = self.embed_positions(input_ids=tf.zeros((1, self.max_source_positions), dtype=tf.int32)) | |
| hidden_states = inputs_embeds + embed_pos | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| encoder_states = () if output_hidden_states else None | |
| all_attentions = () if output_attentions else None | |
| # check if head_mask has a correct number of layers specified if desired | |
| if head_mask is not None: | |
| tf.debugging.assert_equal( | |
| shape_list(head_mask)[0], | |
| len(self.encoder_layers), | |
| message=( | |
| f"The head_mask should be specified for {len(self.encoder_layers)} layers, but it is for" | |
| f" {shape_list(head_mask)[0]}." | |
| ), | |
| ) | |
| for idx, encoder_layer in enumerate(self.encoder_layers): | |
| if output_hidden_states: | |
| encoder_states = encoder_states + (hidden_states,) | |
| # add LayerDrop (see https://arxiv.org/abs/1909.11556 for description) | |
| dropout_probability = random.uniform(0, 1) | |
| if training and (dropout_probability < self.layerdrop): # skip the layer | |
| continue | |
| hidden_states, attn = encoder_layer( | |
| hidden_states, | |
| None, | |
| layer_head_mask=(head_mask[idx] if head_mask is not None else None), | |
| training=training, | |
| ) | |
| if output_attentions: | |
| all_attentions += (attn,) | |
| hidden_states = self.layer_norm(hidden_states) | |
| if output_hidden_states: | |
| encoder_states = encoder_states + (hidden_states,) | |
| if not return_dict: | |
| return tuple(v for v in [hidden_states, encoder_states, all_attentions] if v is not None) | |
| return TFBaseModelOutput( | |
| last_hidden_state=hidden_states, hidden_states=encoder_states, attentions=all_attentions | |
| ) | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "conv1", None) is not None: | |
| with tf.name_scope(self.conv1.name): | |
| self.conv1.build([None, None, self.num_mel_bins]) | |
| if getattr(self, "conv2", None) is not None: | |
| with tf.name_scope(self.conv2.name): | |
| self.conv2.build([None, None, self.embed_dim]) | |
| if getattr(self, "embed_positions", None) is not None: | |
| with tf.name_scope(self.embed_positions.name): | |
| self.embed_positions.build(None) | |
| if getattr(self, "layer_norm", None) is not None: | |
| with tf.name_scope(self.layer_norm.name): | |
| self.layer_norm.build([None, None, self.config.d_model]) | |
| if getattr(self, "encoder_layers", None) is not None: | |
| for layer in self.encoder_layers: | |
| with tf.name_scope(layer.name): | |
| layer.build(None) | |
| class TFWhisperDecoder(keras.layers.Layer): | |
| config_class = WhisperConfig | |
| """ | |
| Transformer decoder consisting of *config.decoder_layers* layers. Each layer is a [`TFWhisperDecoderLayer`] | |
| Args: | |
| config: WhisperConfig | |
| """ | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(**kwargs) | |
| self.config = config | |
| self.dropout = keras.layers.Dropout(config.dropout) | |
| self.layerdrop = config.decoder_layerdrop | |
| self.padding_idx = config.pad_token_id | |
| self.max_target_positions = config.max_target_positions | |
| self.max_source_positions = config.max_source_positions | |
| self.embed_scale = math.sqrt(config.d_model) if config.scale_embedding else 1.0 | |
| self.embed_tokens = keras.layers.Embedding( | |
| input_dim=config.vocab_size, | |
| output_dim=config.d_model, | |
| embeddings_initializer=keras.initializers.TruncatedNormal(stddev=self.config.init_std), | |
| name="embed_tokens", | |
| ) | |
| self.embed_positions = TFWhisperPositionalEmbedding( | |
| self.max_target_positions, config.d_model, name="embed_positions" | |
| ) | |
| self.decoder_layers = [TFWhisperDecoderLayer(config, name=f"layers.{i}") for i in range(config.decoder_layers)] | |
| self.layer_norm = keras.layers.LayerNormalization(epsilon=1e-5, name="layer_norm") | |
| def get_input_embeddings(self): | |
| return self.embed_tokens | |
| def set_input_embeddings(self, value): | |
| self.embed_tokens = value | |
| def _prepare_decoder_attention_mask(self, attention_mask, input_shape, past_key_values_length): | |
| # create causal mask | |
| # [bsz, seq_len] -> [bsz, 1, tgt_seq_len, src_seq_len] | |
| batch_size, seq_len = input_shape[0], input_shape[1] | |
| combined_attention_mask = tf.cond( | |
| tf.math.greater(seq_len, 1), | |
| lambda: _make_causal_mask(input_shape, past_key_values_length=past_key_values_length), | |
| lambda: _expand_mask(tf.ones((batch_size, seq_len + past_key_values_length)), tgt_len=seq_len), | |
| ) | |
| if attention_mask is not None: | |
| # [bsz, seq_len] -> [bsz, 1, tgt_seq_len, src_seq_len] | |
| expanded_attn_mask = _expand_mask(attention_mask, tgt_len=input_shape[-1]) | |
| combined_attention_mask = ( | |
| expanded_attn_mask if combined_attention_mask is None else expanded_attn_mask + combined_attention_mask | |
| ) | |
| return combined_attention_mask | |
| def call( | |
| self, | |
| input_ids=None, | |
| attention_mask=None, | |
| position_ids=None, | |
| encoder_hidden_states=None, | |
| head_mask=None, | |
| cross_attn_head_mask=None, | |
| past_key_values=None, | |
| inputs_embeds=None, | |
| use_cache=None, | |
| output_attentions=None, | |
| output_hidden_states=None, | |
| return_dict=None, | |
| training=False, | |
| ): | |
| r""" | |
| Args: | |
| input_ids (`tf.Tensor` of shape `(batch_size, sequence_length)`): | |
| Indices of input sequence tokens in the vocabulary. Padding will be ignored by default should you | |
| provide it. | |
| Indices can be obtained using [`WhisperTokenizer`]. See [`PreTrainedTokenizer.encode`] and | |
| [`PreTrainedTokenizer.__call__`] for details. | |
| [What are input IDs?](../glossary#input-ids) | |
| attention_mask (`tf.Tensor` of shape `(batch_size, sequence_length)`, *optional*): | |
| Mask to avoid performing attention on padding token indices. Mask values selected in `[0, 1]`: | |
| - 1 for tokens that are **not masked**, | |
| - 0 for tokens that are **masked**. | |
| [What are attention masks?](../glossary#attention-mask) | |
| position_ids (`tf.Tensor` of shape `(batch_size, sequence_length)`, *optional*): | |
| Indices of positions of each decoder input sequence tokens in the position embeddings. Selected in the | |
| range `[0, config.max_position_embeddings - 1]`. | |
| encoder_hidden_states (`tf.Tensor` of shape `(batch_size, encoder_sequence_length, hidden_size)`, *optional*): | |
| Sequence of hidden-states at the output of the last layer of the encoder. Used in the cross-attention | |
| of the decoder. | |
| head_mask (`tf.Tensor` of shape `(decoder_layers, decoder_attention_heads)`, *optional*): | |
| Mask to nullify selected heads of the attention modules. Mask values selected in `[0, 1]`: | |
| - 1 indicates the head is **not masked**, | |
| - 0 indicates the head is **masked**. | |
| cross_attn_head_mask (`tf.Tensor` of shape `(decoder_layers, decoder_attention_heads)`, *optional*): | |
| Mask to nullify selected heads of the attention modules in encoder to avoid performing cross-attention | |
| on hidden heads. Mask values selected in `[0, 1]`: | |
| - 1 indicates the head is **not masked**, | |
| - 0 indicates the head is **masked**. | |
| past_key_values (`tuple(tuple(tf.Tensor))`, *optional*, returned when `use_cache=True` is passed or when `config.use_cache=True`): | |
| Tuple of `tuple(tf.Tensor)` of length `config.n_layers`, with each tuple having 2 tensors of shape | |
| `(batch_size, num_heads, sequence_length, embed_size_per_head)`) and 2 additional tensors of shape | |
| `(batch_size, num_heads, encoder_sequence_length, embed_size_per_head)`. | |
| Contains pre-computed hidden-states (key and values in the self-attention blocks and in the | |
| cross-attention blocks) that can be used (see `past_key_values` input) to speed up sequential decoding. | |
| If `past_key_values` are used, the user can optionally input only the last `decoder_input_ids` (those | |
| that don't have their past key value states given to this model) of shape `(batch_size, 1)` instead of | |
| all `decoder_input_ids` of shape `(batch_size, sequence_length)`. | |
| inputs_embeds (`tf.Tensor` of shape `(batch_size, sequence_length, hidden_size)`, *optional*): | |
| Optionally, instead of passing `input_ids` you can choose to directly pass an embedded representation. | |
| This is useful if you want more control over how to convert `input_ids` indices into associated vectors | |
| than the model's internal embedding lookup matrix. | |
| output_attentions (`bool`, *optional*): | |
| Whether or not to return the attentions tensors of all attention layers. See `attentions` under | |
| returned tensors for more detail. | |
| output_hidden_states (`bool`, *optional*): | |
| Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors | |
| for more detail. | |
| return_dict (`bool`, *optional*): | |
| Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
| """ | |
| output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
| output_hidden_states = ( | |
| output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
| ) | |
| use_cache = use_cache if use_cache is not None else self.config.use_cache | |
| return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
| # retrieve input_ids and inputs_embeds | |
| if input_ids is not None and inputs_embeds is not None: | |
| raise ValueError("You cannot specify both decoder_input_ids and decoder_inputs_embeds at the same time") | |
| elif input_ids is not None: | |
| input_shape = tf.shape(input_ids) | |
| input_ids = tf.reshape(input_ids, (-1, input_shape[-1])) | |
| elif inputs_embeds is not None: | |
| input_shape = tf.shape(inputs_embeds)[:-1] | |
| else: | |
| raise ValueError("You have to specify either decoder_input_ids or decoder_inputs_embeds") | |
| # past_key_values_length | |
| past_key_values_length = tf.shape(past_key_values[0][0])[2] if past_key_values is not None else 0 | |
| if inputs_embeds is None: | |
| check_embeddings_within_bounds(input_ids, self.embed_tokens.input_dim) | |
| inputs_embeds = self.embed_tokens(input_ids) | |
| attention_mask = self._prepare_decoder_attention_mask(attention_mask, input_shape, past_key_values_length) | |
| # embed positions | |
| filled_past_positions = past_key_values_length if position_ids is None else position_ids[0, -1] | |
| positions = self.embed_positions(input_ids, past_key_values_length=filled_past_positions) | |
| hidden_states = inputs_embeds + positions | |
| hidden_states = self.dropout(hidden_states, training=training) | |
| # decoder layers | |
| all_hidden_states = () if output_hidden_states else None | |
| all_self_attns = () if output_attentions else None | |
| all_cross_attentions = () if (output_attentions and encoder_hidden_states is not None) else None | |
| next_decoder_cache = () if use_cache else None | |
| # check if head_mask/cross_attn_head_mask has a correct number of layers specified if desired | |
| for attn_mask_name, attn_mask in [("head_mask", head_mask), ("cross_attn_head_mask", cross_attn_head_mask)]: | |
| if attn_mask is not None: | |
| tf.debugging.assert_equal( | |
| shape_list(attn_mask)[0], | |
| len(self.decoder_layers), | |
| message=( | |
| f"The {attn_mask_name} should be specified for {len(self.decoder_layers)} layers, but it is" | |
| f" for {shape_list(attn_mask)[0]}." | |
| ), | |
| ) | |
| for idx, decoder_layer in enumerate(self.decoder_layers): | |
| # add LayerDrop (see https://arxiv.org/abs/1909.11556 for description) | |
| if output_hidden_states: | |
| all_hidden_states += (hidden_states,) | |
| dropout_probability = random.uniform(0, 1) | |
| if training and (dropout_probability < self.layerdrop): | |
| continue | |
| past_key_value = past_key_values[idx] if past_key_values is not None else None | |
| layer_outputs = decoder_layer( | |
| hidden_states, | |
| attention_mask=attention_mask, | |
| encoder_hidden_states=encoder_hidden_states, | |
| layer_head_mask=(head_mask[idx] if head_mask is not None else None), | |
| cross_attn_layer_head_mask=(cross_attn_head_mask[idx] if cross_attn_head_mask is not None else None), | |
| past_key_value=past_key_value, | |
| training=training, | |
| ) | |
| hidden_states = layer_outputs[0] | |
| if use_cache: | |
| next_decoder_cache += (layer_outputs[3],) | |
| if output_attentions: | |
| all_self_attns += (layer_outputs[1],) | |
| if encoder_hidden_states is not None: | |
| all_cross_attentions += (layer_outputs[2],) | |
| hidden_states = self.layer_norm(hidden_states) | |
| # add hidden states from the last decoder layer | |
| if output_hidden_states: | |
| all_hidden_states += (hidden_states,) | |
| next_cache = next_decoder_cache if use_cache else None | |
| if not return_dict: | |
| return tuple( | |
| v | |
| for v in [hidden_states, next_cache, all_hidden_states, all_self_attns, all_cross_attentions] | |
| if v is not None | |
| ) | |
| return TFBaseModelOutputWithPastAndCrossAttentions( | |
| last_hidden_state=hidden_states, | |
| past_key_values=next_cache, | |
| hidden_states=all_hidden_states, | |
| attentions=all_self_attns, | |
| cross_attentions=all_cross_attentions, | |
| ) | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "embed_tokens", None) is not None: | |
| with tf.name_scope(self.embed_tokens.name): | |
| self.embed_tokens.build(None) | |
| if getattr(self, "embed_positions", None) is not None: | |
| with tf.name_scope(self.embed_positions.name): | |
| self.embed_positions.build(None) | |
| if getattr(self, "layer_norm", None) is not None: | |
| with tf.name_scope(self.layer_norm.name): | |
| self.layer_norm.build([None, None, self.config.d_model]) | |
| if getattr(self, "decoder_layers", None) is not None: | |
| for layer in self.decoder_layers: | |
| with tf.name_scope(layer.name): | |
| layer.build(None) | |
| class TFWhisperMainLayer(keras.layers.Layer): | |
| config_class = WhisperConfig | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(**kwargs) | |
| self.config = config | |
| self.encoder = TFWhisperEncoder(config, name="encoder") | |
| self.decoder = TFWhisperDecoder(config, name="decoder") | |
| def get_input_embeddings(self): | |
| return self.decoder.embed_tokens | |
| def set_input_embeddings(self, value): | |
| self.decoder.embed_tokens = value | |
| def get_encoder(self): | |
| return self.encoder | |
| def get_decoder(self): | |
| return self.decoder | |
| def call( | |
| self, | |
| input_features=None, | |
| decoder_input_ids=None, | |
| decoder_attention_mask=None, | |
| decoder_position_ids=None, | |
| head_mask=None, | |
| decoder_head_mask=None, | |
| cross_attn_head_mask=None, | |
| encoder_outputs=None, | |
| past_key_values=None, | |
| decoder_inputs_embeds=None, | |
| use_cache=None, | |
| output_attentions=None, | |
| output_hidden_states=None, | |
| return_dict=None, | |
| training=False, | |
| ): | |
| r""" | |
| Returns: | |
| Example: | |
| ```python | |
| >>> import tensorflow as tf | |
| >>> from transformers import TFWhisperModel, AutoFeatureExtractor | |
| >>> from datasets import load_dataset | |
| >>> model = TFWhisperModel.from_pretrained("openai/whisper-base") | |
| >>> feature_extractor = AutoFeatureExtractor.from_pretrained("openai/whisper-base") | |
| >>> ds = load_dataset("hf-internal-testing/librispeech_asr_dummy", "clean", split="validation") | |
| >>> inputs = feature_extractor(ds[0]["audio"]["array"], return_tensors="tf") | |
| >>> input_features = inputs.input_features | |
| >>> decoder_input_ids = tf.convert_to_tensor([[1, 1]]) * model.config.decoder_start_token_id | |
| >>> last_hidden_state = model(input_features, decoder_input_ids=decoder_input_ids).last_hidden_state | |
| >>> list(last_hidden_state.shape) | |
| [1, 2, 512] | |
| ```""" | |
| output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
| output_hidden_states = ( | |
| output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
| ) | |
| use_cache = use_cache if use_cache is not None else self.config.use_cache | |
| return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
| if encoder_outputs is None: | |
| encoder_outputs = self.encoder( | |
| input_features, | |
| head_mask=head_mask, | |
| output_attentions=output_attentions, | |
| output_hidden_states=output_hidden_states, | |
| return_dict=return_dict, | |
| training=training, | |
| ) | |
| # If the user passed a tuple for encoder_outputs, we wrap it in a TFBaseModelOutput when return_dict=True | |
| elif return_dict and not isinstance(encoder_outputs, TFBaseModelOutput): | |
| encoder_outputs = TFBaseModelOutput( | |
| last_hidden_state=encoder_outputs[0], | |
| hidden_states=encoder_outputs[1] if len(encoder_outputs) > 1 else None, | |
| attentions=encoder_outputs[2] if len(encoder_outputs) > 2 else None, | |
| ) | |
| # decoder outputs consists of (dec_features, past_key_value, dec_hidden, dec_attn) | |
| decoder_outputs = self.decoder( | |
| input_ids=decoder_input_ids, | |
| attention_mask=decoder_attention_mask, | |
| position_ids=decoder_position_ids, | |
| encoder_hidden_states=encoder_outputs[0], | |
| head_mask=decoder_head_mask, | |
| cross_attn_head_mask=cross_attn_head_mask, | |
| past_key_values=past_key_values, | |
| inputs_embeds=decoder_inputs_embeds, | |
| use_cache=use_cache, | |
| output_attentions=output_attentions, | |
| output_hidden_states=output_hidden_states, | |
| return_dict=return_dict, | |
| training=training, | |
| ) | |
| if not return_dict: | |
| return decoder_outputs + encoder_outputs | |
| return TFSeq2SeqModelOutput( | |
| last_hidden_state=decoder_outputs.last_hidden_state, | |
| past_key_values=decoder_outputs.past_key_values, | |
| decoder_hidden_states=decoder_outputs.hidden_states, | |
| decoder_attentions=decoder_outputs.attentions, | |
| cross_attentions=decoder_outputs.cross_attentions, | |
| encoder_last_hidden_state=encoder_outputs.last_hidden_state, | |
| encoder_hidden_states=encoder_outputs.hidden_states, | |
| encoder_attentions=encoder_outputs.attentions, | |
| ) | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "encoder", None) is not None: | |
| with tf.name_scope(self.encoder.name): | |
| self.encoder.build(None) | |
| if getattr(self, "decoder", None) is not None: | |
| with tf.name_scope(self.decoder.name): | |
| self.decoder.build(None) | |
| class TFWhisperModel(TFWhisperPreTrainedModel): | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(config, **kwargs) | |
| self.model = TFWhisperMainLayer(config, name="model") | |
| def get_input_embeddings(self): | |
| return self.model.decoder.embed_tokens | |
| def set_input_embeddings(self, value): | |
| self.model.decoder.embed_tokens = value | |
| def get_encoder(self): | |
| return self.model.encoder | |
| def get_decoder(self): | |
| return self.model.decoder | |
| def decoder(self): | |
| return self.model.decoder | |
| def encoder(self): | |
| return self.model.encoder | |
| def call( | |
| self, | |
| input_features: TFModelInputType | None = None, | |
| decoder_input_ids: np.ndarray | tf.Tensor | None = None, | |
| decoder_attention_mask: np.ndarray | tf.Tensor | None = None, | |
| decoder_position_ids: np.ndarray | tf.Tensor | None = None, | |
| head_mask: np.ndarray | tf.Tensor | None = None, | |
| decoder_head_mask: np.ndarray | tf.Tensor | None = None, | |
| cross_attn_head_mask: np.ndarray | tf.Tensor | None = None, | |
| encoder_outputs: Optional[Tuple[Tuple[Union[np.ndarray, tf.Tensor]]]] = None, | |
| past_key_values: Optional[Tuple[Tuple[Union[np.ndarray, tf.Tensor]]]] = None, | |
| decoder_inputs_embeds: Optional[Tuple[Union[np.ndarray, tf.Tensor]]] = None, | |
| use_cache: Optional[bool] = None, | |
| output_attentions: Optional[bool] = None, | |
| output_hidden_states: Optional[bool] = None, | |
| return_dict: Optional[bool] = None, | |
| training: bool = False, | |
| ) -> Union[Tuple[tf.Tensor], TFSeq2SeqModelOutput]: | |
| r""" | |
| Returns: | |
| Example: | |
| ```python | |
| >>> import tensorflow as tf | |
| >>> from transformers import TFWhisperModel, AutoFeatureExtractor | |
| >>> from datasets import load_dataset | |
| >>> model = TFWhisperModel.from_pretrained("openai/whisper-base") | |
| >>> feature_extractor = AutoFeatureExtractor.from_pretrained("openai/whisper-base") | |
| >>> ds = load_dataset("hf-internal-testing/librispeech_asr_dummy", "clean", split="validation") | |
| >>> inputs = feature_extractor(ds[0]["audio"]["array"], return_tensors="tf") | |
| >>> input_features = inputs.input_features | |
| >>> decoder_input_ids = tf.convert_to_tensor([[1, 1]]) * model.config.decoder_start_token_id | |
| >>> last_hidden_state = model(input_features, decoder_input_ids=decoder_input_ids).last_hidden_state | |
| >>> list(last_hidden_state.shape) | |
| [1, 2, 512] | |
| ```""" | |
| outputs = self.model( | |
| input_features=input_features, | |
| decoder_input_ids=decoder_input_ids, | |
| decoder_attention_mask=decoder_attention_mask, | |
| decoder_position_ids=decoder_position_ids, | |
| head_mask=head_mask, | |
| decoder_head_mask=decoder_head_mask, | |
| cross_attn_head_mask=cross_attn_head_mask, | |
| encoder_outputs=encoder_outputs, | |
| past_key_values=past_key_values, | |
| decoder_inputs_embeds=decoder_inputs_embeds, | |
| use_cache=use_cache, | |
| output_attentions=output_attentions, | |
| output_hidden_states=output_hidden_states, | |
| return_dict=return_dict, | |
| training=training, | |
| ) | |
| return outputs | |
| def serving_output(self, output): | |
| pkv = tf.tuple(output.past_key_values)[1] if self.config.use_cache else None | |
| dec_hs = tf.convert_to_tensor(output.decoder_hidden_states) if self.config.output_hidden_states else None | |
| dec_attns = tf.convert_to_tensor(output.decoder_attentions) if self.config.output_attentions else None | |
| cross_attns = tf.convert_to_tensor(output.cross_attentions) if self.config.output_attentions else None | |
| enc_hs = tf.convert_to_tensor(output.encoder_hidden_states) if self.config.output_hidden_states else None | |
| enc_attns = tf.convert_to_tensor(output.encoder_attentions) if self.config.output_attentions else None | |
| return TFSeq2SeqModelOutput( | |
| last_hidden_state=output.last_hidden_state, | |
| past_key_values=pkv, | |
| decoder_hidden_states=dec_hs, | |
| decoder_attentions=dec_attns, | |
| cross_attentions=cross_attns, | |
| encoder_last_hidden_state=output.encoder_last_hidden_state, | |
| encoder_hidden_states=enc_hs, | |
| encoder_attentions=enc_attns, | |
| ) | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "model", None) is not None: | |
| with tf.name_scope(self.model.name): | |
| self.model.build(None) | |
| class TFWhisperForConditionalGeneration(TFWhisperPreTrainedModel, TFCausalLanguageModelingLoss): | |
| base_model_prefix = "model" | |
| _keys_to_ignore_on_load_missing = [ | |
| r"encoder.version", | |
| r"decoder.version", | |
| r"proj_out.weight", | |
| ] | |
| _keys_to_ignore_on_save = [ | |
| r"proj_out.weight", | |
| ] | |
| def __init__(self, config: WhisperConfig, **kwargs): | |
| super().__init__(config, **kwargs) | |
| self.model = TFWhisperMainLayer(config, name="model") | |
| def get_encoder(self): | |
| return self.model.get_encoder() | |
| def get_decoder(self): | |
| return self.model.get_decoder() | |
| def get_output_embeddings(self): | |
| return self.get_input_embeddings() | |
| def set_output_embeddings(self, value): | |
| self.set_input_embeddings(value) | |
| def resize_token_embeddings(self, new_num_tokens: int) -> keras.layers.Embedding: | |
| new_embeddings = super().resize_token_embeddings(new_num_tokens) | |
| return new_embeddings | |
| def call( | |
| self, | |
| input_features: TFModelInputType | None = None, | |
| decoder_input_ids: np.ndarray | tf.Tensor | None = None, | |
| decoder_attention_mask: np.ndarray | tf.Tensor | None = None, | |
| decoder_position_ids: np.ndarray | tf.Tensor | None = None, | |
| head_mask: np.ndarray | tf.Tensor | None = None, | |
| decoder_head_mask: np.ndarray | tf.Tensor | None = None, | |
| cross_attn_head_mask: np.ndarray | tf.Tensor | None = None, | |
| encoder_outputs: Optional[Tuple[Tuple[Union[np.ndarray, tf.Tensor]]]] = None, | |
| past_key_values: Optional[Tuple[Tuple[Union[np.ndarray, tf.Tensor]]]] = None, | |
| decoder_inputs_embeds: Optional[Tuple[Union[np.ndarray, tf.Tensor]]] = None, | |
| labels: np.ndarray | tf.Tensor | None = None, | |
| use_cache: Optional[bool] = None, | |
| output_attentions: Optional[bool] = None, | |
| output_hidden_states: Optional[bool] = None, | |
| return_dict: Optional[bool] = None, | |
| training: bool = False, | |
| ) -> Union[Tuple[tf.Tensor], TFSeq2SeqLMOutput]: | |
| r""" | |
| labels (`tf.Tensor` of shape `(batch_size, sequence_length)`, *optional*): | |
| Labels for computing the language modeling loss. Indices should either be in `[0, ..., config.vocab_size]` | |
| or -100 (see `input_ids` docstring). Tokens with indices set to `-100` are ignored (masked), the loss is | |
| only computed for the tokens with labels in `[0, ..., config.vocab_size]`. | |
| Returns: | |
| Example: | |
| ```python | |
| >>> import tensorflow as tf | |
| >>> from transformers import AutoProcessor, TFWhisperForConditionalGeneration | |
| >>> from datasets import load_dataset | |
| >>> processor = AutoProcessor.from_pretrained("openai/whisper-tiny.en") | |
| >>> model = TFWhisperForConditionalGeneration.from_pretrained("openai/whisper-tiny.en") | |
| >>> ds = load_dataset("hf-internal-testing/librispeech_asr_dummy", "clean", split="validation") | |
| >>> inputs = processor(ds[0]["audio"]["array"], return_tensors="tf") | |
| >>> input_features = inputs.input_features | |
| >>> generated_ids = model.generate(input_features=input_features) | |
| >>> transcription = processor.batch_decode(generated_ids, skip_special_tokens=True)[0] | |
| >>> transcription | |
| ' Mr. Quilter is the apostle of the middle classes, and we are glad to welcome his gospel.' | |
| ```""" | |
| return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
| if labels is not None: | |
| if decoder_input_ids is None and decoder_inputs_embeds is None: | |
| decoder_input_ids = shift_tokens_right( | |
| labels, self.config.pad_token_id, self.config.decoder_start_token_id | |
| ) | |
| outputs = self.model( | |
| input_features, | |
| decoder_input_ids=decoder_input_ids, | |
| encoder_outputs=encoder_outputs, | |
| decoder_attention_mask=decoder_attention_mask, | |
| decoder_position_ids=decoder_position_ids, | |
| head_mask=head_mask, | |
| decoder_head_mask=decoder_head_mask, | |
| cross_attn_head_mask=cross_attn_head_mask, | |
| past_key_values=past_key_values, | |
| decoder_inputs_embeds=decoder_inputs_embeds, | |
| use_cache=use_cache, | |
| output_attentions=output_attentions, | |
| output_hidden_states=output_hidden_states, | |
| return_dict=return_dict, | |
| training=training, | |
| ) | |
| decoder_last_hidden_state = outputs[0] | |
| # Decoder and encoder embeddings are tied | |
| lm_logits = tf.matmul(decoder_last_hidden_state, self.get_output_embeddings().weights, transpose_b=True) | |
| loss = None if labels is None else self.hf_compute_loss(labels, lm_logits) | |
| if not return_dict: | |
| output = (lm_logits,) + outputs[1:] | |
| return ((loss,) + output) if loss is not None else output | |
| return TFSeq2SeqLMOutput( | |
| loss=loss, | |
| logits=lm_logits, | |
| past_key_values=outputs.past_key_values, | |
| decoder_hidden_states=outputs.decoder_hidden_states, | |
| decoder_attentions=outputs.decoder_attentions, | |
| cross_attentions=outputs.cross_attentions, | |
| encoder_last_hidden_state=outputs.encoder_last_hidden_state, | |
| encoder_hidden_states=outputs.encoder_hidden_states, | |
| encoder_attentions=outputs.encoder_attentions, | |
| ) | |
| def generate( | |
| self, | |
| inputs: Optional[tf.Tensor] = None, | |
| generation_config: Optional[GenerationConfig] = None, | |
| logits_processor: Optional[TFLogitsProcessorList] = None, | |
| seed: Optional[List[int]] = None, | |
| return_timestamps: Optional[bool] = None, | |
| task: Optional[str] = None, | |
| language: Optional[str] = None, | |
| is_multilingual: Optional[bool] = None, | |
| prompt_ids: Optional[tf.Tensor] = None, | |
| return_token_timestamps=None, | |
| **kwargs, | |
| ): | |
| r""" | |
| Generates sequences of token ids for models with a language modeling head. | |
| <Tip warning={true}> | |
| Most generation-controlling parameters are set in `generation_config` which, if not passed, will be set to the | |
| model's default generation configuration. You can override any `generation_config` by passing the corresponding | |
| parameters to generate, e.g. `.generate(inputs, num_beams=4, do_sample=True)`. | |
| For an overview of generation strategies and code examples, check out the [following | |
| guide](../generation_strategies). | |
| </Tip> | |
| Parameters: | |
| inputs (`tf.Tensor` of varying shape depending on the modality, *optional*): | |
| The sequence used as a prompt for the generation or as model inputs to the encoder. If unset the method | |
| initializes it with `bos_token_id` and a batch size of 1. For decoder-only models `inputs` should of in | |
| the format of `input_ids`. For encoder-decoder models *inputs* can represent any of `input_ids`, | |
| `input_values`, `input_features`, or `pixel_values`. | |
| generation_config (`~generation.GenerationConfig`, *optional*): | |
| The generation configuration to be used as base parametrization for the generation call. `**kwargs` | |
| passed to generate matching the attributes of `generation_config` will override them. If | |
| `generation_config` is not provided, the default will be used, which had the following loading | |
| priority: 1) from the `generation_config.json` model file, if it exists; 2) from the model | |
| configuration. Please note that unspecified parameters will inherit [`~generation.GenerationConfig`]'s | |
| default values, whose documentation should be checked to parameterize generation. | |
| logits_processor (`LogitsProcessorList`, *optional*): | |
| Custom logits processors that complement the default logits processors built from arguments and | |
| generation config. If a logit processor is passed that is already created with the arguments or a | |
| generation config an error is thrown. This feature is intended for advanced users. | |
| seed (`List[int]`, *optional*): | |
| Random seed to control sampling, containing two integers, used when `do_sample` is `True`. See the | |
| `seed` argument from stateless functions in `tf.random`. | |
| return_timestamps (`bool`, *optional*): | |
| Whether to return the timestamps with the text. This enables the `TFWhisperTimestampsLogitsProcessor`. | |
| task (`str`, *optional*): | |
| Task to use for generation, either "translate" or "transcribe". The `model.config.forced_decoder_ids` | |
| will be updated accordingly. | |
| language (`str`, *optional*): | |
| Language token to use for generation, can be either in the form of `<|en|>`, `en` or `english`. You can | |
| find all the possible language tokens in the `model.generation_config.lang_to_id` dictionary. | |
| is_multilingual (`bool`, *optional*): | |
| Whether or not the model is multilingual. | |
| prompt_ids (`tf.Tensor`, *optional*): | |
| Rank-1 tensor of token IDs created by passing text to [`~WhisperProcessor.get_prompt_ids`] that is | |
| provided as a prompt to each chunk. This can be used to provide or "prompt-engineer" a context for | |
| transcription, e.g. custom vocabularies or proper nouns to make it more likely to predict those words | |
| correctly. It cannot be used in conjunction with `decoder_start_token_id` as it overwrites this value. | |
| return_token_timestamps (`bool`, *optional*): | |
| Whether to return token-level timestamps with the text. This can be used with or without the | |
| `return_timestamps` option. To get word-level timestamps, use the tokenizer to group the tokens into | |
| words. | |
| kwargs (`Dict[str, Any]`, *optional*): | |
| Ad hoc parametrization of `generate_config` and/or additional model-specific kwargs that will be | |
| forwarded to the `forward` function of the model. If the model is an encoder-decoder model, encoder | |
| specific kwargs should not be prefixed and decoder specific kwargs should be prefixed with *decoder_*. | |
| Return: | |
| [`~utils.ModelOutput`] or `tf.Tensor`: A [`~utils.ModelOutput`] (if `return_dict_in_generate=True` or when | |
| `config.return_dict_in_generate=True`) or a `tf.Tensor`. | |
| If the model is *not* an encoder-decoder model (`model.config.is_encoder_decoder=False`), the possible | |
| [`~utils.ModelOutput`] types are: | |
| - [`~generation.TFGreedySearchDecoderOnlyOutput`], | |
| - [`~generation.TFSampleDecoderOnlyOutput`], | |
| - [`~generation.TFBeamSearchDecoderOnlyOutput`], | |
| - [`~generation.TFBeamSampleDecoderOnlyOutput`] | |
| If the model is an encoder-decoder model (`model.config.is_encoder_decoder=True`), the possible | |
| [`~utils.ModelOutput`] types are: | |
| - [`~generation.TFGreedySearchEncoderDecoderOutput`], | |
| - [`~generation.TFSampleEncoderDecoderOutput`], | |
| - [`~generation.TFBeamSearchEncoderDecoderOutput`], | |
| - [`~generation.TFBeamSampleEncoderDecoderOutput`] | |
| """ | |
| if generation_config is None: | |
| generation_config = self.generation_config | |
| if return_timestamps is not None: | |
| if not hasattr(generation_config, "no_timestamps_token_id"): | |
| raise ValueError( | |
| "You are trying to return timestamps, but the generation config is not properly set. " | |
| "Make sure to initialize the generation config with the correct attributes that are needed such as `no_timestamps_token_id`. " | |
| "For more details on how to generate the approtiate config, refer to https://github.com/huggingface/transformers/issues/21878#issuecomment-1451902363" | |
| ) | |
| generation_config.return_timestamps = return_timestamps | |
| else: | |
| generation_config.return_timestamps = False | |
| if language is not None: | |
| language = language.lower() | |
| generation_config.language = language | |
| if task is not None: | |
| generation_config.task = task | |
| forced_decoder_ids = None | |
| # Legacy code for backward compatibility | |
| if hasattr(self.config, "forced_decoder_ids") and self.config.forced_decoder_ids is not None: | |
| forced_decoder_ids = self.config.forced_decoder_ids | |
| elif ( | |
| hasattr(self.generation_config, "forced_decoder_ids") | |
| and self.generation_config.forced_decoder_ids is not None | |
| ): | |
| forced_decoder_ids = self.generation_config.forced_decoder_ids | |
| else: | |
| forced_decoder_ids = kwargs.get("forced_decoder_ids", None) | |
| if task is not None or language is not None or (forced_decoder_ids is None and prompt_ids is not None): | |
| forced_decoder_ids = [] | |
| if hasattr(generation_config, "language"): | |
| if generation_config.language in generation_config.lang_to_id.keys(): | |
| language_token = generation_config.language | |
| elif generation_config.language in TO_LANGUAGE_CODE.keys(): | |
| language_token = f"<|{TO_LANGUAGE_CODE[generation_config.language]}|>" | |
| elif generation_config.language in TO_LANGUAGE_CODE.values(): | |
| language_token = f"<|{generation_config.language}|>" | |
| else: | |
| is_language_code = len(generation_config.language) == 2 | |
| raise ValueError( | |
| f"Unsupported language: {generation_config.language}. Language should be one of:" | |
| f" {list(TO_LANGUAGE_CODE.values()) if is_language_code else list(TO_LANGUAGE_CODE.keys())}." | |
| ) | |
| if language_token not in generation_config.lang_to_id: | |
| raise ValueError( | |
| f"{language_token} is not supported by this specific model as it is not in the `generation_config.lang_to_id`." | |
| "(You should just add it to the generation config)" | |
| ) | |
| forced_decoder_ids.append((1, generation_config.lang_to_id[language_token])) | |
| else: | |
| forced_decoder_ids.append((1, None)) # automatically detect the language | |
| if hasattr(generation_config, "task"): | |
| if generation_config.task in TASK_IDS: | |
| forced_decoder_ids.append((2, generation_config.task_to_id[generation_config.task])) | |
| else: | |
| raise ValueError( | |
| f"The `{generation_config.task}`task is not supported. The task should be one of `{TASK_IDS}`" | |
| ) | |
| elif hasattr(generation_config, "task_to_id"): | |
| forced_decoder_ids.append((2, generation_config.task_to_id["transcribe"])) # defaults to transcribe | |
| if hasattr(generation_config, "no_timestamps_token_id") and not generation_config.return_timestamps: | |
| idx = forced_decoder_ids[-1][0] + 1 if forced_decoder_ids else 1 | |
| forced_decoder_ids.append((idx, generation_config.no_timestamps_token_id)) | |
| if forced_decoder_ids is not None: | |
| generation_config.forced_decoder_ids = forced_decoder_ids | |
| if prompt_ids is not None: | |
| if kwargs.get("decoder_start_token_id") is not None: | |
| raise ValueError( | |
| "When specifying `prompt_ids`, you cannot also specify `decoder_start_token_id` as it gets overwritten." | |
| ) | |
| prompt_ids = prompt_ids.tolist() | |
| decoder_start_token_id, *text_prompt_ids = prompt_ids | |
| # Slicing the text prompt ids in a manner consistent with the OpenAI implementation | |
| # to accommodate context space for the prefix (see https://github.com/openai/whisper/blob/c09a7ae299c4c34c5839a76380ae407e7d785914/whisper/decoding.py#L599) | |
| text_prompt_ids = text_prompt_ids[-self.config.max_length // 2 - 1 :] | |
| # Set the decoder_start_token_id to <|startofprev|> | |
| kwargs.update({"decoder_start_token_id": decoder_start_token_id}) | |
| # Update the max generation length to include the prompt | |
| specified_max_length = kwargs.pop("max_new_tokens", None) or kwargs.pop("max_length", None) | |
| default_max_length = generation_config.max_new_tokens or generation_config.max_length | |
| non_prompt_max_length = specified_max_length or default_max_length | |
| kwargs["max_new_tokens"] = non_prompt_max_length + len(text_prompt_ids) | |
| # Reformat the forced_decoder_ids to incorporate the prompt | |
| non_prompt_forced_decoder_ids = ( | |
| kwargs.pop("forced_decoder_ids", None) or generation_config.forced_decoder_ids | |
| ) | |
| forced_decoder_ids = [ | |
| *text_prompt_ids, | |
| generation_config.decoder_start_token_id, | |
| *[token for _rank, token in non_prompt_forced_decoder_ids], | |
| ] | |
| forced_decoder_ids = [(rank + 1, token) for rank, token in enumerate(forced_decoder_ids)] | |
| generation_config.forced_decoder_ids = forced_decoder_ids | |
| # TODO: Implement `WhisperTimeStampLogitsProcessor`. | |
| if generation_config.return_timestamps: | |
| # logits_processor = [TFWhisperTimeStampLogitsProcessor(generation_config)] | |
| raise ValueError("`TFWhisperForConditionalGeneration` doesn't support returning the timestamps yet.") | |
| if return_token_timestamps: | |
| kwargs["output_attentions"] = True | |
| kwargs["return_dict_in_generate"] = True | |
| if getattr(generation_config, "task", None) == "translate": | |
| logger.warning("Token-level timestamps may not be reliable for task 'translate'.") | |
| if not hasattr(generation_config, "alignment_heads"): | |
| raise ValueError( | |
| "Model generation config has no `alignment_heads`, token-level timestamps not available. " | |
| "See https://gist.github.com/hollance/42e32852f24243b748ae6bc1f985b13a on how to add this property to the generation config." | |
| ) | |
| outputs = super().generate( | |
| inputs, | |
| generation_config, | |
| logits_processor, | |
| **kwargs, | |
| ) | |
| if return_token_timestamps and hasattr(generation_config, "alignment_heads"): | |
| outputs["token_timestamps"] = self._extract_token_timestamps(outputs, generation_config.alignment_heads) | |
| return outputs | |
| def serving_output(self, output): | |
| pkv = tf.tuple(output.past_key_values)[1] if self.config.use_cache else None | |
| dec_hs = tf.convert_to_tensor(output.decoder_hidden_states) if self.config.output_hidden_states else None | |
| dec_attns = tf.convert_to_tensor(output.decoder_attentions) if self.config.output_attentions else None | |
| cross_attns = tf.convert_to_tensor(output.cross_attentions) if self.config.output_attentions else None | |
| enc_hs = tf.convert_to_tensor(output.encoder_hidden_states) if self.config.output_hidden_states else None | |
| enc_attns = tf.convert_to_tensor(output.encoder_attentions) if self.config.output_attentions else None | |
| return TFSeq2SeqLMOutput( | |
| logits=output.logits, | |
| past_key_values=pkv, | |
| decoder_hidden_states=dec_hs, | |
| decoder_attentions=dec_attns, | |
| cross_attentions=cross_attns, | |
| encoder_last_hidden_state=output.encoder_last_hidden_state, | |
| encoder_hidden_states=enc_hs, | |
| encoder_attentions=enc_attns, | |
| ) | |
| def prepare_inputs_for_generation( | |
| self, | |
| decoder_input_ids, | |
| past_key_values=None, | |
| use_cache=None, | |
| encoder_outputs=None, | |
| attention_mask=None, | |
| decoder_attention_mask=None, | |
| **kwargs, | |
| ): | |
| # cut decoder_input_ids if past is used | |
| if past_key_values is not None: | |
| decoder_input_ids = decoder_input_ids[:, -1:] | |
| if decoder_attention_mask is not None: # xla | |
| decoder_position_ids = tf.math.cumsum(decoder_attention_mask, axis=-1, exclusive=True)[:, -1:] | |
| elif past_key_values is not None: # no xla + past | |
| decoder_position_ids = past_key_values[0][0].shape[2] | |
| else: # no xla + no past | |
| decoder_position_ids = tf.range(decoder_input_ids.shape[1]) | |
| decoder_position_ids = tf.broadcast_to(decoder_position_ids, decoder_input_ids.shape) | |
| return { | |
| "input_features": None, # Needs to be passed to make Keras.layer.__call__ happy | |
| "encoder_outputs": encoder_outputs, | |
| "past_key_values": past_key_values, | |
| "decoder_input_ids": decoder_input_ids, | |
| "use_cache": use_cache, | |
| "decoder_attention_mask": decoder_attention_mask, | |
| "decoder_position_ids": decoder_position_ids, | |
| } | |
| def build(self, input_shape=None): | |
| if self.built: | |
| return | |
| self.built = True | |
| if getattr(self, "model", None) is not None: | |
| with tf.name_scope(self.model.name): | |
| self.model.build(None) | |