class_docstr: This class implements the beam-search algorithm for the transducer model.
  
  Parameters
  ----------
  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 bloc, Linear..] => chars prob
  blank_id : int
      The blank symbol/index.
  beam : int
      The width of beam. Greedy Search is used when beam = 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)
class_docstr_tok: ["This","class","implements","the","beam-search","algorithm","for","the","transducer","model",".","Parameters","--","--","--","--","--","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","bloc",",","Linear","..","]","=",">","chars","prob","blank_id",":","int","The","blank","symbol\/index",".","beam",":","int","The","width","of","beam",".","Greedy","Search","is","used","when","beam","=","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","(","\u03bb",")",".","log","P","(","y|x",")","+","\u03bb","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",")"]
docstr: 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.
docstr_tok: ["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","."]
code: def transducer_beam_search_decode(self, tn_output):
      """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.
      """
  
      # min between beam and max_target_lent
      nbest_batch = []
      nbest_batch_score = []
      for i_batch in range(tn_output.size(0)):
          # if we use RNN LM keep there hiddens
          # prepare BOS = Blank for the Prediction Network (PN)
          # Prepare Blank prediction
          blank = (
              torch.ones(
                  (1, 1), device=tn_output.device, dtype=torch.int32
              )
              * self.blank_id
          )
          input_PN = (
              torch.ones(
                  (1, 1), device=tn_output.device, dtype=torch.int32
              )
              * self.blank_id
          )
          # First forward-pass on PN
          hyp = {
              "prediction": [self.blank_id],
              "logp_score": 0.0,
              "hidden_dec": None,
          }
          if self.lm_weight > 0:
              lm_dict = {"hidden_lm": None}
              hyp.update(lm_dict)
          beam_hyps = [hyp]
  
          # For each time step
          for t_step in range(tn_output.size(1)):
              # get hyps for extension
              process_hyps = beam_hyps
              beam_hyps = []
              while True:
                  if len(beam_hyps) >= self.beam_size:
                      break
                  # Add norm score
                  a_best_hyp = max(
                      process_hyps,
                      key=partial(get_transducer_key),
                  )
  
                  # Break if best_hyp in A is worse by more than state_beam than best_hyp in B
                  if len(beam_hyps) > 0:
                      b_best_hyp = max(
                          beam_hyps,
                          key=partial(get_transducer_key),
                      )
                      a_best_prob = a_best_hyp["logp_score"]
                      b_best_prob = b_best_hyp["logp_score"]
                      if b_best_prob >= self.state_beam + a_best_prob:
                          break
  
                  # remove best hyp from process_hyps
                  process_hyps.remove(a_best_hyp)
  
                  # forward PN
                  input_PN[0, 0] = a_best_hyp["prediction"][-1]
                  out_PN, hidden = self._forward_PN(
                      input_PN,
                      self.decode_network_lst,
                      a_best_hyp["hidden_dec"],
                  )
                  # do unsqueeze over since tjoint must be have a 4 dim [B,T,U,Hidden]
                  log_probs = self._joint_forward_step(
                      tn_output[i_batch, t_step, :]
                      .unsqueeze(0)
                      .unsqueeze(0)
                      .unsqueeze(0),
                      out_PN.unsqueeze(0),
                  )
  
                  if self.lm_weight > 0:
                      log_probs_lm, hidden_lm = self._lm_forward_step(
                          input_PN, a_best_hyp["hidden_lm"]
                      )
  
                  # Sort outputs at time
                  logp_targets, positions = torch.topk(
                      log_probs.view(-1), k=self.beam_size, dim=-1
                  )
                  best_logp = (
                      logp_targets[0]
                      if positions[0] != blank
                      else logp_targets[1]
                  )
  
                  # Extend hyp by  selection
                  for j in range(logp_targets.size(0)):
                      # hyp
                      topk_hyp = {
                          "prediction": a_best_hyp["prediction"][:],
                          "logp_score": a_best_hyp["logp_score"]
                          + logp_targets[j],
                          "hidden_dec": a_best_hyp["hidden_dec"],
                      }
  
                      if positions[j] == self.blank_id:
                          beam_hyps.append(topk_hyp)
                          if self.lm_weight > 0:
                              topk_hyp["hidden_lm"] = a_best_hyp[
                                  "hidden_lm"
                              ]
                          continue
  
                      if (
                          logp_targets[j]
                          >= best_logp - self.expand_beam
                      ):
                          topk_hyp["prediction"].append(
                              positions[j].item()
                          )
                          topk_hyp["hidden_dec"] = hidden
                          if self.lm_weight > 0:
                              topk_hyp["hidden_lm"] = hidden_lm
                              topk_hyp["logp_score"] += (
                                  self.lm_weight
                                  * log_probs_lm[0, 0, positions[j]]
                              )
                          process_hyps.append(topk_hyp)
          # Add norm score
          nbest_hyps = sorted(
              beam_hyps,
              key=partial(get_transducer_key),
              reverse=True,
          )[: self.nbest]
          all_predictions = []
          all_scores = []
          for hyp in nbest_hyps:
              all_predictions.append(hyp["prediction"][1:])
              all_scores.append(
                  hyp["logp_score"] / len(hyp["prediction"])
              )
          nbest_batch.append(all_predictions)
          nbest_batch_score.append(all_scores)
      return (
          [nbest_utt[0] for nbest_utt in nbest_batch],
          torch.Tensor(
              [
                  nbest_utt_score[0]
                  for nbest_utt_score in nbest_batch_score
              ]
          )
          .exp()
          .mean(),
          nbest_batch,
          nbest_batch_score,
      )
  
file_path: speechbrain/speechbrain/decoders/transducer.py
imports: import torch
  from functools import partial
label_desc: Use image node_id 4 for calling the TransducerBeamSearcher obj's underlying member method code with example usage: obj.transducer_beam_search_decode(tn_output) and returns: unknown,  torch,  nbest_batch,  nbest_batch_score