o %ݫim@s dZddlZddlmZddlmZmZddlZddlm Z ddl m m Z ddl ZddlmZddlmZmZmZmZddlmZddlmZdd lmZeGd d d ZeGd d d ZGddde jZGddde jZ Gddde jZ!Gddde jZ"Gddde jZ#dS)zConformer implementation. Authors ------- * Jianyuan Zhong 2020 * Samuele Cornell 2021 * Sylvain de Langen 2023 * Shucong Zhang 2024 N) dataclass)ListOptional)Swish)MultiheadAttentionPositionalwiseFeedForward RelPosMHAXLRoPEMHA) HyperMixing) LayerNorm)DynChunkTrainConfigc@sBeZdZUdZeed< dZeej ed< dZ eej ed<dS)%ConformerEncoderLayerStreamingContextaStreaming metadata and state for a `ConformerEncoderLayer`. The multi-head attention and Dynamic Chunk Convolution require to save some left context that gets inserted as left padding. See :class:`.ConvolutionModule` documentation for further details. mha_left_context_sizeNmha_left_contextdcconv_left_context) __name__ __module__ __qualname____doc__int__annotations__rrtorchTensorrrrb/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/lobes/models/transformer/Conformer.pyr s r c@s(eZdZUdZeed< eeed<dS) ConformerEncoderStreamingContextz6Streaming metadata and state for a `ConformerEncoder`.dynchunktrain_configlayersN)rrrrr rrr rrrrr?s  rcsVeZdZdZddedddffdd Z dd ejd eejd ee fd dZ Z S)ConvolutionModuleajThis is an implementation of convolution module in Conformer. Arguments --------- input_size : int The expected size of the input embedding dimension. kernel_size: int, optional Kernel size of non-bottleneck convolutional layer. bias: bool, optional Whether to use bias in the non-bottleneck conv layer. activation: torch.nn.Module Activation function used after non-bottleneck conv layer. dropout: float, optional Dropout rate. causal: bool, optional Whether the convolution should be causal or not. dilation: int, optional Dilation factor for the non bottleneck conv layer. Example ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> net = ConvolutionModule(512, 3) >>> output = net(x) >>> output.shape torch.Size([8, 60, 512]) TFc st||_||_||_|jr|dd|d|_n |dd|dd|_t||_t tj |d|dd|dtj dd|_ tj |||d|j|||d|_ t t||tj|||dt||_dS)Nr!) kernel_sizestridebiasdim)r#r$paddingdilationgroupsr%)r%)super__init__r#causalr)r(nnr layer_norm SequentialConv1dGLU bottleneckconvLinearDropout after_conv)self input_sizer#r% activationdropoutr-r) __class__rrr,is<    zConvolutionModule.__init__Nxmaskrc Cs|dur|jr Jd|jdksJd|j}|jd}|jd|dkr/||jd|}nd}||}|dd}||}tj||j |fdd}|j d||j |d}tj|d|j fdd}|dd}|j ddd }tj ||j j|j j|j jd|j j|j jd }|dd}||}tj|d|d fd }tj |ddd }|dkr|ddd| ddf}n-||}|dd}||}| |}|jr|d d|j f}|dd}||}|dur||d|S)aApplies the convolution to an input tensor `x`. Arguments --------- x: torch.Tensor Input tensor to the convolution module. mask: torch.Tensor, optional Mask to be applied over the output of the convolution using `masked_fill_`, if specified. dynchunktrain_config: DynChunkTrainConfig, optional If specified, makes the module support Dynamic Chunk Convolution (DCConv) as implemented by `Dynamic Chunk Convolution for Unified Streaming and Non-Streaming Conformer ASR `_. This allows masking future frames while preserving better accuracy than a fully causal convolution, at a small speed cost. This should only be used for training (or, if you know what you're doing, for masked evaluation at inference time), as the forward streaming function should be used at inference time. Returns ------- out: torch.Tensor The output tensor. Nz5Chunked convolution not supported with causal paddingr!z:Current DynChunkTrain logic does not support dilation != 1rr")value)sizestep) start_dimend_dim)weightr%r$r(r)r*)r'sizes.r )r-r) chunk_sizeshaper/ transposer3Fpadr(unfoldflattenconv1dr4rEr%r$r*r7r unflatten masked_fill_)r8r>r?rrH batch_sizefinal_right_paddingoutrrrforwards`              zConvolutionModule.forward)NN) rrrrrr,rrrr rU __classcell__rrr<rrKs$ 9rc seZdZdZdddeddddffdd Z    dd eejd eejd ejd ee fddZ dde d ejfddZ de fddZZS)ConformerEncoderLayeraThis is an implementation of Conformer encoder layer. Arguments --------- d_model : int The expected size of the input embedding. d_ffn : int Hidden size of self-attention Feed Forward layer. nhead : int Number of attention heads. kernel_size : int, optional Kernel size of convolution model. kdim : int, optional Dimension of the key. vdim : int, optional Dimension of the value. activation: torch.nn.Module Activation function used in each Conformer layer. bias : bool, optional Whether convolution module. dropout : int, optional Dropout for the encoder. causal : bool, optional Whether the convolutions should be causal or not. attention_type : str, optional type of attention layer, e.g. regularMHA for regular MultiHeadAttention. Example ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> pos_embs = torch.rand((1, 2*60-1, 512)) >>> net = ConformerEncoderLayer(d_ffn=512, nhead=8, d_model=512, kernel_size=3) >>> output = net(x, pos_embs=pos_embs) >>> output[0].shape torch.Size([8, 60, 512]) rNTr Frc st| dkrt||| ||d|_n)| dkr"t||| | d|_n| dkr1t||d|dd|_n | dkr=t||| d |_t||||| | d |_t t |t ||| |d t | |_t t |t ||| |d t | |_t ||_t ||_t | |_dS) N regularMHAnheadd_modelr;kdimvdimr num_heads embed_dimr;mask_pos_future hypermixingF)input_output_dim hypernet_sizetiedr_fix_tm_hidden_sizer )r_r`r;r-d_ffnr9r;r:)r+r,r mha_layerrr r rconvolution_moduler.r0r rr6 ffn_module1 ffn_module2norm1norm2drop r8r[rirZr#r\r]r:r%r;r-attention_typer<rrr,tsp       zConformerEncoderLayer.__init__src_masksrc_key_padding_maskpos_embsrc Csd}|dur |d}|d||}|}||}|j||||||d\}}||}||j|||d}||d||}||fS)a Arguments ---------- src : torch.Tensor The sequence to the encoder layer. src_mask : torch.Tensor, optional The mask for the src sequence. src_key_padding_mask : torch.Tensor, optional The mask for the src keys per batch. pos_embs: torch.Tensor, torch.nn.Module, optional Module or tensor containing the input sequence positional embeddings dynchunktrain_config: Optional[DynChunkTrainConfig] Dynamic Chunk Training configuration object for streaming, specifically involved here to apply Dynamic Chunk Convolution to the convolution module. NrF? attn_maskkey_padding_maskrur) unsqueezerlrnrjrkrorm) r8r>rsrtrur conv_maskskip self_attnrrrrUs(   zConformerEncoderLayer.forwardcontextcCs*|jd}|d||}|jdurtj|j|fdd}|jdkr0|d|j dddf|_|}||}|j|||dd|d\}}||}|d| dddf}|jdurctj|j|fdd}|d|j j dddf|_|| |}|d| dddf}| |d| |}||fS) aConformer layer streaming forward (typically for DynamicChunkTraining-trained models), which is to be used at inference time. Relies on a mutable context object as initialized by `make_streaming_context` that should be used across chunks. Invoked by `ConformerEncoder.forward_streaming`. Arguments --------- x : torch.Tensor Input tensor for this layer. Batching is supported as long as you keep the context consistent. context : ConformerEncoderStreamingContext Mutable streaming context; the same object should be passed across calls. pos_embs : torch.Tensor, optional Positional embeddings, if used. Returns ------- x : torch.Tensor Output tensor. self_attn : list List of self attention values. rvNr!r&r.rw) rIrlrrcatrrnrjrrkr(rorm)r8r>rruorig_lenr}r~rrrforward_streamings:      z'ConformerEncoderLayer.forward_streamingrcCs t|dS)aUCreates a blank streaming context for this encoding layer. Arguments --------- mha_left_context_size : int How many left frames should be saved and used as left context to the current chunk when streaming Returns ------- ConformerEncoderLayerStreamingContext r)r )r8rrrrmake_streaming_contextJs z,ConformerEncoderLayer.make_streaming_contextNNNNN)rrrrrr,rrrr rUr rrrrVrrr<rrWMs>+P 6 WrWc seZdZdZdddeddddddf fdd Z    dd eejd eejd eejd ee fddZ ddejde d eejfddZ d e fddZ ZS)ConformerEncoderaThis class implements the Conformer encoder. Arguments --------- num_layers : int Number of layers. d_model : int Embedding dimension size. d_ffn : int Hidden size of self-attention Feed Forward layer. nhead : int Number of attention heads. kernel_size : int, optional Kernel size of convolution model. kdim : int, optional Dimension of the key. vdim : int, optional Dimension of the value. activation: torch.nn.Module Activation function used in each Confomer layer. bias : bool, optional Whether convolution module. dropout : int, optional Dropout for the encoder. causal: bool, optional Whether the convolutions should be causal or not. attention_type: str, optional type of attention layer, e.g. regulaMHA for regular MultiHeadAttention. 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 ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> pos_emb = torch.rand((1, 2*60-1, 512)) >>> net = ConformerEncoder(1, 512, 512, 8) >>> output, _ = net(x, pos_embs=pos_emb) >>> output.shape torch.Size([8, 60, 512]) >>> import torch >>> from speechbrain.lobes.models.transformer.Conformer import ConformerEncoder >>> x = torch.rand((8, 60, 512)); pos_emb = torch.rand((1, 2*60-1, 512)); >>> net = ConformerEncoder(4, 512, 512, 8, output_hidden_states=True) >>> output, _, hs = net(x, pos_embs=pos_emb) >>> hs[0].shape torch.Size([8, 60, 512]) rNTr Frc sbttj f ddt|D|_tdd|_||_ |_ | |_ dS)Nc*g|]}t  d qS) rirZr[r\r]r;r:r#r%r-rr)rW).0i r:rrr%r-rir[r;r\r#rZr]rr  z-ConformerEncoder.__init__..ư>eps) r+r,rr. ModuleListrangerr normlayerdrop_probrroutput_hidden_states)r8 num_layersr[rirZr#r\r]r:r%r;r-rrrrr<rrr,s  zConformerEncoder.__init__rsrtrurc Cs|jdkr|durtd|jd|}|jdkr!tt|j}g}|jr)|g} t|jD]+\} } |j rA|jdksA|| |jkrY| |||||d\}} | | |jrY| |q.| |}|jrg||| fS||fS)a Arguments --------- src : torch.Tensor The sequence to the encoder layer. src_mask : torch.Tensor, optional The mask for the src sequence. src_key_padding_mask : torch.Tensor, optional The mask for the src keys per batch. pos_embs: torch.Tensor, torch.nn.Module, Module or tensor containing the input sequence positional embeddings If custom pos_embs are given it needs to have the shape (1, 2*S-1, E) where S is the sequence length, and E is the embedding dimension. dynchunktrain_config: Optional[DynChunkTrainConfig] Dynamic Chunk Training configuration object for streaming, specifically involved here to apply Dynamic Chunk Convolution to the convolution module. Returns ------- output : torch.Tensor The output of the Conformer. 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. rN/The chosen attention type for the Conformer is B. For this attention type, the positional embeddings are mandatoryr )rsrtrur) rr ValueErrorrrrandlenrr enumeratetrainingappendr) r8srcrsrtruroutput keep_probs attention_lsthidden_state_lstr enc_layer attentionrrrrUs> $       zConformerEncoder.forwardrrc Cs|jdks |jdkr|durtd|jd|}g}t|jD]\}}|j|||j|d\}}||q ||}||fS)aMConformer streaming forward (typically for DynamicChunkTraining-trained models), which is to be used at inference time. Relies on a mutable context object as initialized by `make_streaming_context` that should be used across chunks. Arguments --------- src : torch.Tensor Input tensor. Batching is supported as long as you keep the context consistent. context : ConformerEncoderStreamingContext Mutable streaming context; the same object should be passed across calls. pos_embs : torch.Tensor, optional Positional embeddings, if used. Returns ------- output : torch.Tensor The output of the streaming conformer. attention_lst : list The attention values. rr Nrr)rur)rrrrrrrr) r8rrrurrrrrrrrrs      z"ConformerEncoder.forward_streamingcstfdd|jDdS)a,Creates a blank streaming context for the encoder. Arguments --------- dynchunktrain_config: Optional[DynChunkTrainConfig] Dynamic Chunk Training configuration object for streaming Returns ------- ConformerEncoderStreamingContext csg|] }|jdqS)r)rleft_context_size_frames)rlayerrzrrrGs z;ConformerEncoder.make_streaming_context..)rr)rr)r8rrrzrr9s   z'ConformerEncoder.make_streaming_contextrr)rrrrrr,rrrr rUrrrrVrrr<rr\sF;- O 2rcsFeZdZdZddeddddffdd Z      d dd ZZS) ConformerDecoderLayeraThis is an implementation of Conformer encoder layer. Arguments --------- d_model : int The expected size of the input embedding. d_ffn : int Hidden size of self-attention Feed Forward layer. nhead : int Number of attention heads. kernel_size : int, optional Kernel size of convolution model. kdim : int, optional Dimension of the key. vdim : int, optional Dimension of the value. activation : torch.nn.Module, optional Activation function used in each Conformer layer. bias : bool, optional Whether convolution module. dropout : int, optional Dropout for the encoder. causal : bool, optional Whether the convolutions should be causal or not. attention_type : str, optional type of attention layer, e.g. regularMHA for regular MultiHeadAttention. Example ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> pos_embs = torch.rand((1, 2*60-1, 512)) >>> net = ConformerEncoderLayer(d_ffn=512, nhead=8, d_model=512, kernel_size=3) >>> output = net(x, pos_embs=pos_embs) >>> output[0].shape torch.Size([8, 60, 512]) NTr rc st| s td| dkrt||| ||d|_n | dkr(t||| | d|_t||||| | d|_t t |t ||| |dt | |_t t |t ||| |dt | |_t ||_t ||_t | |_dS)NzWDecoder is not causal, in most applications it should be causal, you have been warned !rXrYrr^rgrh)r+r,warningswarnrrjrrrkr.r0r rr6rlrmrnrorprqr<rrr,ws\     zConformerDecoderLayer.__init__c Csr|d||}|} ||} |j| |||||d\} } | | } | || } || d|| } | | | fS)am Arguments --------- tgt: torch.Tensor The sequence to the decoder layer. memory: torch.Tensor The sequence from the last layer of the encoder. tgt_mask: torch.Tensor, optional, optional The mask for the tgt sequence. memory_mask: torch.Tensor, optional The mask for the memory sequence. tgt_key_padding_mask: torch.Tensor, optional The mask for the tgt keys per batch. memory_key_padding_mask: torch.Tensor, optional The mask for the memory keys per batch. pos_embs_tgt: torch.Tensor, torch.nn.Module, optional Module or tensor containing the target sequence positional embeddings for each attention layer. pos_embs_src: torch.Tensor, torch.nn.Module, optional Module or tensor containing the source sequence positional embeddings for each attention layer. Returns ------- x: torch.Tensor The output tensor self_attn : torch.Tensor self_attn : torch.Tensor The self attention tensor rvrw)rlrnrjrkrorm) r8tgtmemorytgt_mask memory_masktgt_key_padding_maskmemory_key_padding_mask pos_embs_tgt pos_embs_srcr}r>r~rrrrUs(   zConformerDecoderLayer.forwardNNNNNNrrrrrr,rUrVrrr<rrPs",HrcsHeZdZdZdddeddddffdd Z      d d d ZZS) ConformerDecoderaThis class implements the Transformer decoder. Arguments --------- num_layers: int Number of layers. nhead: int Number of attention heads. d_ffn: int Hidden size of self-attention Feed Forward layer. d_model: int Embedding dimension size. kdim: int, optional Dimension for key. vdim: int, optional Dimension for value. dropout: float, optional Dropout rate. activation: torch.nn.Module, optional Activation function used after non-bottleneck conv layer. kernel_size : int, optional Kernel size of convolutional layer. bias : bool, optional Whether convolution module. causal: bool, optional Whether the convolutions should be causal or not. attention_type: str, optional type of attention layer, e.g. regularMHA for regular MultiHeadAttention. Example ------- >>> src = torch.rand((8, 60, 512)) >>> tgt = torch.rand((8, 60, 512)) >>> net = ConformerDecoder(1, 8, 1024, 512, attention_type="regularMHA") >>> output, _, _ = net(tgt, src) >>> output.shape torch.Size([8, 60, 512]) Nr Trc sVttj f ddt|D|_tjj j dd|_ dS)Ncrr)r)r_rrrr1rz-ConformerDecoder.__init__..rr) r+r,rr.rrrsbnnet normalizationr r) r8rrZrir[r\r]r;r:r#r%r-rrr<rrr, s zConformerDecoder.__init__c  Cs`|} gg} } |jD]} | | |||||||d\} } }| | | |q || } | | | fS)a Arguments --------- tgt: torch.Tensor The sequence to the decoder layer. memory: torch.Tensor The sequence from the last layer of the encoder. tgt_mask: torch.Tensor, optional, optional The mask for the tgt sequence. memory_mask: torch.Tensor, optional The mask for the memory sequence. tgt_key_padding_mask : torch.Tensor, optional The mask for the tgt keys per batch. memory_key_padding_mask : torch.Tensor, optional The mask for the memory keys per batch. pos_embs_tgt: torch.Tensor, torch.nn.Module, optional Module or tensor containing the target sequence positional embeddings for each attention layer. pos_embs_src: torch.Tensor, torch.nn.Module, optional Module or tensor containing the source sequence positional embeddings for each attention layer. Returns ------- output: torch.Tensor Conformer decoder output. self_attns : list Location of self attentions. multihead_attns : list Location of multihead attentions. )rrrrrr)rrr)r8rrrrrrrrr self_attnsmultihead_attns dec_layerr~multihead_attnrrrrUDs"(     zConformerDecoder.forwardrrrrr<rrs$.(r)$rr dataclassesrtypingrrrtorch.nnr.torch.nn.functional functionalrK speechbrainrspeechbrain.nnet.activationsrspeechbrain.nnet.attentionrrrr speechbrain.nnet.hypermixingr speechbrain.nnet.normalizationr (speechbrain.utils.dynamic_chunk_trainingr r rModulerrWrrrrrrrs8       u(