o %ݫiFd @s"dZddlmZddlmZmZddlZddlmZddlm Z ddl m Z m Z m Z mZddlmZdd lmZdd lmZdd lmZdd lmZeeZeGd ddZ  ddejdedeedeejfddZ     ddedeefddZGddde Z Gdddej!Z"dS)zTransformer for ASR in the SpeechBrain style. Authors * Jianyuan Zhong 2020 * Titouan Parcollet 2024 * Luca Della Libera 2024 * Shucong Zhang 2024 ) dataclass)AnyOptionalN)nn)length_to_mask)NormalizedEmbeddingTransformerInterfaceget_key_padding_maskget_lookahead_mask)Swish) ModuleList)Linear)DynChunkTrainConfig) get_loggerc@s$eZdZUdZeed< eed<dS)TransformerASRStreamingContextz=Streaming metadata and state for a `TransformerASR` instance.dynchunktrain_configencoder_contextN)__name__ __module__ __qualname____doc__r__annotations__rrrg/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/lobes/models/transformer/TransformerASR.pyr s rFsrccausalrreturnc Cs|r |dusJt|S|durdS|d}|j}||}tj||jd}tj|||d||jd|d|}|d|dddfk}|sd|j} ||| d8}||d|dddfk7}|S)a2Prepare the source transformer mask that restricts which frames can attend to which frames depending on causal or other simple restricted attention methods. Arguments --------- src: torch.Tensor The source tensor to build a mask from. The contents of the tensor are not actually used currently; only its shape and other metadata (e.g. device). causal: bool Whether strict causality shall be used. Frames will not be able to attend to any future frame. dynchunktrain_config: DynChunkTrainConfig, optional Dynamic Chunk Training configuration. This implements a simple form of chunkwise attention. Incompatible with `causal`. Returns ------- torch.Tensor A boolean mask Tensor of shape (timesteps, timesteps). N)device) r size chunk_sizetorcharangerrepeat_interleaveis_infinite_left_contextleft_context_size) rrr timestepsr! num_chunks timestep_idxmask_idxsrc_masknum_left_chunksrrrmake_transformer_src_mask/s(   r-c Csnd}|durt||jd}t|}t|||d}|dur-t||d} t|} nd} d} || || fS)aThis function generates masks for training the transformer model, opinionated for an ASR context with encoding masks and, optionally, decoding masks (if specifying `tgt`). Arguments --------- src : torch.Tensor The sequence to the encoder (required). tgt : torch.Tensor The sequence to the decoder. wav_len : torch.Tensor The lengths of the inputs. pad_idx : int The index for token (default=0). causal: bool Whether strict causality shall be used. See `make_asr_src_mask` dynchunktrain_config: DynChunkTrainConfig, optional Dynamic Chunk Training configuration. See `make_asr_src_mask` Returns ------- src_key_padding_mask : torch.Tensor Key padding mask for ignoring padding tgt_key_padding_mask : torch.Tensor Key padding mask for ignoring padding src_mask : torch.Tensor Mask for ignoring invalid (e.g. future) timesteps tgt_mask : torch.Tensor Mask for ignoring invalid (e.g. future) timesteps Nr)rr)pad_idx)r"roundshaperboolr-r r ) rtgtwav_lenr.rrsrc_key_padding_maskabs_lenr+tgt_key_padding_masktgt_maskrrrmake_transformer_src_tgt_masksjs&   r8cseZdZdZddddddejddd d d eejd d ddejdddfde e de e de e de ej de ej de e de e de e de e de ej de e ffdd Zd-dd Zed.d!d"Z   d/d#e efd$d%Zd&efd'd(Zifd#efd)d*Zd+d,ZZS)0TransformerASRa This is an implementation of transformer model for ASR. The architecture is based on the paper "Attention Is All You Need": https://arxiv.org/pdf/1706.03762.pdf Arguments --------- tgt_vocab: int Size of vocabulary. input_size: int Input feature size. d_model : int, optional Embedding dimension size. (default=512). nhead : int, optional The number of heads in the multi-head attention models (default=8). num_encoder_layers : int, optional The number of sub-encoder-layers in the encoder (default=6). num_decoder_layers : int, optional The number of sub-decoder-layers in the decoder (default=6). d_ffn : int, optional The dimension of the feedforward network model (default=2048). dropout : int, optional The dropout value (default=0.1). activation : torch.nn.Module, optional The activation function of FFN layers. Recommended: relu or gelu (default=relu). 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 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. 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. Example ------- >>> src = torch.rand([8, 120, 512]) >>> tgt = torch.randint(0, 720, [8, 120]) >>> net = TransformerASR( ... 720, 512, 512, 8, 1, 1, 1024, activation=torch.nn.GELU ... ) >>> enc_out, dec_out = net.forward(src, tgt) >>> enc_out.shape torch.Size([8, 120, 512]) >>> dec_out.shape torch.Size([8, 120, 512]) iig?fixed_abs_sineFT transformer regularMHAi Ni g kernel_sizebiasencoder_moduleconformer_activationbranchformer_activationattention_type max_lengthrcsgu_linear_unitsgate_activationuse_linear_after_convcs|dur tdd}tjdid|d|d|d|d|d|d | d | d | d | d | d|d|d|d|d|d|d|d|d|d|d|tt||dddtj||_ |dkrptt |||_ | dS)Na`causal` not specified for `TransformerASR`, assuming `True` for compatibility. We strongly recommend that you explicitly set this. If you are using a model or recipe defined before v1.0, it might now be BROKEN! If so, please see https://github.com/speechbrain/speechbrain/issues/2604Td_modelnheadnum_encoder_layersnum_decoder_layersd_ffndropout activationpositional_encodingnormalize_beforer@rArBrCrDrErFrrGrHrIoutput_hidden_stateslayerdrop_probF) input_size n_neuronsrA combine_dimsrr) loggerwarningsuper__init__r r r"rDropoutcustom_src_modulercustom_tgt_module _init_params)self tgt_vocabrUrJrKrLrMrNrOrPrQrRr@rArBrCrDrErFrrGrHrIrSrT __class__rrr[s         zTransformerASR.__init__rc Csl|jdkr|j\}}}}|||||}t||||j|d\} } } } ||}|jdks2|jdkr5d} n|jdkr@||} n|jdkrN|||}d} |j || | | d}|dur]|S|j rf|\}}}n|\}}| |}|jdksy|jdkr|| |}d} d}n|jdks|jdkr|||}d}d} |j ||d| | | || d \}}}|j r|||fS||fS) a Arguments --------- src : torch.Tensor The sequence to the encoder. tgt : torch.Tensor The sequence to the decoder. wav_len: torch.Tensor, optional Torch Tensor of shape (batch, ) containing the relative length to padded length for each example. pad_idx : int, optional The index for token (default=0). Returns ------- encoder_out : torch.Tensor The output of the encoder. decoder_out : torch.Tensor The output of the decoder hidden_state_lst : list, optional The output of the hidden layers of the encoder. Only works if output_hidden_states is set to true. )rr. hypermixingRoPEMHAN RelPosMHAXLr<)rr+r4pos_embs)r2memory memory_maskr7r6memory_key_padding_mask pos_embs_tgt pos_embs_src)ndimr0reshaper8rr]rErQpositional_encoding_typeencoderrSr^positional_encoding_decoderdecoder)r`rr2r3r.bztch1ch2r4r6r+r7pos_embs_encoderoutputs encoder_out_ hidden_statespos_embs_target decoder_outrrrforwardFsn                zTransformerASR.forwardc Cst|}d}|durdt|}||}|jdks!|jdkr-|||}d}d}n|jdks7|jdkrB|||}d}d}|j||||||d\}} } || dfS) aThis method implements a decoding step for the transformer model. Arguments --------- tgt : torch.Tensor The sequence to the decoder. encoder_out : torch.Tensor Hidden output of the encoder. enc_len : torch.LongTensor The actual length of encoder states. Returns ------- prediction Nrrgrfr<re)r7rkrlrm) r rr1r^rErrrprQrs) r`r2rzenc_lenr7r4rxr} prediction self_attnsmultihead_attnsrrrdecodes0      zTransformerASR.decodercCs|dkr|j\}}}}|||||}t|d|||j|d\} } } } ||}|jdks4|jdkr7d} n|jdkrB||} n|jdkrP|||}d} |j || | | |d} |j rf| \}} }||fS| \}} |S) a Encoder forward pass Arguments --------- src : torch.Tensor The sequence to the encoder. wav_len : torch.Tensor, optional Torch Tensor of shape (batch, ) containing the relative length to padded length for each example. pad_idx : int The index used for padding. dynchunktrain_config : DynChunkTrainConfig Dynamic chunking config. Returns ------- encoder_out : torch.Tensor rdN)r.rrrerfrgr<)rr+r4rhr) dimr0ror8rr]rErQrprqrS)r`rr3r.rrtrurvrwr4r{r+pos_embs_sourceryrzr|rrrencodesJ       zTransformerASR.encodecontextc Cs|dkr|j\}}}}|||||}|jjdj}|dur$|}nt|j} | d|jd7<tj| d |}| |}|j dkrM| |} n|j dkrUd} n|j dkrc|| |}d} |jj|| |jd \} } | S) ap Streaming encoder forward pass Arguments --------- src : torch.Tensor The sequence (chunk) to the encoder. context : TransformerASRStreamingContext Mutable reference to the streaming context. This holds the state needed to persist across chunk inferences and can be built using `make_streaming_context`. This will get mutated by this function. Returns ------- Encoder output for this chunk. Example ------- >>> import torch >>> from speechbrain.lobes.models.transformer.TransformerASR import TransformerASR >>> from speechbrain.utils.dynamic_chunk_training import DynChunkTrainConfig >>> net = TransformerASR( ... tgt_vocab=100, ... input_size=64, ... d_model=64, ... nhead=8, ... num_encoder_layers=1, ... num_decoder_layers=0, ... d_ffn=128, ... attention_type="RelPosMHAXL", ... positional_encoding=None, ... encoder_module="conformer", ... normalize_before=True, ... causal=False, ... ) >>> ctx = net.make_streaming_context(DynChunkTrainConfig(16, 1)) >>> src1 = torch.rand([8, 16, 64]) >>> src2 = torch.rand([8, 16, 64]) >>> out1 = net.encode_streaming(src1, ctx) >>> out1.shape torch.Size([8, 16, 64]) >>> ctx.encoder_context.layers[0].mha_left_context.shape torch.Size([8, 16, 64]) >>> out2 = net.encode_streaming(src2, ctx) >>> out2.shape torch.Size([8, 16, 64]) >>> ctx.encoder_context.layers[0].mha_left_context.shape torch.Size([8, 16, 64]) >>> combined_out = torch.concat((out1, out2), dim=1) >>> combined_out.shape torch.Size([8, 32, 64]) rdrN)r rgrfr<)rrhr)rr0rorlayersmha_left_contextlistr"emptytor]rErQrprqforward_streaming) r`rrrtrurvrwknown_left_contextpos_encoding_dummy target_shaperrzr{rrrencode_streaming"s* 6       zTransformerASR.encode_streamingcCst||jj|fi|dS)aCreates a blank streaming context for this transformer and its encoder. Arguments --------- dynchunktrain_config : DynChunkTrainConfig Runtime chunkwise attention configuration. encoder_kwargs : dict Parameters to be forward to the encoder's `make_streaming_context`. Metadata required for the encoder could differ depending on the encoder. Returns ------- TransformerASRStreamingContext )rr)rrqmake_streaming_context)r`rencoder_kwargsrrrrsz%TransformerASR.make_streaming_contextcCs,|D]}|dkrtjj|qdS)Nr) parametersrr"rinitxavier_normal_)r`prrrr_s  zTransformerASR._init_paramsNrN)NrN)rrrrrReLUr GELUIdentityrintr1strModuler[rr"no_gradrrrrrrr_ __classcell__rrrbrr9s|S  Od 3 G` r9cs:eZdZdZfddZd ddZdd Zd d ZZS) EncoderWrapperaRThis is a wrapper of any ASR transformer encoder. By default, the TransformerASR .forward() function encodes and decodes. With this wrapper the .forward() function becomes .encode() only. Important: The TransformerASR class must contain a .encode() function. Arguments --------- transformer : sb.lobes.models.TransformerInterface A Transformer instance that contains a .encode() function. *args : tuple **kwargs : dict Arguments to forward to parent class. Example ------- >>> src = torch.rand([8, 120, 512]) >>> tgt = torch.randint(0, 720, [8, 120]) >>> net = TransformerASR( ... 720, 512, 512, 8, 1, 1, 1024, activation=torch.nn.GELU ... ) >>> encoder = EncoderWrapper(net) >>> enc_out = encoder(src) >>> enc_out.shape torch.Size([8, 120, 512]) cs&tj|i|||_|jj|_dSr)rZr[r>r)r`r>argskwargsrbrrr[szEncoderWrapper.__init__NrcKs|jj|||fi|}|S)z:Processes the input tensor x and returns an output tensor.)r>r)r`xwav_lensr.rrrrrszEncoderWrapper.forwardcCs|j||}|S)zdProcesses the input audio chunk tensor `x`, using and updating the mutable encoder `context`)r>r)r`rrrrrrsz EncoderWrapper.forward_streamingcOs|jj|i|S)zInitializes a streaming context. Forwards all arguments to the underlying transformer. See :meth:`speechbrain.lobes.models.transformer.TransformerASR.make_streaming_context`. )r>r)r`rrrrrrsz%EncoderWrapper.make_streaming_contextr) rrrrr[rrrrrrrbrrs   r)FN)NNrFN)#r dataclassesrtypingrrr"rspeechbrain.dataio.dataior0speechbrain.lobes.models.transformer.Transformerrrr r speechbrain.nnet.activationsr speechbrain.nnet.containersr speechbrain.nnet.linearr (speechbrain.utils.dynamic_chunk_trainingrspeechbrain.utils.loggerrrrXrTensorr1r-r8r9rrrrrrsR         = =~