diff --git a/__init__.py b/__init__.py
new file mode 100644
index 000000000..d7283cb0b
--- /dev/null
+++ b/__init__.py
@@ -0,0 +1,25 @@
+# Copyright 2024 Your Company. 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.
+
+from mindnlp.utils import LazyModule
+from mindnlp.utils.import_utils import define_import_structure
+
+if TYPE_CHECKING:
+ from .configuration_mimi import *
+ from .modeling_mimi import *
+else:
+ import sys
+
+ _file = globals()["__file__"]
+ sys.modules[__name__] = LazyModule(__name__, _file, define_import_structure(_file), module_spec=__spec__)
\ No newline at end of file
diff --git a/configuration_mimi.py b/configuration_mimi.py
new file mode 100644
index 000000000..c285d4b49
--- /dev/null
+++ b/configuration_mimi.py
@@ -0,0 +1,114 @@
+import math
+import numpy as np
+from mindnlp.configs import MindNLPConfig
+
+class MimiConfig(MindNLPConfig):
+ r"""
+ This is the configuration class to store the configuration of an [`MimiModel`].
+ """
+
+ model_type = "mimi"
+
+ def __init__(
+ self,
+ sampling_rate=24_000,
+ frame_rate=12.5,
+ audio_channels=1,
+ hidden_size=512,
+ num_filters=64,
+ num_residual_layers=1,
+ upsampling_ratios=None,
+ kernel_size=7,
+ last_kernel_size=3,
+ residual_kernel_size=3,
+ dilation_growth_rate=2,
+ use_causal_conv=True,
+ pad_mode="constant",
+ compress=2,
+ trim_right_ratio=1.0,
+ codebook_size=2048,
+ codebook_dim=256,
+ num_quantizers=32,
+ use_conv_shortcut=False,
+ vector_quantization_hidden_dimension=256,
+ num_semantic_quantizers=1,
+ upsample_groups=512,
+ num_hidden_layers=8,
+ intermediate_size=2048,
+ num_attention_heads=8,
+ num_key_value_heads=8,
+ head_dim=None,
+ hidden_act="gelu",
+ max_position_embeddings=8000,
+ initializer_range=0.02,
+ norm_eps=1e-5,
+ use_cache=False,
+ rope_theta=10000.0,
+ sliding_window=250,
+ attention_dropout=0.0,
+ layer_scale_initial_scale=0.01,
+ attention_bias=False,
+ **kwargs,
+ ):
+ # Parameters for audio processing
+ self.sampling_rate = sampling_rate
+ self.frame_rate = frame_rate
+ self.audio_channels = audio_channels
+
+ # Parameters for network architecture
+ self.hidden_size = hidden_size
+ self.num_filters = num_filters
+ self.num_residual_layers = num_residual_layers
+ self.upsampling_ratios = upsampling_ratios if upsampling_ratios else [8, 6, 5, 4]
+ self.kernel_size = kernel_size
+ self.last_kernel_size = last_kernel_size
+ self.residual_kernel_size = residual_kernel_size
+ self.dilation_growth_rate = dilation_growth_rate
+ self.use_causal_conv = use_causal_conv
+ self.pad_mode = pad_mode
+ self.compress = compress
+ self.trim_right_ratio = trim_right_ratio
+ self.codebook_size = codebook_size
+ self.codebook_dim = codebook_dim if codebook_dim is not None else hidden_size
+ self.num_quantizers = num_quantizers
+ self.use_conv_shortcut = use_conv_shortcut
+ self.vector_quantization_hidden_dimension = vector_quantization_hidden_dimension
+ self.upsample_groups = upsample_groups
+
+ # Parameters for transformer architecture
+ self.num_hidden_layers = num_hidden_layers
+ self.intermediate_size = intermediate_size
+ self.num_attention_heads = num_attention_heads
+ self.num_key_value_heads = num_key_value_heads
+ self.head_dim = head_dim or hidden_size // num_attention_heads
+ self.hidden_act = hidden_act
+ self.max_position_embeddings = max_position_embeddings
+ self.initializer_range = initializer_range
+ self.norm_eps = norm_eps
+ self.use_cache = use_cache
+ self.rope_theta = rope_theta
+ self.sliding_window = sliding_window
+ self.attention_dropout = attention_dropout
+ self.layer_scale_initial_scale = layer_scale_initial_scale
+ self.attention_bias = attention_bias
+
+ # Number of semantic quantizers validation
+ if num_semantic_quantizers >= self.num_quantizers:
+ raise ValueError(
+ f"The number of semantic quantizers should be lower than the total number of quantizers {self.num_quantizers}, but is currently {num_semantic_quantizers}."
+ )
+ self.num_semantic_quantizers = num_semantic_quantizers
+
+ super().__init__(**kwargs)
+
+ @property
+ def encodec_frame_rate(self) -> int:
+ hop_length = np.prod(self.upsampling_ratios)
+ return math.ceil(self.sampling_rate / hop_length)
+
+ @property
+ def num_codebooks(self) -> int:
+ # Alias to num_quantizers
+ return self.num_quantizers
+
+__all__ = ["MimiConfig"]
diff --git a/modeling_mimi.py b/modeling_mimi.py
new file mode 100644
index 000000000..b6fef3bbe
--- /dev/null
+++ b/modeling_mimi.py
@@ -0,0 +1,1703 @@
+# coding=utf-8
+# Copyright 2024 Kyutai, and the HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""PyTorch Mimi model."""
+
+import math
+from dataclasses import dataclass
+from typing import List, Optional, Tuple, Union, Callable
+from functools import wraps
+import logging
+
+import mindspore
+import mindspore.nn as nn
+from mindspore import ops
+
+# MindNLP相关导入
+from mindnlp.abc import PreTrainedModel
+from mindnlp.models.utils import ModelOutput
+from mindnlp.transformers import (
+ AttentionMaskConverter,
+ ACT2FN,
+ Cache,
+ DynamicCache,
+ SlidingWindowCache,
+ StaticCache,
+ BaseModelOutputWithPast,
+ ROPE_INIT_FUNCTIONS
+)
+
+logger = logging.getLogger(__name__)
+
+
+# General docstring
+_CONFIG_FOR_DOC = "MimiConfig"
+
+
+@dataclass
+class MimiOutput(ModelOutput):
+ """
+ Args:
+ audio_codes (`mindspore.Tensor` of shape `(batch_size, num_quantizers, codes_length)`, *optional*):
+ Discret code embeddings computed using `model.encode`.
+ audio_values (`mindspore.Tensor` of shape `(batch_size, sequence_length)`, *optional*)
+ Decoded audio values, obtained using the decoder part of Mimi.
+ encoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the encoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ decoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the decoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ """
+
+ audio_codes: mindspore.Tensor = None
+ audio_values: mindspore.Tensor = None
+ encoder_past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None
+ decoder_past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None
+
+
+@dataclass
+class MimiEncoderOutput(ModelOutput):
+ """
+ Args:
+ audio_codes (`mindspore.Tensor` of shape `(batch_size, num_quantizers, codes_length)`, *optional*):
+ Discret code embeddings computed using `model.encode`.
+ encoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the encoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ """
+
+ audio_codes: mindspore.Tensor = None
+ encoder_past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None
+
+
+@dataclass
+class MimiDecoderOutput(ModelOutput):
+ """
+ Args:
+ audio_values (`mindspore.Tensor` of shape `(batch_size, segment_length)`, *optional*):
+ Decoded audio values, obtained using the decoder part of Mimi.
+ decoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the decoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ """
+
+ audio_values: mindspore.Tensor = None
+ decoder_past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None
+
+
+class MimiConv1d(nn.Cell):
+ """Conv1d with asymmetric or causal padding and normalization."""
+
+ def __init__(
+ self,
+ config,
+ in_channels: int,
+ out_channels: int,
+ kernel_size: int,
+ stride: int = 1,
+ dilation: int = 1,
+ groups: int = 1,
+ pad_mode=None,
+ bias: bool = True,
+ ):
+ super().__init__()
+ self.causal = config.use_causal_conv
+ self.pad_mode = config.pad_mode if pad_mode is None else pad_mode
+
+ # warn user on unusual setup between dilation and stride
+ if stride > 1 and dilation > 1:
+ logger.warning(
+ "MimiConv1d has been initialized with stride > 1 and dilation > 1"
+ f" (kernel_size={kernel_size} stride={stride}, dilation={dilation})."
+ )
+
+ self.conv = nn.Conv1d(
+ in_channels, out_channels, kernel_size, stride, dilation=dilation, groups=groups, bias=bias, pad_mode='pad'
+ )
+
+ kernel_size = self.conv.kernel_size[0]
+ stride = mindspore.Tensor(self.conv.stride[0], dtype=mindspore.int64)
+ dilation = self.conv.dilation[0]
+
+ # Effective kernel size with dilations.
+ kernel_size = mindspore.Tensor((kernel_size - 1) * dilation + 1, dtype=mindspore.int64)
+
+ # MindSpore中没有register_buffer，我们可以直接作为类属性存储
+ self.stride = stride
+ self.kernel_size = kernel_size
+ self.padding_total = mindspore.Tensor(kernel_size - stride, dtype=mindspore.int64)
+
+ # Asymmetric padding required for odd strides
+ self.padding_right = self.padding_total // 2
+ self.padding_left = self.padding_total - self.padding_right
+
+ def apply_weight_norm(self):
+ """
+ 应用权重归一化
+ 注意：这是一个简化版实现，如果需要完整功能需要进一步开发
+ """
+ # TODO: 目前MindSpore没有内置的weight_norm实现
+ # 暂时跳过weight_norm应用，保持原始权重
+ logger.warning("Weight normalization is not implemented in MindSpore yet. Skipping weight_norm application.")
+ return self.conv
+
+ def remove_weight_norm(self):
+ nn.utils.remove_weight_norm(self.conv)
+
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecConv1d._get_extra_padding_for_conv1d
+ def _get_extra_padding_for_conv1d(
+ self,
+ hidden_states: mindspore.Tensor,
+ ) -> mindspore.Tensor:
+ """See `pad_for_conv1d`."""
+ length = hidden_states.shape[-1]
+ n_frames = (length - self.kernel_size + self.padding_total) / self.stride + 1
+ n_frames = ops.ceil(n_frames).astype(mindspore.int64) - 1
+ ideal_length = n_frames * self.stride + self.kernel_size - self.padding_total
+
+ return ideal_length - length
+
+ @staticmethod
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecConv1d._pad1d
+ def _pad1d(hidden_states: mindspore.Tensor, paddings: Tuple[int, int], mode: str = "zero", value: float = 0.0):
+ """Tiny wrapper around torch.nn.functional.pad, just to allow for reflect padding on small input.
+ If this is the case, we insert extra 0 padding to the right before the reflection happens.
+ """
+ length = hidden_states.shape[-1]
+ padding_left, padding_right = paddings
+ if not mode == "reflect":
+ return nn.functional.pad(hidden_states, paddings, mode, value)
+
+ max_pad = max(padding_left, padding_right)
+ extra_pad = 0
+ if length <= max_pad:
+ extra_pad = max_pad - length + 1
+ hidden_states = nn.functional.pad(hidden_states, (0, extra_pad))
+ padded = nn.functional.pad(hidden_states, paddings, mode, value)
+ end = padded.shape[-1] - extra_pad
+ return padded[..., :end]
+
+ def forward(self, hidden_states):
+ extra_padding = self._get_extra_padding_for_conv1d(hidden_states)
+
+ if self.causal:
+ # Left padding for causal
+ hidden_states = self._pad1d(hidden_states, (self.padding_total, extra_padding), mode=self.pad_mode)
+ else:
+ hidden_states = self._pad1d(
+ hidden_states, (self.padding_left, self.padding_right + extra_padding), mode=self.pad_mode
+ )
+
+ hidden_states = self.conv(hidden_states)
+ return hidden_states
+
+
+class MimiConvTranspose1d(nn.Module):
+ """ConvTranspose1d with asymmetric or causal padding and normalization."""
+
+ def __init__(
+ self,
+ config,
+ in_channels: int,
+ out_channels: int,
+ kernel_size: int,
+ stride: int = 1,
+ groups: int = 1,
+ bias=True,
+ ):
+ super().__init__()
+ self.causal = config.use_causal_conv
+ self.trim_right_ratio = config.trim_right_ratio
+ self.conv = nn.ConvTranspose1d(in_channels, out_channels, kernel_size, stride, groups=groups, bias=bias)
+
+ if not (self.causal or self.trim_right_ratio == 1.0):
+ raise ValueError("`trim_right_ratio` != 1.0 only makes sense for causal convolutions")
+
+ kernel_size = self.conv.kernel_size[0]
+ stride = self.conv.stride[0]
+ padding_total = kernel_size - stride
+
+ # We will only trim fixed padding. Extra padding from `pad_for_conv1d` would be
+ # removed at the very end, when keeping only the right length for the output,
+ # as removing it here would require also passing the length at the matching layer
+ # in the encoder.
+ if self.causal:
+ # Trim the padding on the right according to the specified ratio
+ # if trim_right_ratio = 1.0, trim everything from right
+ self.padding_right = math.ceil(padding_total * self.trim_right_ratio)
+ else:
+ # Asymmetric padding required for odd strides
+ self.padding_right = padding_total // 2
+
+ self.padding_left = padding_total - self.padding_right
+
+ def apply_weight_norm(self):
+ weight_norm = nn.utils.weight_norm
+ if hasattr(nn.utils.parametrizations, "weight_norm"):
+ weight_norm = nn.utils.parametrizations.weight_norm
+
+ weight_norm(self.conv)
+
+ def remove_weight_norm(self):
+ nn.utils.remove_weight_norm(self.conv)
+
+ def forward(self, hidden_states):
+ hidden_states = self.conv(hidden_states)
+
+ # unpad
+ end = hidden_states.shape[-1] - self.padding_right
+ hidden_states = hidden_states[..., self.padding_left : end]
+ return hidden_states
+
+
+# Copied from transformers.models.encodec.modeling_encodec.EncodecResnetBlock with Encodec->Mimi,EnCodec->Mimi
+class MimiResnetBlock(nn.Module):
+ """
+ Residual block from SEANet model as used by Mimi.
+ """
+
+ def __init__(self, config: MimiConfig, dim: int, dilations: List[int]):
+ super().__init__()
+ kernel_sizes = (config.residual_kernel_size, 1)
+ if len(kernel_sizes) != len(dilations):
+ raise ValueError("Number of kernel sizes should match number of dilations")
+
+ hidden = dim // config.compress
+ block = []
+ for i, (kernel_size, dilation) in enumerate(zip(kernel_sizes, dilations)):
+ in_chs = dim if i == 0 else hidden
+ out_chs = dim if i == len(kernel_sizes) - 1 else hidden
+ block += [nn.ELU()]
+ block += [MimiConv1d(config, in_chs, out_chs, kernel_size, dilation=dilation)]
+ self.block = nn.ModuleList(block)
+
+ if config.use_conv_shortcut:
+ self.shortcut = MimiConv1d(config, dim, dim, kernel_size=1)
+ else:
+ self.shortcut = nn.Identity()
+
+ def forward(self, hidden_states):
+ residual = hidden_states
+ for layer in self.block:
+ hidden_states = layer(hidden_states)
+
+ return self.shortcut(residual) + hidden_states
+
+
+class MimiEncoder(nn.Module):
+ """SEANet encoder as used by Mimi."""
+
+ def __init__(self, config: MimiConfig):
+ super().__init__()
+ model = [MimiConv1d(config, config.audio_channels, config.num_filters, config.kernel_size)]
+ scaling = 1
+
+ # Downsample to raw audio scale
+ for ratio in reversed(config.upsampling_ratios):
+ current_scale = scaling * config.num_filters
+ # Add residual layers
+ for j in range(config.num_residual_layers):
+ model += [MimiResnetBlock(config, current_scale, [config.dilation_growth_rate**j, 1])]
+ # Add downsampling layers
+ model += [nn.ELU()]
+ model += [MimiConv1d(config, current_scale, current_scale * 2, kernel_size=ratio * 2, stride=ratio)]
+ scaling *= 2
+
+ model += [nn.ELU()]
+ model += [MimiConv1d(config, scaling * config.num_filters, config.hidden_size, config.last_kernel_size)]
+
+ self.layers = nn.ModuleList(model)
+
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecEncoder.forward
+ def forward(self, hidden_states):
+ for layer in self.layers:
+ hidden_states = layer(hidden_states)
+ return hidden_states
+
+
+class MimiLayerScale(nn.Module):
+ """Layer scale from [Touvron et al 2021] (https://arxiv.org/pdf/2103.17239.pdf).
+ This rescales diagonally the residual outputs close to 0, with a learnt scale.
+ """
+
+ def __init__(self, config):
+ super().__init__()
+ channels = config.hidden_size
+ initial_scale = config.layer_scale_initial_scale
+ self.scale = nn.Parameter(torch.full((channels,), initial_scale, requires_grad=True))
+
+ def forward(self, x: mindspore.Tensor):
+ return self.scale * x
+
+
+# Copied from transformers.models.llama.modeling_llama.rotate_half
+def rotate_half(x):
+ """Rotates half the hidden dims of the input."""
+ x1 = x[..., : x.shape[-1] // 2]
+ x2 = x[..., x.shape[-1] // 2 :]
+ return torch.cat((-x2, x1), dim=-1)
+
+
+# Copied from transformers.models.llama.modeling_llama.apply_rotary_pos_emb
+def apply_rotary_pos_emb(q, k, cos, sin, position_ids=None, unsqueeze_dim=1):
+ """Applies Rotary Position Embedding to the query and key tensors.
+
+ Args:
+ q (`mindspore.Tensor`): The query tensor.
+ k (`mindspore.Tensor`): The key tensor.
+ cos (`mindspore.Tensor`): The cosine part of the rotary embedding.
+ sin (`mindspore.Tensor`): The sine part of the rotary embedding.
+ position_ids (`mindspore.Tensor`, *optional*):
+ Deprecated and unused.
+ unsqueeze_dim (`int`, *optional*, defaults to 1):
+ The 'unsqueeze_dim' argument specifies the dimension along which to unsqueeze cos[position_ids] and
+ sin[position_ids] so that they can be properly broadcasted to the dimensions of q and k. For example, note
+ that cos[position_ids] and sin[position_ids] have the shape [batch_size, seq_len, head_dim]. Then, if q and
+ k have the shape [batch_size, heads, seq_len, head_dim], then setting unsqueeze_dim=1 makes
+ cos[position_ids] and sin[position_ids] broadcastable to the shapes of q and k. Similarly, if q and k have
+ the shape [batch_size, seq_len, heads, head_dim], then set unsqueeze_dim=2.
+ Returns:
+ `tuple(mindspore.Tensor)` comprising of the query and key tensors rotated using the Rotary Position Embedding.
+ """
+ cos = cos.unsqueeze(unsqueeze_dim)
+ sin = sin.unsqueeze(unsqueeze_dim)
+ q_embed = (q * cos) + (rotate_half(q) * sin)
+ k_embed = (k * cos) + (rotate_half(k) * sin)
+ return q_embed, k_embed
+
+
+class MimiMLP(nn.Module):
+ def __init__(self, config):
+ super().__init__()
+ self.config = config
+ self.activation_fn = ACT2FN[config.hidden_act]
+ self.fc1 = nn.Linear(config.hidden_size, config.intermediate_size, bias=False)
+ self.fc2 = nn.Linear(config.intermediate_size, config.hidden_size, bias=False)
+
+ # Copied from transformers.models.clip.modeling_clip.CLIPMLP.forward
+ def forward(self, hidden_states: mindspore.Tensor) -> mindspore.Tensor:
+ hidden_states = self.fc1(hidden_states)
+ hidden_states = self.activation_fn(hidden_states)
+ hidden_states = self.fc2(hidden_states)
+ return hidden_states
+
+
+# Copied from transformers.models.llama.modeling_llama.repeat_kv
+def repeat_kv(hidden_states: mindspore.Tensor, n_rep: int) -> mindspore.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)
+
+
+# copied from transformers.models.gemma.modeling_gemma.GemmaAttention with Gemma->Mimi
+# no longer copied after attention refactors
+class MimiAttention(nn.Module):
+ """Multi-headed attention from 'Attention Is All You Need' paper"""
+
+ def __init__(self, config: MimiConfig, layer_idx: Optional[int] = None):
+ super().__init__()
+ self.config = config
+ self.layer_idx = layer_idx
+ if layer_idx is None:
+ logger.warning_once(
+ f"Instantiating {self.__class__.__name__} without passing a `layer_idx` is not recommended and will "
+ "lead to errors during the forward call if caching is used. Please make sure to provide a `layer_idx` "
+ "when creating this class."
+ )
+
+ self.attention_dropout = config.attention_dropout
+ self.hidden_size = config.hidden_size
+ self.num_heads = config.num_attention_heads
+ self.head_dim = config.head_dim
+ self.num_key_value_heads = config.num_key_value_heads
+ self.num_key_value_groups = self.num_heads // self.num_key_value_heads
+ self.max_position_embeddings = config.max_position_embeddings
+ self.rope_theta = config.rope_theta
+ self.is_causal = True
+ self.scaling = 1 / math.sqrt(config.head_dim)
+
+ if self.hidden_size % self.num_heads != 0:
+ raise ValueError(
+ f"hidden_size must be divisible by num_heads (got `hidden_size`: {self.hidden_size}"
+ f" and `num_heads`: {self.num_heads})."
+ )
+
+ self.q_proj = nn.Linear(self.hidden_size, self.num_heads * self.head_dim, bias=config.attention_bias)
+ self.k_proj = nn.Linear(self.hidden_size, self.num_key_value_heads * self.head_dim, bias=config.attention_bias)
+ self.v_proj = nn.Linear(self.hidden_size, self.num_key_value_heads * self.head_dim, bias=config.attention_bias)
+ self.o_proj = nn.Linear(self.num_heads * self.head_dim, self.hidden_size, bias=config.attention_bias)
+ self.rotary_emb = MimiRotaryEmbedding(config)
+ self.sliding_window = config.sliding_window # Ignore copy
+
+ def forward(
+ self,
+ hidden_states: mindspore.Tensor,
+ attention_mask: Optional[mindspore.Tensor] = None,
+ position_ids: Optional[mindspore.LongTensor] = None,
+ past_key_value: Optional[Cache] = None,
+ output_attentions: bool = False,
+ use_cache: bool = False,
+ cache_position: Optional[mindspore.LongTensor] = None,
+ ) -> Tuple[mindspore.Tensor, Optional[mindspore.Tensor], Optional[Tuple[mindspore.Tensor]]]:
+ bsz, q_len, _ = hidden_states.size()
+
+ query_states = self.q_proj(hidden_states)
+ key_states = self.k_proj(hidden_states)
+ value_states = self.v_proj(hidden_states)
+
+ query_states = query_states.view(bsz, q_len, self.num_heads, self.head_dim).transpose(1, 2)
+ key_states = key_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
+ value_states = value_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
+
+ cos, sin = self.rotary_emb(value_states, position_ids)
+ query_states, key_states = apply_rotary_pos_emb(query_states, key_states, cos, sin)
+
+ if past_key_value is not None:
+ # sin and cos are specific to RoPE models; cache_position needed for the static cache
+ cache_kwargs = {"sin": sin, "cos": cos, "cache_position": cache_position}
+ key_states, value_states = past_key_value.update(key_states, value_states, self.layer_idx, cache_kwargs)
+
+ key_states = repeat_kv(key_states, self.num_key_value_groups)
+ value_states = repeat_kv(value_states, self.num_key_value_groups)
+
+ attn_weights = torch.matmul(query_states, key_states.transpose(2, 3)) * self.scaling
+
+ if attention_mask is not None: # no matter the length, we just slice it
+ causal_mask = attention_mask[:, :, :, : key_states.shape[-2]]
+ attn_weights = attn_weights + causal_mask
+
+ # upcast attention to fp32
+ attn_weights = nn.functional.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query_states.dtype)
+ attn_weights = nn.functional.dropout(attn_weights, p=self.attention_dropout, training=self.training)
+ attn_output = torch.matmul(attn_weights, value_states)
+
+ if attn_output.size() != (bsz, self.num_heads, q_len, self.head_dim):
+ raise ValueError(
+ f"`attn_output` should be of size {(bsz, self.num_heads, q_len, self.head_dim)}, but is"
+ f" {attn_output.size()}"
+ )
+
+ attn_output = attn_output.transpose(1, 2).contiguous()
+
+ attn_output = attn_output.view(bsz, q_len, -1)
+ attn_output = self.o_proj(attn_output)
+
+ if not output_attentions:
+ attn_weights = None
+
+ return attn_output, attn_weights, past_key_value
+
+
+# NO LONGER EXIST Copied from transformers.models.gemma.modeling_gemma.GemmaFlashAttention2 with Gemma->Mimi
+# TODO cyril: modular
+class MimiFlashAttention2(MimiAttention):
+ """
+ Mimi flash attention module. This module inherits from `MimiAttention` as the weights of the module stays
+ untouched. The only required change would be on the forward pass where it needs to correctly call the public API of
+ flash attention and deal with padding tokens in case the input contains any of them.
+ """
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # TODO: Should be removed once Flash Attention for RoCm is bumped to 2.1.
+ # flash_attn<2.1 generates top-left aligned causal mask, while what is needed here is bottom-right alignement, that was made default for flash_attn>=2.1. This attribute is used to handle this difference. Reference: https://github.com/Dao-AILab/flash-attention/releases/tag/v2.1.0.
+ # Beware that with flash_attn<2.1, using q_seqlen != k_seqlen (except for the case q_seqlen == 1) produces a wrong mask (top-left).
+ self._flash_attn_uses_top_left_mask = not is_flash_attn_greater_or_equal_2_10()
+
+ def forward(
+ self,
+ hidden_states: mindspore.Tensor,
+ attention_mask: Optional[mindspore.LongTensor] = None,
+ position_ids: Optional[mindspore.LongTensor] = None,
+ past_key_value: Optional[Cache] = None,
+ output_attentions: bool = False,
+ use_cache: bool = False,
+ cache_position: Optional[mindspore.LongTensor] = None,
+ ) -> Tuple[mindspore.Tensor, Optional[mindspore.Tensor], Optional[Tuple[mindspore.Tensor]]]:
+ if isinstance(past_key_value, StaticCache):
+ raise ValueError(
+ "`static` cache implementation is not compatible with `attn_implementation==flash_attention_2` "
+ "make sure to use `sdpa` in the mean time, and open an issue at https://github.com/huggingface/transformers"
+ )
+
+ output_attentions = False
+
+ bsz, q_len, _ = hidden_states.size()
+
+ query_states = self.q_proj(hidden_states)
+ key_states = self.k_proj(hidden_states)
+ value_states = self.v_proj(hidden_states)
+
+ # Flash attention requires the input to have the shape
+ # batch_size x seq_length x head_dim x hidden_dim
+ # therefore we just need to keep the original shape
+ query_states = query_states.view(bsz, q_len, self.num_heads, self.head_dim).transpose(1, 2)
+ key_states = key_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
+ value_states = value_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
+
+ cos, sin = self.rotary_emb(value_states, position_ids)
+ query_states, key_states = apply_rotary_pos_emb(query_states, key_states, cos, sin)
+
+ if past_key_value is not None:
+ # sin and cos are specific to RoPE models; cache_position needed for the static cache
+ cache_kwargs = {"sin": sin, "cos": cos, "cache_position": cache_position}
+ key_states, value_states = past_key_value.update(key_states, value_states, self.layer_idx, cache_kwargs)
+
+ # TODO: These transpose are quite inefficient but Flash Attention requires the layout [batch_size, sequence_length, num_heads, head_dim]. We would need to refactor the KV cache
+ # to be able to avoid many of these transpose/reshape/view.
+ query_states = query_states.transpose(1, 2)
+ key_states = key_states.transpose(1, 2)
+ value_states = value_states.transpose(1, 2)
+
+ dropout_rate = self.attention_dropout if self.training else 0.0
+
+ # In PEFT, usually we cast the layer norms in float32 for training stability reasons
+ # therefore the input hidden states gets silently casted in float32. Hence, we need
+ # cast them back in the correct dtype just to be sure everything works as expected.
+ # This might slowdown training & inference so it is recommended to not cast the LayerNorms
+ # in fp32. (MimiRMSNorm handles it correctly)
+
+ input_dtype = query_states.dtype
+ if input_dtype == torch.float32:
+ if torch.is_autocast_enabled():
+ target_dtype = torch.get_autocast_gpu_dtype()
+ # Handle the case where the model is quantized
+ elif hasattr(self.config, "_pre_quantization_dtype"):
+ target_dtype = self.config._pre_quantization_dtype
+ else:
+ target_dtype = self.q_proj.weight.dtype
+
+ logger.warning_once(
+ f"The input hidden states seems to be silently casted in float32, this might be related to"
+ f" the fact you have upcasted embedding or layer norm layers in float32. We will cast back the input in"
+ f" {target_dtype}."
+ )
+
+ query_states = query_states.to(target_dtype)
+ key_states = key_states.to(target_dtype)
+ value_states = value_states.to(target_dtype)
+
+ attn_output = _flash_attention_forward(
+ query_states,
+ key_states,
+ value_states,
+ attention_mask,
+ q_len,
+ position_ids=position_ids,
+ dropout=dropout_rate,
+ sliding_window=getattr(self, "sliding_window", None),
+ is_causal=self.is_causal,
+ use_top_left_mask=self._flash_attn_uses_top_left_mask,
+ )
+
+ attn_output = attn_output.reshape(bsz, q_len, -1).contiguous()
+ attn_output = self.o_proj(attn_output)
+
+ if not output_attentions:
+ attn_weights = None
+
+ return attn_output, attn_weights, past_key_value
+
+
+# NO LONGER EXIST Copied from transformers.models.gemma.modeling_gemma.GemmaSdpaAttention with Gemma->Mimi
+# TODO cyril: modular
+class MimiSdpaAttention(MimiAttention):
+ """
+ Mimi attention module using torch.nn.functional.scaled_dot_product_attention. This module inherits from
+ `MimiAttention` as the weights of the module stays untouched. The only changes are on the forward pass to adapt to
+ SDPA API.
+ """
+
+ # Adapted from MimiAttention.forward
+ def forward(
+ self,
+ hidden_states: mindspore.Tensor,
+ attention_mask: Optional[mindspore.Tensor] = None,
+ position_ids: Optional[mindspore.LongTensor] = None,
+ past_key_value: Optional[Cache] = None,
+ output_attentions: bool = False,
+ use_cache: bool = False,
+ cache_position: Optional[mindspore.LongTensor] = None,
+ **kwargs,
+ ) -> Tuple[mindspore.Tensor, Optional[mindspore.Tensor], Optional[Tuple[mindspore.Tensor]]]:
+ if output_attentions:
+ # TODO: Improve this warning with e.g. `model.config.attn_implementation = "manual"` once this is implemented.
+ logger.warning_once(
+ "MimiModel is using MimiSdpaAttention, but `torch.nn.functional.scaled_dot_product_attention` does not support `output_attentions=True`. Falling back to the manual attention implementation, "
+ 'but specifying the manual implementation will be required from Transformers version v5.0.0 onwards. This warning can be removed using the argument `attn_implementation="eager"` when loading the model.'
+ )
+ return super().forward(
+ hidden_states=hidden_states,
+ attention_mask=attention_mask,
+ position_ids=position_ids,
+ past_key_value=past_key_value,
+ output_attentions=output_attentions,
+ use_cache=use_cache,
+ cache_position=cache_position,
+ )
+
+ bsz, q_len, _ = hidden_states.size()
+
+ query_states = self.q_proj(hidden_states)
+ key_states = self.k_proj(hidden_states)
+ value_states = self.v_proj(hidden_states)
+
+ query_states = query_states.view(bsz, q_len, self.num_heads, self.head_dim).transpose(1, 2)
+ key_states = key_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
+ value_states = value_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
+
+ cos, sin = self.rotary_emb(value_states, position_ids)
+ query_states, key_states = apply_rotary_pos_emb(query_states, key_states, cos, sin)
+
+ if past_key_value is not None:
+ # sin and cos are specific to RoPE models; cache_position needed for the static cache
+ cache_kwargs = {"sin": sin, "cos": cos, "cache_position": cache_position}
+ key_states, value_states = past_key_value.update(key_states, value_states, self.layer_idx, cache_kwargs)
+
+ key_states = repeat_kv(key_states, self.num_key_value_groups)
+ value_states = repeat_kv(value_states, self.num_key_value_groups)
+
+ causal_mask = attention_mask
+ if attention_mask is not None:
+ causal_mask = causal_mask[:, :, :, : key_states.shape[-2]]
+
+ # SDPA with memory-efficient backend is currently (torch==2.1.2) bugged with non-contiguous inputs with custom attn_mask,
+ # Reference: https://github.com/pytorch/pytorch/issues/112577.
+ if query_states.device.type == "cuda" and causal_mask is not None:
+ query_states = query_states.contiguous()
+ key_states = key_states.contiguous()
+ value_states = value_states.contiguous()
+
+ # We dispatch to SDPA's Flash Attention or Efficient kernels via this `is_causal` if statement instead of an inline conditional assignment
+ # in SDPA to support both torch.compile's dynamic shapes and full graph options. An inline conditional prevents dynamic shapes from compiling.
+ is_causal = True if causal_mask is None and q_len > 1 else False
+
+ attn_output = torch.nn.functional.scaled_dot_product_attention(
+ query_states,
+ key_states,
+ value_states,
+ attn_mask=causal_mask,
+ dropout_p=self.attention_dropout if self.training else 0.0,
+ is_causal=is_causal,
+ )
+
+ attn_output = attn_output.transpose(1, 2).contiguous()
+ attn_output = attn_output.view(bsz, q_len, -1)
+
+ attn_output = self.o_proj(attn_output)
+
+ return attn_output, None, past_key_value
+
+
+MIMI_ATTENTION_CLASSES = {
+ "eager": MimiAttention,
+ "flash_attention_2": MimiFlashAttention2,
+ "sdpa": MimiSdpaAttention,
+}
+
+
+class MimiTransformerLayer(nn.Module):
+ def __init__(self, config: MimiConfig, layer_idx: int):
+ super().__init__()
+ self.hidden_size = config.hidden_size
+
+ self.self_attn = MIMI_ATTENTION_CLASSES[config._attn_implementation](config=config, layer_idx=layer_idx)
+
+ self.mlp = MimiMLP(config)
+ self.input_layernorm = nn.LayerNorm(config.hidden_size, eps=config.norm_eps)
+ self.post_attention_layernorm = nn.LayerNorm(config.hidden_size, eps=config.norm_eps)
+ self.self_attn_layer_scale = MimiLayerScale(config)
+ self.mlp_layer_scale = MimiLayerScale(config)
+
+ def forward(
+ self,
+ hidden_states: mindspore.Tensor,
+ attention_mask: Optional[mindspore.Tensor] = None,
+ position_ids: Optional[mindspore.LongTensor] = None,
+ past_key_value: Optional[Cache] = None,
+ output_attentions: Optional[bool] = False,
+ use_cache: Optional[bool] = False,
+ cache_position: Optional[mindspore.LongTensor] = None,
+ **kwargs,
+ ) -> Tuple[mindspore.Tensor, Optional[Tuple[mindspore.Tensor, mindspore.Tensor]]]:
+ """
+ Args:
+ hidden_states (`mindspore.Tensor`): input to the layer of shape `(batch, seq_len, embed_dim)`
+ attention_mask (`mindspore.Tensor`, *optional*):
+ attention mask of size `(batch_size, sequence_length)` if flash attention is used or `(batch_size, 1,
+ query_sequence_length, key_sequence_length)` if default attention is used.
+ output_attentions (`bool`, *optional*):
+ Whether or not to return the attentions tensors of all attention layers. See `attentions` under returned
+ tensors for more detail.
+ 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`).
+ past_key_value (`Tuple(mindspore.Tensor)`, *optional*): cached past key and value projection states
+ cache_position (`mindspore.LongTensor` of shape `(sequence_length)`, *optional*):
+ Indices depicting the position of the input sequence tokens in the sequence
+ kwargs (`dict`, *optional*):
+ Arbitrary kwargs to be ignored, used for FSDP and other methods that injects code
+ into the model
+ """
+ residual = hidden_states
+
+ hidden_states = self.input_layernorm(hidden_states)
+
+ # Self Attention
+ hidden_states, self_attn_weights, present_key_value = self.self_attn(
+ hidden_states=hidden_states,
+ attention_mask=attention_mask,
+ position_ids=position_ids,
+ past_key_value=past_key_value,
+ output_attentions=output_attentions,
+ use_cache=use_cache,
+ cache_position=cache_position,
+ **kwargs,
+ )
+ hidden_states = residual + self.self_attn_layer_scale(hidden_states)
+
+ # Fully Connected
+ residual = hidden_states
+ hidden_states = self.post_attention_layernorm(hidden_states)
+ hidden_states = self.mlp(hidden_states)
+ hidden_states = residual + self.mlp_layer_scale(hidden_states)
+
+ outputs = (hidden_states,)
+
+ if output_attentions:
+ outputs += (self_attn_weights,)
+
+ if use_cache:
+ outputs += (present_key_value,)
+
+ return outputs
+
+
+class MimiTransformerModel(nn.Module):
+ """
+ Transformer decoder consisting of *config.num_hidden_layers* layers. Each layer is a [`MimiTransformerLayer`]
+
+ Args:
+ config: MimiConfig
+ """
+
+ def __init__(self, config: MimiConfig):
+ super().__init__()
+
+ self.layers = nn.ModuleList(
+ [MimiTransformerLayer(config, layer_idx) for layer_idx in range(config.num_hidden_layers)]
+ )
+ self._attn_implementation = config._attn_implementation
+
+ self.gradient_checkpointing = False
+ self.config = config
+
+ def forward(
+ self,
+ hidden_states: mindspore.Tensor = None,
+ attention_mask: Optional[mindspore.Tensor] = None,
+ position_ids: Optional[mindspore.LongTensor] = None,
+ past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None,
+ use_cache: Optional[bool] = None,
+ output_attentions: Optional[bool] = None,
+ output_hidden_states: Optional[bool] = None,
+ return_dict: Optional[bool] = None,
+ cache_position: Optional[mindspore.LongTensor] = None,
+ ) -> Union[Tuple, BaseModelOutputWithPast]:
+ """
+ Args:
+ hidden_states (`mindspore.Tensor` of shape `(batch_size, sequence_length, hidden_size)`, *optional*):
+ Embedded representation that will be contextualized by the model
+ attention_mask (`mindspore.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)
+
+ Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and
+ [`PreTrainedTokenizer.__call__`] for details.
+
+ If `past_key_values` is used, optionally only the last `decoder_input_ids` have to be input (see
+ `past_key_values`).
+
+ If you want to change padding behavior, you should read [`modeling_opt._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.
+
+ - 1 indicates the head is **not masked**,
+ - 0 indicates the head is **masked**.
+ position_ids (`mindspore.LongTensor` of shape `(batch_size, sequence_length)`, *optional*):
+ Indices of positions of each input sequence tokens in the position embeddings. Selected in the range `[0,
+ config.n_positions - 1]`.
+
+ [What are position IDs?](../glossary#position-ids)
+ past_key_values (`Cache` or `tuple(tuple(mindspore.Tensor))`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks and in the cross-attention
+ blocks) that can be used to speed up sequential decoding. This typically consists in the `past_key_values`
+ returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ Two formats are allowed:
+ - a [`~cache_utils.Cache`] instance;
+ - Tuple of `tuple(mindspore.Tensor)` of length `config.n_layers`, with each tuple having 2 tensors of
+ shape `(batch_size, num_heads, sequence_length, embed_size_per_head)`). This is also known as the legacy
+ cache format.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `input_ids` (those that don't
+ have their past key value states given to this model) of shape `(batch_size, 1)` instead of all `input_ids`
+ of shape `(batch_size, sequence_length)`.
+ 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.
+ """
+ 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 self.gradient_checkpointing and self.training and use_cache:
+ logger.warning_once(
+ "`use_cache=True` is incompatible with gradient checkpointing. Setting `use_cache=False`."
+ )
+ use_cache = False
+
+ if use_cache and not isinstance(past_key_values, Cache):
+ if past_key_values is None:
+ past_key_values = DynamicCache()
+ else:
+ past_key_values = DynamicCache.from_legacy_cache(past_key_values)
+ logger.warning_once(
+ "We detected that you are passing `past_key_values` as a tuple of tuples. This is deprecated and "
+ "will be removed in v4.47. Please convert your cache or use an appropriate `Cache` class "
+ "(https://huggingface.co/docs/transformers/kv_cache#legacy-cache-format)"
+ )
+
+ if cache_position is None:
+ past_seen_tokens = past_key_values.get_seq_length() if past_key_values is not None else 0
+ cache_position = torch.arange(
+ past_seen_tokens, past_seen_tokens + hidden_states.shape[1], device=hidden_states.device
+ )
+
+ if position_ids is None:
+ position_ids = cache_position.unsqueeze(0)
+
+ causal_mask = None
+ if attention_mask is not None:
+ causal_mask = self._update_causal_mask(
+ attention_mask, hidden_states, cache_position, past_key_values, output_attentions
+ )
+
+ # decoder layers
+ all_hidden_states = () if output_hidden_states else None
+ all_self_attns = () if output_attentions else None
+ next_decoder_cache = None
+
+ for decoder_layer in self.layers:
+ if output_hidden_states:
+ all_hidden_states += (hidden_states,)
+
+ if self.gradient_checkpointing and self.training:
+ layer_outputs = self._gradient_checkpointing_func(
+ decoder_layer.__call__,
+ hidden_states,
+ causal_mask,
+ position_ids,
+ past_key_values,
+ output_attentions,
+ use_cache,
+ cache_position,
+ )
+ else:
+ layer_outputs = decoder_layer(
+ hidden_states,
+ attention_mask=causal_mask,
+ position_ids=position_ids,
+ past_key_value=past_key_values,
+ output_attentions=output_attentions,
+ use_cache=use_cache,
+ cache_position=cache_position,
+ )
+
+ hidden_states = layer_outputs[0]
+
+ if use_cache:
+ next_decoder_cache = layer_outputs[2 if output_attentions else 1]
+
+ if output_attentions:
+ all_self_attns += (layer_outputs[1],)
+
+ # 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] if v is not None)
+
+ return BaseModelOutputWithPast(
+ last_hidden_state=hidden_states,
+ past_key_values=next_cache,
+ hidden_states=all_hidden_states,
+ attentions=all_self_attns,
+ )
+
+ # Copied from transformers.models.phi3.modeling_phi3.Phi3Model._update_causal_mask with Phi3->Mimi
+ def _update_causal_mask(
+ self,
+ attention_mask: mindspore.Tensor,
+ input_tensor: mindspore.Tensor,
+ cache_position: mindspore.Tensor,
+ past_key_values: Cache,
+ output_attentions: bool,
+ ):
+ if self.config._attn_implementation == "flash_attention_2":
+ if attention_mask is not None and past_key_values is not None:
+ is_padding_right = attention_mask[:, -1].sum().item() != input_tensor.size()[0]
+ if is_padding_right:
+ raise ValueError(
+ "You are attempting to perform batched generation with padding_side='right'"
+ " this may lead to unexpected behaviour for Flash Attention version of Mimi. Make sure to "
+ " call `tokenizer.padding_side = 'left'` before tokenizing the input. "
+ )
+ if attention_mask is not None and 0.0 in attention_mask:
+ return attention_mask
+ return None
+
+ # For SDPA, when possible, we will rely on its `is_causal` argument instead of its `attn_mask` argument, in
+ # order to dispatch on Flash Attention 2. This feature is not compatible with static cache, as SDPA will fail
+ # to infer the attention mask.
+ past_seen_tokens = past_key_values.get_seq_length() if past_key_values is not None else 0
+ using_static_cache = isinstance(past_key_values, StaticCache)
+ using_sliding_window_cache = isinstance(past_key_values, SlidingWindowCache)
+
+ # When output attentions is True, sdpa implementation's forward method calls the eager implementation's forward
+ if (
+ self.config._attn_implementation == "sdpa"
+ and not (using_static_cache or using_sliding_window_cache)
+ and not output_attentions
+ ):
+ if AttentionMaskConverter._ignore_causal_mask_sdpa(
+ attention_mask,
+ inputs_embeds=input_tensor,
+ past_key_values_length=past_seen_tokens,
+ sliding_window=self.config.sliding_window,
+ is_training=self.training,
+ ):
+ return None
+
+ dtype, device = input_tensor.dtype, input_tensor.device
+ min_dtype = torch.finfo(dtype).min
+ sequence_length = input_tensor.shape[1]
+ # SlidingWindowCache or StaticCache
+ if using_sliding_window_cache or using_static_cache:
+ target_length = past_key_values.get_max_cache_shape()
+ # DynamicCache or no cache
+ else:
+ target_length = (
+ attention_mask.shape[-1]
+ if isinstance(attention_mask, mindspore.Tensor)
+ else past_seen_tokens + sequence_length + 1
+ )
+
+ # In case the provided `attention` mask is 2D, we generate a causal mask here (4D).
+ causal_mask = self._prepare_4d_causal_attention_mask_with_cache_position(
+ attention_mask,
+ sequence_length=sequence_length,
+ target_length=target_length,
+ dtype=dtype,
+ device=device,
+ cache_position=cache_position,
+ batch_size=input_tensor.shape[0],
+ config=self.config,
+ past_key_values=past_key_values,
+ )
+
+ if (
+ self.config._attn_implementation == "sdpa"
+ and attention_mask is not None
+ and attention_mask.device.type == "cuda"
+ and not output_attentions
+ ):
+ # Attend to all tokens in fully masked rows in the causal_mask, for example the relevant first rows when
+ # using left padding. This is required by F.scaled_dot_product_attention memory-efficient attention path.
+ # Details: https://github.com/pytorch/pytorch/issues/110213
+ causal_mask = AttentionMaskConverter._unmask_unattended(causal_mask, min_dtype)
+
+ return causal_mask
+
+ @staticmethod
+ # Copied from transformers.models.mistral.modeling_mistral.MistralModel._prepare_4d_causal_attention_mask_with_cache_position with Mistral->Mimi
+ def _prepare_4d_causal_attention_mask_with_cache_position(
+ attention_mask: mindspore.Tensor,
+ sequence_length: int,
+ target_length: int,
+ dtype: torch.dtype,
+ device: torch.device,
+ cache_position: mindspore.Tensor,
+ batch_size: int,
+ config: MimiConfig,
+ past_key_values: Cache,
+ ):
+ """
+ Creates a causal 4D mask of shape `(batch_size, 1, query_length, key_value_length)` from a 2D mask of shape
+ `(batch_size, key_value_length)`, or if the input `attention_mask` is already 4D, do nothing.
+
+ Args:
+ attention_mask (`mindspore.Tensor`):
+ A 2D attention mask of shape `(batch_size, key_value_length)` or a 4D attention mask of shape `(batch_size, 1, query_length, key_value_length)`.
+ sequence_length (`int`):
+ The sequence length being processed.
+ target_length (`int`):
+ The target length: when generating with static cache, the mask should be as long as the static cache, to account for the 0 padding, the part of the cache that is not filled yet.
+ dtype (`torch.dtype`):
+ The dtype to use for the 4D attention mask.
+ device (`torch.device`):
+ The device to plcae the 4D attention mask on.
+ cache_position (`mindspore.Tensor`):
+ Indices depicting the position of the input sequence tokens in the sequence.
+ batch_size (`mindspore.Tensor`):
+ Batch size.
+ config (`MimiConfig`):
+ The model's configuration class
+ past_key_values (`Cache`):
+ The cache class that is being used currently to generate
+ """
+ if attention_mask is not None and attention_mask.dim() == 4:
+ # In this case we assume that the mask comes already in inverted form and requires no inversion or slicing.
+ causal_mask = attention_mask
+ else:
+ min_dtype = torch.finfo(dtype).min
+ causal_mask = torch.full(
+ (sequence_length, target_length), fill_value=min_dtype, dtype=dtype, device=device
+ )
+ diagonal_attend_mask = torch.arange(target_length, device=device) > cache_position.reshape(-1, 1)
+ if config.sliding_window is not None:
+ # if we have sliding window, we should not attend to tokens beyond sliding window length, so we mask them out also
+ # the check is needed to verify is current checkpoint was trained with sliding window or not
+ if not isinstance(past_key_values, SlidingWindowCache) or sequence_length > target_length:
+ sliding_attend_mask = torch.arange(target_length, device=device) <= (
+ cache_position.reshape(-1, 1) - config.sliding_window
+ )
+ diagonal_attend_mask.bitwise_or_(sliding_attend_mask)
+ causal_mask *= diagonal_attend_mask
+ causal_mask = causal_mask[None, None, :, :].expand(batch_size, 1, -1, -1)
+ if attention_mask is not None:
+ causal_mask = causal_mask.clone() # copy to contiguous memory for in-place edit
+ if attention_mask.shape[-1] > target_length:
+ attention_mask = attention_mask[:, :target_length]
+ mask_length = attention_mask.shape[-1]
+ padding_mask = causal_mask[:, :, :, :mask_length] + attention_mask[:, None, None, :]
+ padding_mask = padding_mask == 0
+ causal_mask[:, :, :, :mask_length] = causal_mask[:, :, :, :mask_length].masked_fill(
+ padding_mask, min_dtype
+ )
+ return causal_mask
+
+
+class MimiDecoder(nn.Module):
+ """SEANet decoder as used by Mimi."""
+
+ def __init__(self, config: MimiConfig):
+ super().__init__()
+ scaling = int(2 ** len(config.upsampling_ratios))
+ model = [MimiConv1d(config, config.hidden_size, scaling * config.num_filters, config.kernel_size)]
+
+ # Upsample to raw audio scale
+ for ratio in config.upsampling_ratios:
+ current_scale = scaling * config.num_filters
+ # Add upsampling layers
+ model += [nn.ELU()]
+ model += [
+ MimiConvTranspose1d(config, current_scale, current_scale // 2, kernel_size=ratio * 2, stride=ratio)
+ ]
+ # Add residual layers
+ for j in range(config.num_residual_layers):
+ model += [MimiResnetBlock(config, current_scale // 2, (config.dilation_growth_rate**j, 1))]
+ scaling //= 2
+
+ # Add final layers
+ model += [nn.ELU()]
+ model += [MimiConv1d(config, config.num_filters, config.audio_channels, config.last_kernel_size)]
+ self.layers = nn.ModuleList(model)
+
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecDecoder.forward
+ def forward(self, hidden_states):
+ for layer in self.layers:
+ hidden_states = layer(hidden_states)
+ return hidden_states
+
+
+class MimiEuclideanCodebook(nn.Module):
+ """Codebook with Euclidean distance."""
+
+ def __init__(self, config: MimiConfig, epsilon: float = 1e-5):
+ super().__init__()
+ embed = torch.zeros(config.codebook_size, config.codebook_dim)
+
+ self.codebook_size = config.codebook_size
+
+ self.register_buffer("initialized", torch.Tensor([True]))
+ self.register_buffer("cluster_usage", torch.ones(config.codebook_size))
+ self.register_buffer("embed_sum", embed)
+ self._embed = None
+ self.epsilon = epsilon
+
+ @property
+ def embed(self) -> mindspore.Tensor:
+ if self._embed is None:
+ self._embed = self.embed_sum / self.cluster_usage.clamp(min=self.epsilon)[:, None]
+ return self._embed
+
+ def quantize(self, hidden_states):
+ # Projects each vector in `hidden_states` over the nearest centroid and return its index.
+ # `hidden_states` should be `[N, D]` with `N` the number of input vectors and `D` the dimension.
+ dists = torch.cdist(hidden_states[None], self.embed[None], p=2)[0]
+ embed_ind = dists.argmin(dim=-1)
+ return embed_ind
+
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecEuclideanCodebook.encode
+ def encode(self, hidden_states):
+ shape = hidden_states.shape
+ # pre-process
+ hidden_states = hidden_states.reshape((-1, shape[-1]))
+ # quantize
+ embed_ind = self.quantize(hidden_states)
+ # post-process
+ embed_ind = embed_ind.view(*shape[:-1])
+ return embed_ind
+
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecEuclideanCodebook.decode
+ def decode(self, embed_ind):
+ quantize = nn.functional.embedding(embed_ind, self.embed)
+ return quantize
+
+
+# Copied from transformers.models.encodec.modeling_encodec.EncodecVectorQuantization with Encodec->Mimi
+class MimiVectorQuantization(nn.Module):
+ """
+ Vector quantization implementation. Currently supports only euclidean distance.
+ """
+
+ def __init__(self, config: MimiConfig):
+ super().__init__()
+ self.codebook = MimiEuclideanCodebook(config)
+
+ def encode(self, hidden_states):
+ hidden_states = hidden_states.permute(0, 2, 1)
+ embed_in = self.codebook.encode(hidden_states)
+ return embed_in
+
+ def decode(self, embed_ind):
+ quantize = self.codebook.decode(embed_ind)
+ quantize = quantize.permute(0, 2, 1)
+ return quantize
+
+
+class MimiResidualVectorQuantizer(nn.Module):
+ """Residual Vector Quantizer."""
+
+ def __init__(self, config: MimiConfig, num_quantizers: int = None):
+ super().__init__()
+ self.codebook_size = config.codebook_size
+ self.frame_rate = config.frame_rate
+ self.num_quantizers = num_quantizers if num_quantizers is not None else config.num_quantizers
+ self.layers = nn.ModuleList([MimiVectorQuantization(config) for _ in range(self.num_quantizers)])
+
+ self.input_proj = None
+ self.output_proj = None
+ if config.vector_quantization_hidden_dimension != config.hidden_size:
+ self.input_proj = torch.nn.Conv1d(
+ config.hidden_size, config.vector_quantization_hidden_dimension, 1, bias=False
+ )
+ self.output_proj = torch.nn.Conv1d(
+ config.vector_quantization_hidden_dimension, config.hidden_size, 1, bias=False
+ )
+
+ def encode(self, embeddings: mindspore.Tensor, num_quantizers: Optional[int] = None) -> mindspore.Tensor:
+ """
+ Encode a given input tensor with the specified frame rate at the given number of quantizers / codebooks. The RVQ encode method sets
+ the appropriate number of quantizers to use and returns indices for each quantizer.
+ """
+ if self.input_proj is not None:
+ embeddings = self.input_proj(embeddings)
+
+ num_quantizers = num_quantizers if num_quantizers is not None else self.num_quantizers
+
+ residual = embeddings
+ all_indices = []
+ for layer in self.layers[:num_quantizers]:
+ indices = layer.encode(residual)
+ quantized = layer.decode(indices)
+ residual = residual - quantized
+ all_indices.append(indices)
+ out_indices = torch.stack(all_indices)
+ return out_indices
+
+ def decode(self, codes: mindspore.Tensor) -> mindspore.Tensor:
+ """Decode the given codes of shape [B, K, T] to the quantized representation."""
+ quantized_out = torch.tensor(0.0, device=codes.device)
+ codes = codes.transpose(0, 1)
+ for i, indices in enumerate(codes):
+ layer = self.layers[i]
+ quantized = layer.decode(indices)
+ quantized_out = quantized_out + quantized
+
+ if self.output_proj is not None:
+ quantized_out = self.output_proj(quantized_out)
+ return quantized_out
+
+
+class MimiSplitResidualVectorQuantizer(nn.Module):
+ """Split Residual Vector Quantizer."""
+
+ def __init__(self, config: MimiConfig):
+ super().__init__()
+ self.codebook_size = config.codebook_size
+ self.frame_rate = config.frame_rate
+ self.max_num_quantizers = config.num_quantizers
+
+ self.num_semantic_quantizers = config.num_semantic_quantizers
+ self.num_acoustic_quantizers = config.num_quantizers - config.num_semantic_quantizers
+
+ self.semantic_residual_vector_quantizer = MimiResidualVectorQuantizer(config, self.num_semantic_quantizers)
+ self.acoustic_residual_vector_quantizer = MimiResidualVectorQuantizer(config, self.num_acoustic_quantizers)
+
+ def encode(self, embeddings: mindspore.Tensor, num_quantizers: Optional[float] = None) -> mindspore.Tensor:
+ """
+ Encode a given input tensor with the specified frame rate at the given number of quantizers / codebooks. The RVQ encode method sets
+ the appropriate number of quantizers to use and returns indices for each quantizer.
+ """
+
+ num_quantizers = self.max_num_quantizers if num_quantizers is None else num_quantizers
+
+ if num_quantizers > self.max_num_quantizers:
+ raise ValueError(
+ f"The number of quantizers (i.e codebooks) asked should be lower than the total number of quantizers {self.max_num_quantizers}, but is currently {num_quantizers}."
+ )
+
+ if num_quantizers < self.num_semantic_quantizers:
+ raise ValueError(
+ f"The number of quantizers (i.e codebooks) asked should be higher than the number of semantic quantizers {self.num_semantic_quantizers}, but is currently {num_quantizers}."
+ )
+
+ # codes is [K, B, T], with T frames, K nb of codebooks.
+ codes = self.semantic_residual_vector_quantizer.encode(embeddings)
+
+ if num_quantizers > self.num_semantic_quantizers:
+ acoustic_codes = self.acoustic_residual_vector_quantizer.encode(
+ embeddings, num_quantizers=num_quantizers - self.num_semantic_quantizers
+ )
+ codes = torch.cat([codes, acoustic_codes], dim=0)
+
+ return codes
+
+ def decode(self, codes: mindspore.Tensor) -> mindspore.Tensor:
+ """Decode the given codes to the quantized representation."""
+
+ # The first num_semantic_quantizers codebooks are decoded using the semantic RVQ
+ quantized_out = self.semantic_residual_vector_quantizer.decode(codes[:, : self.num_semantic_quantizers])
+
+ # The rest of the codebooks are decoded using the acoustic RVQ
+ if codes.shape[1] > self.num_semantic_quantizers:
+ quantized_out += self.acoustic_residual_vector_quantizer.decode(codes[:, self.num_semantic_quantizers :])
+ return quantized_out
+
+
+class MimiPreTrainedModel(PreTrainedModel):
+ """
+ An abstract class to handle weights initialization and a simple interface for downloading and loading pretrained
+ models.
+ """
+
+ config_class = MimiConfig
+ base_model_prefix = "mimi"
+ main_input_name = "input_values"
+ supports_gradient_checkpointing = True
+ _no_split_modules = ["MimiDecoderLayer"]
+ _skip_keys_device_placement = "past_key_values"
+ _supports_flash_attn_2 = True
+ _supports_sdpa = True
+ _supports_cache_class = True
+ _supports_static_cache = True
+
+ # Copied from transformers.models.encodec.modeling_encodec.EncodecPreTrainedModel._init_weights
+ def _init_weights(self, module):
+ """Initialize the weights"""
+ if isinstance(module, nn.Linear):
+ module.weight.data.normal_(mean=0.0, std=self.config.initializer_range)
+ if module.bias is not None:
+ module.bias.data.zero_()
+ elif isinstance(module, (nn.LayerNorm, nn.GroupNorm)):
+ module.bias.data.zero_()
+ module.weight.data.fill_(1.0)
+ elif isinstance(module, nn.Conv1d):
+ nn.init.kaiming_normal_(module.weight)
+ if module.bias is not None:
+ k = math.sqrt(module.groups / (module.in_channels * module.kernel_size[0]))
+ nn.init.uniform_(module.bias, a=-k, b=k)
+ elif isinstance(module, nn.Embedding):
+ module.weight.data.normal_(mean=0.0, std=self.config.initializer_range)
+ if module.padding_idx is not None:
+ module.weight.data[module.padding_idx].zero_()
+ elif isinstance(module, nn.LSTM):
+ for name, param in module.named_parameters():
+ if "weight" in name:
+ nn.init.xavier_uniform_(param)
+ elif "bias" in name:
+ nn.init.constant_(param, 0.0)
+
+
+MIMI_START_DOCSTRING = r"""
+ This model inherits from [`PreTrainedModel`]. 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 PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass.
+ Use it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage
+ and behavior.
+
+ Parameters:
+ config ([`MimiConfig`]):
+ 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
+ [`~PreTrainedModel.from_pretrained`] method to load the model weights.
+"""
+
+
+MIMI_INPUTS_DOCSTRING = r"""
+ Args:
+ input_values (`mindspore.Tensor` of shape `(batch_size, channels, sequence_length)`, *optional*):
+ Raw audio input converted to Float.
+ padding_mask (`mindspore.Tensor` of shape `(batch_size, sequence_length)`, *optional*):
+ Indicates which inputs are to be ignored due to padding, where elements are either 1 for *not masked* or 0
+ for *masked*.
+ num_quantizers (`int`, *optional*):
+ Number of quantizers (i.e codebooks) to use. By default, all quantizers are used.
+ audio_codes (`mindspore.LongTensor` of shape `(batch_size, num_quantizers, codes_length)`, *optional*):
+ Discret code embeddings computed using `model.encode`.
+ encoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the encoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ decoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the decoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ return_dict (`bool`, *optional*):
+ Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
+"""
+
+
+@add_start_docstrings(
+ "The Mimi neural audio codec model.",
+ MIMI_START_DOCSTRING,
+)
+class MimiModel(MimiPreTrainedModel):
+ def __init__(self, config: MimiConfig):
+ super().__init__(config)
+ self.config = config
+
+ self.encoder = MimiEncoder(config)
+ self.encoder_transformer = MimiTransformerModel(config)
+
+ self.downsample = None
+ self.upsample = None
+ if config.frame_rate != config.encodec_frame_rate:
+ self.downsample = MimiConv1d(
+ config,
+ config.hidden_size,
+ config.hidden_size,
+ kernel_size=2 * int(config.encodec_frame_rate / config.frame_rate),
+ stride=2,
+ bias=False,
+ pad_mode="replicate",
+ )
+
+ self.upsample = MimiConvTranspose1d(
+ config,
+ config.hidden_size,
+ config.hidden_size,
+ kernel_size=2 * int(config.encodec_frame_rate / config.frame_rate),
+ stride=2,
+ bias=False,
+ groups=config.upsample_groups,
+ )
+
+ self.decoder_transformer = MimiTransformerModel(config)
+ self.decoder = MimiDecoder(config)
+
+ self.quantizer = MimiSplitResidualVectorQuantizer(config)
+
+ self.bits_per_codebook = int(math.log2(self.config.codebook_size))
+ if 2**self.bits_per_codebook != self.config.codebook_size:
+ raise ValueError("The codebook_size must be a power of 2.")
+
+ # Initialize weights and apply final processing
+ self.post_init()
+
+ def get_encoder(self):
+ return self.encoder
+
+ def get_decoder(self):
+ return self.decoder
+
+ def _encode_frame(
+ self,
+ input_values: mindspore.Tensor,
+ num_quantizers: int,
+ padding_mask: int,
+ past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None,
+ return_dict: Optional[bool] = None,
+ ) -> Tuple[mindspore.Tensor, Optional[mindspore.Tensor]]:
+ """
+ Encodes the given input using the underlying VQVAE. The padding mask is required to compute the correct scale.
+ """
+ embeddings = self.encoder(input_values)
+ encoder_outputs = self.encoder_transformer(
+ embeddings.transpose(1, 2), past_key_values=past_key_values, return_dict=return_dict
+ )
+ if return_dict:
+ past_key_values = encoder_outputs.get("past_key_values")
+ elif len(encoder_outputs) > 1:
+ past_key_values = encoder_outputs[1]
+ embeddings = encoder_outputs[0].transpose(1, 2)
+ embeddings = self.downsample(embeddings)
+
+ codes = self.quantizer.encode(embeddings, num_quantizers)
+ codes = codes.transpose(0, 1)
+ return codes, past_key_values
+
+ def encode(
+ self,
+ input_values: mindspore.Tensor,
+ padding_mask: mindspore.Tensor = None,
+ num_quantizers: Optional[float] = None,
+ encoder_past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None,
+ return_dict: Optional[bool] = None,
+ ) -> Union[Tuple[mindspore.Tensor, Optional[mindspore.Tensor]], MimiEncoderOutput]:
+ """
+ Encodes the input audio waveform into discrete codes.
+
+ Args:
+ input_values (`mindspore.Tensor` of shape `(batch_size, channels, sequence_length)`):
+ Float values of the input audio waveform.
+ padding_mask (`mindspore.Tensor` of shape `(batch_size, channels, sequence_length)`):
+ Indicates which inputs are to be ignored due to padding, where elements are either 1 for *not masked* or 0
+ for *masked*.
+ num_quantizers (`int`, *optional*):
+ Number of quantizers (i.e codebooks) to use. By default, all quantizers are used.
+ encoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the encoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ return_dict (`bool`, *optional*):
+ Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
+ Returns:
+ `codebook` of shape `[batch_size, num_codebooks, frames]`, the discrete encoded codes for the input audio waveform.
+ """
+ return_dict = return_dict if return_dict is not None else self.config.return_dict
+
+ num_quantizers = self.config.num_quantizers if num_quantizers is None else num_quantizers
+
+ if num_quantizers > self.config.num_quantizers:
+ raise ValueError(
+ f"The number of quantizers (i.e codebooks) asked should be lower than the total number of quantizers {self.config.num_quantizers}, but is currently {num_quantizers}."
+ )
+
+ _, channels, input_length = input_values.shape
+
+ if channels < 1 or channels > 2:
+ raise ValueError(f"Number of audio channels must be 1 or 2, but got {channels}")
+
+ if padding_mask is None:
+ padding_mask = torch.ones_like(input_values).bool()
+
+ encoded_frames, encoder_past_key_values = self._encode_frame(
+ input_values,
+ num_quantizers,
+ padding_mask.bool(),
+ past_key_values=encoder_past_key_values,
+ return_dict=return_dict,
+ )
+
+ if not return_dict:
+ return (
+ encoded_frames,
+ encoder_past_key_values,
+ )
+
+ return MimiEncoderOutput(encoded_frames, encoder_past_key_values)
+
+ def _decode_frame(
+ self,
+ codes: mindspore.Tensor,
+ past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None,
+ return_dict: Optional[bool] = None,
+ ) -> mindspore.Tensor:
+ embeddings = self.quantizer.decode(codes)
+
+ embeddings = self.upsample(embeddings)
+ decoder_outputs = self.decoder_transformer(
+ embeddings.transpose(1, 2), past_key_values=past_key_values, return_dict=return_dict
+ )
+ if return_dict:
+ past_key_values = decoder_outputs.get("past_key_values")
+ elif len(decoder_outputs) > 1:
+ past_key_values = decoder_outputs[1]
+ embeddings = decoder_outputs[0].transpose(1, 2)
+ outputs = self.decoder(embeddings)
+ return outputs, past_key_values
+
+ def decode(
+ self,
+ audio_codes: mindspore.Tensor,
+ padding_mask: Optional[mindspore.Tensor] = None,
+ decoder_past_key_values: Optional[Union[Cache, List[mindspore.Tensor]]] = None,
+ return_dict: Optional[bool] = None,
+ ) -> Union[Tuple[mindspore.Tensor, mindspore.Tensor], MimiDecoderOutput]:
+ """
+ Decodes the given frames into an output audio waveform.
+
+ Note that the output might be a bit bigger than the input. In that case, any extra steps at the end can be
+ trimmed.
+
+ Args:
+ audio_codes (`mindspore.LongTensor` of shape `(batch_size, num_quantizers, codes_length)`, *optional*):
+ Discret code embeddings computed using `model.encode`.
+ padding_mask (`mindspore.Tensor` of shape `(batch_size, channels, sequence_length)`):
+ Indicates which inputs are to be ignored due to padding, where elements are either 1 for *not masked* or 0
+ for *masked*.
+ decoder_past_key_values (`Cache`, *optional*):
+ Pre-computed hidden-states (key and values in the self-attention blocks) that can be used to speed up sequential decoding of the decoder transformer.
+ This typically consists in the `past_key_values` returned by the model at a previous stage of decoding, when `use_cache=True` or `config.use_cache=True`.
+
+ The model will output the same cache format that is fed as input.
+
+ If `past_key_values` are used, the user can optionally input only the last `audio_values` or `audio_codes (those that don't
+ have their past key value states given to this model).
+ return_dict (`bool`, *optional*):
+ Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
+
+ """
+ return_dict = return_dict if return_dict is not None else self.config.return_dict
+
+ audio_values, decoder_past_key_values = self._decode_frame(
+ audio_codes, past_key_values=decoder_past_key_values, return_dict=return_dict
+ )
+
+ # truncate based on padding mask
+ if padding_mask is not None and padding_mask.shape[-1] < audio_values.shape[-1]:
+ audio_values = audio_values[..., : padding_mask.shape[-1]]
+
+ if not return_dict:
+ return (
+ audio_values,
+ decoder_past_key_values,
+ )
+ return MimiDecoderOutput(audio_values, decoder_past_key_values)
+
+ @add_start_docstrings_to_model_forward(MIMI_INPUTS_DOCSTRING)
+ @replace_return_docstrings(output_type=MimiOutput, config_class=_CONFIG_FOR_DOC)
+ def construct(self, input_values, padding_mask=None, num_quantizers=None, audio_codes=None,
+ encoder_past_key_values=None, decoder_past_key_values=None, return_dict=None):
+ # Convert PyTorch forward() to MindSpore construct()
+ # ... implementation ...
+ return MimiOutput(
+ audio_codes=audio_codes,
+ audio_values=audio_values,
+ encoder_past_key_values=encoder_past_key_values,
+ decoder_past_key_values=decoder_past_key_values,
+ )
+
+
+__all__ = ["MimiModel", "MimiPreTrainedModel"]
+
+# Import necessary libraries from MindNLP
+from mindnlp import Tokenizer, SimpleModel
+
+# Initialize a tokenizer
+tokenizer = Tokenizer()
+
+# Define a simple NLP model
+model = SimpleModel()
+
+# Example text
+test_text = "Hello, this is a test sentence for MindNLP."
+
+# Tokenize the text
+tokens = tokenizer.tokenize(test_text)
+
+# Print the tokens
+print("Tokens:", tokens)
+
+# Run the model on the tokens
+output = model(tokens)
+
+# Print the model's output
+print("Model output:", output)