o %ݫi9@sdZddlZddlmZddlZddlmZddlZddl m Z ddl m Z ddl mZddlmZdd lmZdd lmZGd d d ejZGd ddejZGdddejZGdddejZGdddejZGdddejZGdddejZddZddZdddZdS) zTransformer implementation in the SpeechBrain style. Authors * Jianyuan Zhong 2020 * Samuele Cornell 2021 * Shucong Zhang 2024 N)Optional)Swish) RelPosEncXL)Conv1d)map_old_state_dict_weights)BranchformerEncoder)ConformerEncoderc seZdZdZddddddejdddd d d d eejd d ddddddejdddfde e de e de e de ej de ej de e de e de e de e de e de e de e de e de ej de e ffd d! Zd"d#ZZS)$TransformerInterfaceuDThis is an interface for transformer model. Users can modify the attributes and define the forward function as needed according to their own tasks. The architecture is based on the paper "Attention Is All You Need": https://arxiv.org/pdf/1706.03762.pdf Arguments --------- d_model: int The number of expected features in the encoder/decoder inputs (default=512). nhead: int The number of heads in the multi-head attention models (default=8). num_encoder_layers: int, optional The number of encoder layers in1ì the encoder. num_decoder_layers: int, optional The number of decoder layers in the decoder. d_ffn: int, optional The dimension of the feedforward network model hidden layer. dropout: int, optional The dropout value. activation: torch.nn.Module, optional The activation function for Feed-Forward Network layer, e.g., relu or gelu or swish. custom_src_module: torch.nn.Module, optional Module that processes the src features to expected feature dim. custom_tgt_module: torch.nn.Module, optional Module that processes the src features to expected feature dim. positional_encoding: str, optional Type of positional encoding used. e.g. 'fixed_abs_sine' for fixed absolute positional encodings. normalize_before: bool, optional Whether normalization should be applied before or after MHA or FFN in Transformer layers. Defaults to True as this was shown to lead to better performance and training stability. kernel_size: int, optional Kernel size in convolutional layers when Conformer is used. bias: bool, optional Whether to use bias in Conformer convolutional layers. encoder_module: str, optional Choose between Branchformer, Conformer and Transformer for the encoder. The decoder is fixed to be a Transformer. conformer_activation: torch.nn.Module, optional Activation module used after Conformer convolutional layers. E.g. Swish, ReLU etc. it has to be a torch Module. branchformer_activation: torch.nn.Module, optional Activation module used within the Branchformer Encoder. E.g. Swish, ReLU etc. it has to be a torch Module. attention_type: str, optional Type of attention layer used in all Transformer or Conformer layers. e.g. regularMHA or RelPosMHA. max_length: int, optional Max length for the target and source sequence in input. Used for positional encodings. causal: bool, optional Whether the encoder should be causal or not (the decoder is always causal). If causal the Conformer convolutional layer is causal. encoder_kdim: int, optional Dimension of the key for the encoder. encoder_vdim: int, optional Dimension of the value for the encoder. decoder_kdim: int, optional Dimension of the key for the decoder. decoder_vdim: int, optional Dimension of the value for the decoder. csgu_linear_units: int, optional Number of neurons in the hidden linear units of the CSGU Module. -> Branchformer gate_activation: torch.nn.Module, optional Activation function used at the gate of the CSGU module. -> Branchformer use_linear_after_conv: bool, optional If True, will apply a linear transformation of size input_size//2. -> Branchformer output_hidden_states: bool, optional Whether the model should output the hidden states as a list of tensor. layerdrop_prob: float The probability to drop an entire layer. iig?Nfixed_abs_sineT transformer regularMHA Fi kernel_sizebiasencoder_moduleconformer_activationbranchformer_activationattention_type max_lengthcausal encoder_kdim encoder_vdim decoder_kdim decoder_vdimcsgu_linear_unitsgate_activationuse_linear_after_convcst||_||_| |_||_||_||_||_||_ ||_ |dvs&J| dvs,J||dks6Jd| dkrAt |||_ n| durF |dkrUt ||_ t |||_|dkr_t |||_|dkr|durl|||_|dkrt||||||| |j|j|j|j|j |j d |_n@|d krt||||||| | |j|j|j |j d |_| sJd |dusJd n|dkrt|||||| |j||||j |j d |_|dkr| dur| ||_t||||||| dd|j|jd |_dSdS)N)r RelPosMHAXL hypermixingRoPEMHA)r NrzGnumber of encoder layers and number of decoder layers cannot both be 0!r r"r$r) nhead num_layersd_ffnd_modeldropout activationnormalize_beforerrkdimvdimoutput_hidden_stateslayerdrop_prob conformer) r%r&r'r(r)r*rrrrr.r/z+normalize_before must be True for Conformerz%conformer_activation must not be None branchformer) r%r&r(r)r*rrrr r!r.r/Tr) r&r%r'r(r)r*r+rrr,r-)super__init__rrpositional_encoding_typerrrrr.r/PositionalEncodingpositional_encodingrpositional_encoding_decodercustom_src_moduleTransformerEncoderencoderr rcustom_tgt_moduleTransformerDecoderdecoder)selfr(r%num_encoder_layersnum_decoder_layersr'r)r*r8r;r6r+rrrrrrrrrrrrrr r!r.r/ __class__d/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/lobes/models/transformer/Transformer.pyr3cs         zTransformerInterface.__init__cKst)z?Users should modify this function according to their own tasks.)NotImplementedError)r>kwagsrCrCrDforwardszTransformerInterface.forward)__name__ __module__ __qualname____doc__nnReLUrGELUIdentityrintboolstrModuler3rG __classcell__rCrCrArDr s~L r cs*eZdZdZdfdd ZddZZS)r5aThis class implements the absolute sinusoidal positional encoding function. PE(pos, 2i) = sin(pos/(10000^(2i/dmodel))) PE(pos, 2i+1) = cos(pos/(10000^(2i/dmodel))) Arguments --------- input_size: int Embedding dimension. max_len : int, optional Max length of the input sequences (default 2500). Example ------- >>> a = torch.rand((8, 120, 512)) >>> enc = PositionalEncoding(input_size=a.shape[-1]) >>> b = enc(a) >>> b.shape torch.Size([1, 120, 512]) rcst|ddkrtd|d||_tj|j|dd}td|jd}t td|dt d| }t |||dddddf<t |||dddddf<|d}|d |dS) NrzGCannot use sin/cos positional encoding with odd channels (got channels=)F) requires_gradrg@pe)r2r3 ValueErrormax_lentorchzerosarange unsqueezefloatexpmathlogsincosregister_buffer)r> input_sizerZrX positions denominatorrArCrDr3s"      zPositionalEncoding.__init__cCs$|jddd|dfS)z Arguments --------- x : torch.Tensor Input feature shape (batch, time, fea) Returns ------- The positional encoding. Nr)rXsizeclonedetachr>xrCrCrDrG$s$ zPositionalEncoding.forward)rrHrIrJrKr3rGrTrCrCrArDr5sr5c sjeZdZdZdddejdddddgdf fdd Z   dd eej d eej d eej fd dZ Z S)TransformerEncoderLayeraMThis is an implementation of self-attention encoder layer. Arguments --------- d_ffn: int, optional The dimension of the feedforward network model hidden layer. nhead: int The number of heads in the multi-head attention models (default=8). d_model: int The number of expected features in the encoder/decoder inputs (default=512). kdim: int, optional Dimension of the key. vdim: int, optional Dimension of the value. dropout: int, optional The dropout value. activation: torch.nn.Module, optional The activation function for Feed-Forward Network layer, e.g., relu or gelu or swish. normalize_before: bool, optional Whether normalization should be applied before or after MHA or FFN in Transformer layers. Defaults to True as this was shown to lead to better performance and training stability. attention_type: str, optional Type of attention layer used in all Transformer or Conformer layers. e.g. regularMHA or RelPosMHA. ffn_type: str type of ffn: regularFFN/1dcnn ffn_cnn_kernel_size_list: list of int kernel size of 2 1d-convs if ffn_type is 1dcnn causal: bool, optional Whether the encoder should be causal or not (the decoder is always causal). If causal the Conformer convolutional layer is causal. Example ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> net = TransformerEncoderLayer(512, 8, d_model=512) >>> output = net(x) >>> output[0].shape torch.Size([8, 60, 512]) NrFr regularFFNc sRt| dkrtjjj|||||d|_n1| dkr(tjjj|||| d|_n | dkr:tjjj ||d|dd|_n| dkrHtjj ||||_| d krYtjjj ||||d |_ n&| d krt t||| d | rid nddt t||| d| ryd ndd|_ tjjj|dd|_tjjj|dd|_tj ||_tj ||_||_| |_dS)Nr)r%r(r)r,r-r"mask_pos_futurer#F)input_output_dim hypernet_sizetied num_headsfix_tm_hidden_sizer$rpr'rfr)r*1dcnnrrsame) in_channels out_channelsrpaddingrư>eps)r2r3sbnnet attentionMultiheadAttentionself_attr"r# HyperMixingr$PositionalwiseFeedForwardpos_ffnrL SequentialrrM normalization LayerNormnorm1norm2r[Dropoutdropout1dropout2r+ pos_ffn_type) r>r'r%r(r,r-r)r*r+rffn_typeffn_cnn_kernel_size_listrrArCrDr3^sp        z TransformerEncoderLayer.__init__src_masksrc_key_padding_maskpos_embscCs|jr ||}n|}|j||||||d\}}|||}|js'||}|jr0||}n|}||}|||}|jsF||}||fS)a' Arguments --------- src : torch.Tensor The sequence to the encoder layer. src_mask : torch.Tensor The mask for the src query for each example in the batch. src_key_padding_mask : torch.Tensor, optional The mask for the src keys for each example in the batch. pos_embs: torch.Tensor, optional The positional embeddings tensor. Returns ------- output : torch.Tensor The output of the transformer encoder layer. ) attn_maskkey_padding_maskr)r+rrrrrr)r>srcrrrsrc1output self_attnrCrCrDrGs,       zTransformerEncoderLayer.forward)NNN rHrIrJrKrLrMr3rr[TensorrGrTrCrCrArDro2s,0Proc steZdZdZdddddejdddddddgdf fdd Z    dd eej d eej d eej fd dZ Z S)r9aGThis class implements the transformer encoder. Arguments --------- num_layers : int Number of transformer layers to include. nhead : int Number of attention heads. d_ffn : int Hidden size of self-attention Feed Forward layer. input_shape : tuple Expected shape of the input. d_model : int The dimension of the input embedding. kdim : int Dimension for key (Optional). vdim : int Dimension for value (Optional). dropout : float Dropout for the encoder (Optional). activation: torch.nn.Module, optional The activation function for Feed-Forward Network layer, e.g., relu or gelu or swish. normalize_before: bool, optional Whether normalization should be applied before or after MHA or FFN in Transformer layers. Defaults to True as this was shown to lead to better performance and training stability. causal: bool, optional Whether the encoder should be causal or not (the decoder is always causal). If causal the Conformer convolutional layer is causal. layerdrop_prob: float The probability to drop an entire layer attention_type: str, optional Type of attention layer used in all Transformer or Conformer layers. e.g. regularMHA or RelPosMHA. ffn_type: str type of ffn: regularFFN/1dcnn ffn_cnn_kernel_size_list: list of int conv kernel size of 2 1d-convs if ffn_type is 1dcnn output_hidden_states: bool, optional Whether the model should output the hidden states as a list of tensor. Example ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> net = TransformerEncoder(1, 8, 512, d_model=512) >>> output, _ = net(x) >>> output.shape torch.Size([8, 60, 512]) >>> import torch >>> x = torch.rand((8, 60, 512)) >>> net = TransformerEncoder(1, 8, 512, d_model=512, output_hidden_states=True) >>> output, attn_list, hidden_list = net(x) >>> hidden_list[0].shape torch.Size([8, 60, 512]) >>> len(hidden_list) 2 NrFrrprqcsdttj f ddt|D|_tjj j dd|_ | |_ ||_ dS)Ncs,g|]}t   d qS)) r'r%r(r,r-r)r*r+rrrr)ro).0i r*rrr'r(r)rrr,r%r+r-rCrD 7s"z/TransformerEncoder.__init__..rr)r2r3r[rL ModuleListrangelayersrrrrnormr/r.)r>r&r%r' input_shaper(r,r-r)r*r+rr/rrrr.rArrDr3!s   zTransformerEncoder.__init__rrrc Cs|dusJd|}|jdkrtt|j}g}|jr|g} t|jD]*\} } |jr7|jdks7|| |jkrN| ||||d\}} || |jrN| |q$| |}|jr\||| fS||fS)a, Arguments --------- src : torch.Tensor The sequence to the encoder layer (required). src_mask : torch.Tensor The mask for the src sequence (optional). src_key_padding_mask : torch.Tensor The mask for the src keys per batch (optional). pos_embs : torch.Tensor The positional embedding tensor dynchunktrain_config : config Not supported for this encoder. Returns ------- output : torch.Tensor The output of the transformer. attention_lst : list The attention values. hidden_state_lst : list, optional The output of the hidden layers of the encoder. Only works if output_hidden_states is set to true. Nz3Dynamic Chunk Training unsupported for this encoderr)rrr) r/r[randlenrr. enumeratetrainingappendr) r>rrrrdynchunktrain_configr keep_probs attention_lsthidden_state_lstr enc_layerrrCrCrDrGMs8 !      zTransformerEncoder.forward)NNNNrrCrCrArDr9s6A/r9csTeZdZdZdddejdddffdd Z      d dd Zfd d ZZ S) TransformerDecoderLayeraThis class implements the self-attention decoder layer. Arguments --------- d_ffn : int Hidden size of self-attention Feed Forward layer. nhead : int Number of attention heads. d_model : int Dimension of the model. kdim : int Dimension for key (optional). vdim : int Dimension for value (optional). dropout : float Dropout for the decoder (optional). activation : Callable Function to use between layers, default nn.ReLU normalize_before : bool Whether to normalize before layers. attention_type : str Type of attention to use, "regularMHA" or "RelPosMHAXL" causal : bool Whether to mask future positions. Example ------- >>> src = torch.rand((8, 60, 512)) >>> tgt = torch.rand((8, 60, 512)) >>> net = TransformerDecoderLayer(1024, 8, d_model=512) >>> output, self_attn, multihead_attn = net(src, tgt) >>> output.shape torch.Size([8, 60, 512]) NrFrc st||_| dkr'tjjj|||||d|_tjjj|||||d|_n| dkrCtjjj |||| d|_tjjj |||| d|_tjjj ||||d|_ tjj j |dd|_tjj j |dd|_tjj j |dd|_tj||_tj||_tj||_||_dS)Nr)r%r(r,r-r)r"rrryrr)r2r3r%rrrrrmultihead_attnr"rrrrrrnorm3r[rLrrrdropout3r+) r>r'r%r(r,r-r)r*r+rrrArCrDr3sJ   z TransformerDecoderLayer.__init__c Cs|jr ||} n|} |j| | | |||d\} } ||| }|js'||}|jr0||} n|} |j| |||||d\} } ||| }|jsN||}|jrW||} n|} || } || | }|jsm||}|| | fS)a3 Arguments ---------- tgt: torch.Tensor The sequence to the decoder layer (required). memory: torch.Tensor The sequence from the last layer of the encoder (required). tgt_mask: torch.Tensor The mask for the tgt sequence (optional). memory_mask: torch.Tensor The mask for the memory sequence (optional). tgt_key_padding_mask: torch.Tensor The mask for the tgt keys per batch (optional). memory_key_padding_mask: torch.Tensor The mask for the memory keys per batch (optional). pos_embs_tgt: torch.Tensor The positional embeddings for the target (optional). pos_embs_src: torch.Tensor The positional embeddings for the source (optional). )querykeyvaluerrr) r+rrrrrrrrr) r>tgtmemorytgt_mask memory_masktgt_key_padding_maskmemory_key_padding_mask pos_embs_tgt pos_embs_srctgt1tgt2rmultihead_attentionrCrCrDrGsH            zTransformerDecoderLayer.forwardcs2ddi}t||}tj||g|Ri|dS)zFLoad the model from a state_dict and map the old keys to the new keys.mutihead_attentionrN)rr2_load_from_state_dict)r> state_dictprefixargskwargsmappingrArCrDrBs  z-TransformerDecoderLayer._load_from_state_dictNNNNNN) rHrIrJrKrLrMr3rGrrTrCrCrArDrs$(< UrcsHeZdZdZdddejdddffdd Z      d dd ZZS) r<aPThis class implements the Transformer decoder. Arguments --------- num_layers : int Number of transformer layers for the decoder. nhead : int Number of attention heads. d_ffn : int Hidden size of self-attention Feed Forward layer. d_model : int Dimension of the model. kdim : int, optional Dimension for key (Optional). vdim : int, optional Dimension for value (Optional). dropout : float, optional Dropout for the decoder (Optional). activation : Callable The function to apply between layers, default nn.ReLU normalize_before : bool Whether to normalize before layers. causal : bool Whether to allow future information in decoding. attention_type : str Type of attention to use, "regularMHA" or "RelPosMHAXL" Example ------- >>> src = torch.rand((8, 60, 512)) >>> tgt = torch.rand((8, 60, 512)) >>> net = TransformerDecoder(1, 8, 1024, d_model=512) >>> output, _, _ = net(src, tgt) >>> output.shape torch.Size([8, 60, 512]) NrFrc sTttj f ddt|D|_tjj j dd|_ dS)Ncs(g|]}t d qS)) r'r%r(r,r-r)r*r+rr)r)r_ r*rrr'r(r)r,r%r+r-rCrDrs z/TransformerDecoder.__init__..rr) r2r3r[rLrrrrrrrr) r>r&r%r'r(r,r-r)r*r+rrrArrDr3os  zTransformerDecoder.__init__c  Cs`|} gg} } |jD]} | | |||||||d\} } }| | | |q || } | | | fS)a; Arguments ---------- tgt : torch.Tensor The sequence to the decoder layer (required). memory : torch.Tensor The sequence from the last layer of the encoder (required). tgt_mask : torch.Tensor The mask for the tgt sequence (optional). memory_mask : torch.Tensor The mask for the memory sequence (optional). tgt_key_padding_mask : torch.Tensor The mask for the tgt keys per batch (optional). memory_key_padding_mask : torch.Tensor The mask for the memory keys per batch (optional). pos_embs_tgt : torch.Tensor The positional embeddings for the target (optional). pos_embs_src : torch.Tensor The positional embeddings for the source (optional). )rrrrrr)rrr)r>rrrrrrrrr self_attnsmultihead_attns dec_layerrrrCrCrDrGs"     zTransformerDecoder.forwardr) rHrIrJrKrLrMr3rGrTrCrCrArDr<Is"+&r<cs(eZdZdZfddZddZZS)NormalizedEmbeddingauThis class implements the normalized embedding layer for the transformer. Since the dot product of the self-attention is always normalized by sqrt(d_model) and the final linear projection for prediction shares weight with the embedding layer, we multiply the output of the embedding by sqrt(d_model). Arguments --------- d_model: int The number of expected features in the encoder/decoder inputs (default=512). vocab: int The vocab size. Example ------- >>> emb = NormalizedEmbedding(512, 1000) >>> trg = torch.randint(0, 999, (8, 50)) >>> emb_fea = emb(trg) cs*ttjjj||dd|_||_dS)Nr)num_embeddings embedding_dimblank_id)r2r3rr embedding Embeddingembr()r>r(vocabrArCrDr3s  zNormalizedEmbedding.__init__cCs||t|jS)z:Processes the input tensor x and returns an output tensor.)rrasqrtr(rlrCrCrDrGszNormalizedEmbedding.forwardrnrCrCrArDrs rcCsrt|jdkr|j\}}}}|||||}|||j}t|jdkr5|jdd}| S| S)aiCreates a binary mask to prevent attention to padded locations. We suggest using ``get_mask_from_lengths`` instead of this function. Arguments --------- padded_input: torch.Tensor Padded input. pad_idx: int idx for padding element. Returns ------- key_padded_mask: torch.Tensor Binary mask to prevent attention to padding. Example ------- >>> a = torch.LongTensor([[1,1,0], [2,3,0], [4,5,0]]) >>> get_key_padding_mask(a, pad_idx=0) tensor([[False, False, True], [False, False, True], [False, False, True]]) rU)dim) rshapereshapeeqtodevicer_prodrQrk) padded_inputpad_idxbztimech1ch2key_padded_maskrCrCrDget_key_padding_masksrcCsh|jd}ttj||f|jddkdd}||dktd|dktd}| |jS)aCreates a binary mask for each sequence which masks future frames. Arguments --------- padded_input: torch.Tensor Padded input tensor. Returns ------- mask : torch.Tensor Binary mask for masking future frames. Example ------- >>> a = torch.LongTensor([[1,1,0], [2,3,0], [4,5,0]]) >>> get_lookahead_mask(a) tensor([[0., -inf, -inf], [0., 0., -inf], [0., 0., 0.]]) r)rrz-infr) rr[triuonesr transposer_ masked_fillrkr)rseq_lenmaskrCrCrDget_lookahead_mask s rcCs@|dur t|}tj||j|jd}|d|dkS)asCreates a binary mask from sequence lengths Arguments --------- lengths: torch.Tensor A tensor of sequence lengths max_len: int (Optional) Maximum sequence length, defaults to None. Returns ------- mask: torch.Tensor the mask where padded elements are set to True. Then one can use tensor.masked_fill_(mask, 0) for the masking. Example ------- >>> lengths = torch.tensor([3, 2, 4]) >>> get_mask_from_lengths(lengths) tensor([[False, False, False, True], [False, False, True, True], [False, False, False, False]]) N)rdtyperr)r[maxitemr]rrr^)lengthsrZ seq_rangerCrCrDget_mask_from_lengths-s  r)N) rKratypingrr[torch.nnrL speechbrainrspeechbrain.nnet.activationsrspeechbrain.nnet.attentionrspeechbrain.nnet.CNNrspeechbrain.utils.checkpointsr Branchformerr Conformerr rSr r5ror9rr<rrrrrCrCrCrDs4        e63.9{ '"