o eicX@sldZddlmZddlmZddlmZmZddlZeGdddej j Z Gdd d ej j Z d d Z dS) zsDecoders and output normalization for Transducer sequence. Author: Abdelwahab HEBA 2020 Sung-Lin Yeh 2020 ) dataclass)partial)AnyOptionalNc@s"eZdZUdZdZeeed<dS)(TransducerGreedySearcherStreamingContextzSimple wrapper for the hidden state of the transducer greedy searcher. Used by :meth:`~TransducerBeamSearcher.transducer_greedy_decode_streaming`. Nhidden)__name__ __module__ __qualname____doc__rrr__annotations__r r ]/home/ubuntu/transcripts/venv/lib/python3.10/site-packages/speechbrain/decoders/transducer.pyrs rcseZdZdZ      d fdd Zd d Z d!d d Zdejde fddZ ddZ ddZ ddZ ddZddZd"ddZddZZS)#TransducerBeamSearcheru9 This class implements the beam-search algorithm for the transducer model. Arguments --------- decode_network_lst : list List of prediction network (PN) layers. tjoint: transducer_joint module This module perform the joint between TN and PN. classifier_network : list List of output layers (after performing joint between TN and PN) exp: (TN,PN) => joint => classifier_network_list [DNN block, Linear..] => chars prob blank_id : int The blank symbol/index. beam_size : int The width of beam. Greedy Search is used when beam_size = 1. nbest : int Number of hypotheses to keep. lm_module : torch.nn.ModuleList Neural networks modules for LM. lm_weight : float The weight of LM when performing beam search (λ). log P(y|x) + λ log P_LM(y). (default: 0.3) state_beam : float The threshold coefficient in log space to decide if hyps in A (process_hyps) is likely to compete with hyps in B (beam_hyps), if not, end the while loop. Reference: https://arxiv.org/pdf/1911.01629.pdf expand_beam : float The threshold coefficient to limit the number of expanded hypotheses that are added in A (process_hyp). Reference: https://arxiv.org/pdf/1911.01629.pdf Reference: https://github.com/kaldi-asr/kaldi/blob/master/src/decoder/simple-decoder.cc (See PruneToks) Example ------- searcher = TransducerBeamSearcher( decode_network_lst=[hparams["emb"], hparams["dec"]], tjoint=hparams["Tjoint"], classifier_network=[hparams["transducer_lin"]], blank_id=0, beam_size=hparams["beam_size"], nbest=hparams["nbest"], lm_module=hparams["lm_model"], lm_weight=hparams["lm_weight"], state_beam=2.3, expand_beam=2.3, ) >>> from speechbrain.nnet.transducer.transducer_joint import Transducer_joint >>> import speechbrain as sb >>> emb = sb.nnet.embedding.Embedding( ... num_embeddings=35, ... embedding_dim=3, ... consider_as_one_hot=True, ... blank_id=0 ... ) >>> dec = sb.nnet.RNN.GRU( ... hidden_size=10, input_shape=(1, 40, 34), bidirectional=False ... ) >>> lin = sb.nnet.linear.Linear(input_shape=(1, 40, 10), n_neurons=35) >>> joint_network= sb.nnet.linear.Linear(input_shape=(1, 1, 40, 35), n_neurons=35) >>> tjoint = Transducer_joint(joint_network, joint="sum") >>> searcher = TransducerBeamSearcher( ... decode_network_lst=[emb, dec], ... tjoint=tjoint, ... classifier_network=[lin], ... blank_id=0, ... beam_size=1, ... nbest=1, ... lm_module=None, ... lm_weight=0.0, ... ) >>> enc = torch.rand([1, 20, 10]) >>> hyps, _, _, _ = searcher(enc) Nffffff@c st||_||_||_||_||_||_||_||_ |dur)|dkr)t d| |_ | |_ t jjdd|_|jdkrB|j|_dS|j|_dS)NrzLanguage model is not provided.dim)super__init__decode_network_lsttjointclassifier_networkblank_id beam_sizenbestlm lm_weight ValueError state_beam expand_beamtorchnn LogSoftmaxsoftmaxtransducer_greedy_decodesearchertransducer_beam_search_decode) selfrrrrrr lm_moduler!r#r$ __class__r rres"    zTransducerBeamSearcher.__init__cCs||}|S)z Arguments --------- tn_output : torch.Tensor Output from transcription network with shape [batch, time_len, hiddens]. Returns ------- Topk hypotheses )r*)r, tn_outputhypsr r rforwards zTransducerBeamSearcher.forwardFcCsddt|dDddt|dDd}tj|ddf|jtjd|j}|dur9|||j\}}n|\}}t|dD]}| |dd|ddf d d| d} tj | d ddd \} } g} t| dD]1} | |  |jkr|d | | |  |d | | | 7<| | || d<| | qvt| dkr|| ||\}}|||j|\}}||| <|| ||}qD|d t|d ddf}|r|||ff7}|S) aTransducer greedy decoder is a greedy decoder over batch which apply Transducer rules: 1- for each time step in the Transcription Network (TN) output: -> Update the ith utterance only if the previous target != the new one (we save the hiddens and the target) -> otherwise: ---> keep the previous target prediction from the decoder Arguments --------- tn_output : torch.Tensor Output from transcription network with shape [batch, time_len, hiddens]. hidden_state : (torch.Tensor, torch.Tensor) Hidden state to initially feed the decode network with. This is useful in conjunction with `return_hidden` to be able to perform beam search in a streaming context, so that you can reuse the last hidden state as an initial state across calls. return_hidden : bool Whether the return tuple should contain an extra 5th element with the hidden state at of the last step. See `hidden_state`. Returns ------- Tuple of 4 or 5 elements (if `return_hidden`). First element: List[List[int]] List of decoded tokens Second element: torch.Tensor Outputs a logits tensor [B,T,1,Output_Dim]; padding has not been removed. Third element: None nbest; irrelevant for greedy decode Fourth element: None nbest scores; irrelevant for greedy decode Fifth element: Present if `return_hidden`, (torch.Tensor, torch.Tensor) Tuple representing the hidden state required to call `transducer_greedy_decode` where you left off in a streaming context. cSsg|]}gqSr r .0_r r r zCTransducerBeamSearcher.transducer_greedy_decode..rcSsg|]}dqS)rr r3r r rr6r7) prediction logp_scoresrdevicedtypeNrr8r9)rangesizer%onesr;int32r _forward_PNr_joint_forward_step unsqueezemaxsqueezeitemappendlen_get_sentence_to_update_update_hiddensTensorexpmean)r,r0 hidden_state return_hiddenhypinput_PNout_PNrt_step log_probs logp_targets positionshave_update_hypiselected_input_PNselected_hiddenselected_out_PNretr r rr)sr/      z/TransducerBeamSearcher.transducer_greedy_decodexcontextcCs&|j||jdd\}}}}}||_|S)aTiny wrapper for :meth:`~TransducerBeamSearcher.transducer_greedy_decode` with an API that makes it suitable to be passed as a `decoding_function` for streaming. Arguments --------- x : torch.Tensor Outputs of the prediction network (equivalent to `tn_output`) context : TransducerGreedySearcherStreamingContext Mutable streaming context object, which must be specified and reused across calls when streaming. You can obtain an initial context by initializing a default object. Returns ------- hyp : torch.Tensor T)rO)r)r)r,r]r^rP_scoresr5rr r r"transducer_greedy_decode_streamings z9TransducerBeamSearcher.transducer_greedy_decode_streamingc CsBg}g}t|dD]}}tjd|jtjd|j}tjd|jtjd|j}|jgddd}|jdkr>ddi}|||g} t|dD]} | } g} t | |j krXnt | t t d } t | dkr~t | t t d } | d }| d }||j|kr~n| | | d d |d<|||j| d\}}|||| ddfddd|d}|jdkr||| d\}}tj|d |j d d\}}|d|kr|dn|d}t|dD]j}| d dd| d ||| dd}|||jkr| ||jdkr| d|d<q||||jkrL|d ||||d<|jdkrG||d<|d |j|dd||f7<| |qqPqHt| t t d dd|j}g}g}|D]}||d dd||d t |d qc||||q dd|Dtdd|D||fS)a.Transducer beam search decoder is a beam search decoder over batch which apply Transducer rules: 1- for each utterance: 2- for each time steps in the Transcription Network (TN) output: -> Do forward on PN and Joint network -> Select topK <= beam -> Do a while loop extending the hyps until we reach blank -> otherwise: --> extend hyp by the new token Arguments --------- tn_output : torch.Tensor Output from transcription network with shape [batch, time_len, hiddens]. Returns ------- torch.Tensor Outputs a logits tensor [B,T,1,Output_Dim]; padding has not been removed. r)rrr:rN)r8 logp_score hidden_dec hidden_lmrT)keyrar8r)rrrb)kr)rdreversecSg|]}|dqSrr )r4 nbest_uttr r rr6zHTransducerBeamSearcher.transducer_beam_search_decode..cSrgrhr )r4nbest_utt_scorer r rr6rj)r=r>r%r?r;r@rr!updaterHrrDrget_transducer_keyr#removerArrBrC_lm_forward_steptopkviewrGr$rFsortedrrKrLrM)r,r0 nbest_batchnbest_batch_scorei_batchblankrQrPlm_dict beam_hypsrS process_hyps a_best_hyp b_best_hyp a_best_prob b_best_probrRrrT log_probs_lmrcrUrV best_logpjtopk_hyp nbest_hypsall_predictions all_scoresr r rr+.s           R    z4TransducerBeamSearcher.transducer_beam_search_decodecCsRt|||}|||j}||}Wd|S1s"wY|S)zJoin predictions (TN & PN).N)r%no_gradr_forward_after_jointrr()r,h_irRoutrTr r rrBs     z*TransducerBeamSearcher._joint_forward_stepcCsRt|j||d\}}||}Wd||fS1s wY||fS)aVThis method should implement one step of forwarding operation for language model. Arguments --------- inp_tokens : torch.Tensor The input tensor of the current timestep. memory : No limit The memory variables input for this timestep. (e.g., RNN hidden states). Return ------ log_probs : torch.Tensor Log-probabilities of the current timestep output. hs : No limit The memory variables are generated in this timestep. (e.g., RNN hidden states). )hxN)r%rr r()r, inp_tokensmemorylogitshsrTr r rros   z'TransducerBeamSearcher._lm_forward_stepcCs|||ddf}t|tr/|ddd|ddf}|ddd|ddf}||f}||fS|dd|ddf}||fS)aSelect and return the updated hiddens and output from the Prediction Network. Arguments --------- selected_sentences : list List of updated sentences (indexes). output_PN: torch.Tensor Output tensor from prediction network (PN). hidden : torch.Tensor Optional: None, hidden tensor to be used for recurrent layers in the prediction network. Returns ------- selected_output_PN: torch.Tensor Outputs a logits tensor [B_selected,U, hiddens]. hidden_update_hyp: torch.Tensor Selected hiddens tensor. Nrr isinstancetuple)r,selected_sentences output_PNrselected_output_PNhidden0_update_hyphidden1_update_hyphidden_update_hypr r rrIs z.TransducerBeamSearcher._get_sentence_to_updatecCsdt|tr%|d|ddd|ddf<|d|ddd|ddf<|S||dd|ddf<|S)aUpdate hidden tensor by a subset of hidden tensor (updated ones). Arguments --------- selected_sentences : list List of index to be updated. updated_hidden : torch.Tensor Hidden tensor of the selected sentences for update. hidden : torch.Tensor Hidden tensor to be updated. Returns ------- torch.Tensor Updated hidden tensor. rNrr)r,rupdated_hiddenrr r rrJs z&TransducerBeamSearcher._update_hiddenscCs6|D]}|jjdvr|||\}}q||}q||fS)aCompute forward-pass through a list of prediction network (PN) layers. Arguments --------- out_PN : torch.Tensor Input sequence from prediction network with shape [batch, target_seq_lens]. decode_network_lst: list List of prediction network (PN) layers. hidden : torch.Tensor Optional: None, hidden tensor to be used for recurrent layers in the prediction network Returns ------- out_PN : torch.Tensor Outputs a logits tensor [B,U, hiddens]. hidden : torch.Tensor Hidden tensor to be used for the next step by recurrent layers in prediction network. )RNNLSTMGRULiGRU LiGRU_Layer)r/r)r,rRrrlayerr r rrA-s   z"TransducerBeamSearcher._forward_PNcCs|D]}||}q|S)a.Compute forward-pass through a list of classifier neural network. Arguments --------- out : torch.Tensor Output from joint network with shape [batch, target_len, time_len, hiddens] classifier_network : list List of output layers (after performing joint between TN and PN) exp: (TN,PN) => joint => classifier_network_list [DNN block, Linear..] => chars prob Returns ------- torch.Tensor Outputs a logits tensor [B, U,T, Output_Dim]; r )r,rrrr r rrQs z+TransducerBeamSearcher._forward_after_joint)rrNrrr)NF)N)rr r r rr2r)r%rKrr`r+rBrorIrJrAr __classcell__r r r.rrs2Q# {   $rcCs|dt|d}|S)aArgument function to customize the sort order (in sorted & max). To be used as `key=partial(get_transducer_key)`. Arguments --------- x : dict one of the items under comparison Returns ------- float Normalized log-score. rar8)rH)r]logp_keyr r rrmhsrm)r dataclassesr functoolsrtypingrrr%r&Modulerrrmr r r rs    S