class_docstr: Beam search implementation for Transducer.
  
  Args:
      decoder: Decoder module.
      joint_network: Joint network module.
      beam_size: Size of the beam.
      lm: LM module.
      lm_weight: LM weight for soft fusion.
      search_type: Search algorithm to use during inference.
      max_sym_exp: Number of maximum symbol expansions at each time step. (TSD)
      u_max: Maximum expected target sequence length. (ALSD)
      nstep: Number of maximum expansion steps at each time step. (mAES)
      expansion_gamma: Allowed logp difference for prune-by-value method. (mAES)
      expansion_beta:
           Number of additional candidates for expanded hypotheses selection. (mAES)
      score_norm: Normalize final scores by length.
      nbest: Number of final hypothesis.
      streaming: Whether to perform chunk-by-chunk beam search.
class_docstr_tok: ["Beam","search","implementation","for","Transducer",".","Args",":","decoder",":","Decoder","module",".","joint_network",":","Joint","network","module",".","beam_size",":","Size","of","the","beam",".","lm",":","LM","module",".","lm_weight",":","LM","weight","for","soft","fusion",".","search_type",":","Search","algorithm","to","use","during","inference",".","max_sym_exp",":","Number","of","maximum","symbol","expansions","at","each","time","step",".","(","TSD",")","u_max",":","Maximum","expected","target","sequence","length",".","(","ALSD",")","nstep",":","Number","of","maximum","expansion","steps","at","each","time","step",".","(","mAES",")","expansion_gamma",":","Allowed","logp","difference","for","prune-by-value","method",".","(","mAES",")","expansion_beta",":","Number","of","additional","candidates","for","expanded","hypotheses","selection",".","(","mAES",")","score_norm",":","Normalize","final","scores","by","length",".","nbest",":","Number","of","final","hypothesis",".","streaming",":","Whether","to","perform","chunk-by-chunk","beam","search","."]
docstr: Alignment-length synchronous beam search implementation.
  
  Based on https://ieeexplore.ieee.org/document/9053040
  
  Args:
      h: Encoder output sequences. (T, D)
  
  Returns:
      nbest_hyps: N-best hypothesis.
docstr_tok: ["Alignment-length","synchronous","beam","search","implementation",".","Based","on","https",":","\/\/ieeexplore.ieee.org\/document\/9053040","Args",":","h",":","Encoder","output","sequences",".","(","T",",","D",")","Returns",":","nbest_hyps",":","N-best","hypothesis","."]
code: def align_length_sync_decoding(
      self,
      enc_out: torch.Tensor,
  ) -> List[Hypothesis]:
      """Alignment-length synchronous beam search implementation.
  
      Based on https://ieeexplore.ieee.org/document/9053040
  
      Args:
          h: Encoder output sequences. (T, D)
  
      Returns:
          nbest_hyps: N-best hypothesis.
  
      """
      t_max = int(enc_out.size(0))
      u_max = min(self.u_max, (t_max - 1))
  
      B = [
          Hypothesis(
              yseq=[0], score=0.0, dec_state=self.decoder.init_state(1)
          )
      ]
      final = []
  
      if self.use_lm:
          B[0].lm_state = self.lm.zero_state()
  
      for i in range(t_max + u_max):
          A = []
  
          B_ = []
          B_enc_out = []
  
          for hyp in B:
              u = len(hyp.yseq) - 1
              t = i - u
  
              if t > (t_max - 1):
                  continue
  
              B_.append(hyp)
              B_enc_out.append((t, enc_out[t]))
  
          if B_:
              beam_enc_out = torch.stack([b[1] for b in B_enc_out])
              beam_dec_out, beam_state = self.decoder.batch_score(B_)
  
              beam_logp = torch.log_softmax(
                  self.joint_network(beam_enc_out, beam_dec_out),
                  dim=-1,
              )
              beam_topk = beam_logp[:, 1:].topk(self.beam_size, dim=-1)
  
              if self.use_lm:
                  beam_lm_scores, beam_lm_states = self.lm.batch_score(
                      self.create_lm_batch_inputs([b.yseq for b in B_]),
                      [b.lm_state for b in B_],
                      None,
                  )
  
              for i, hyp in enumerate(B_):
                  new_hyp = Hypothesis(
                      score=(hyp.score + float(beam_logp[i, 0])),
                      yseq=hyp.yseq[:],
                      dec_state=hyp.dec_state,
                      lm_state=hyp.lm_state,
                  )
  
                  A.append(new_hyp)
  
                  if B_enc_out[i][0] == (t_max - 1):
                      final.append(new_hyp)
  
                  for logp, k in zip(
                      beam_topk[0][i], beam_topk[1][i] + 1
                  ):
                      new_hyp = Hypothesis(
                          score=(hyp.score + float(logp)),
                          yseq=(hyp.yseq[:] + [int(k)]),
                          dec_state=self.decoder.select_state(
                              beam_state, i
                          ),
                          lm_state=hyp.lm_state,
                      )
  
                      if self.use_lm:
                          new_hyp.score += (
                              self.lm_weight * beam_lm_scores[i, k]
                          )
                          new_hyp.lm_state = beam_lm_states[i]
  
                      A.append(new_hyp)
  
              B = sorted(A, key=lambda x: x.score, reverse=True)[
                  : self.beam_size
              ]
              B = self.recombine_hyps(B)
  
      if final:
          return final
  
      return B
  
imports: from dataclasses import dataclass
  from typing import Any, Dict, List, Optional, Tuple, Union
  import numpy
  import torch
  from espnet2.asr_transducer.decoder.abs_decoder import AbsDecoder
  from espnet2.asr_transducer.joint_network import JointNetwork
label_desc: Use image node_id 9 for calling the BeamSearchTransducer obj's underlying member method code with example usage: obj.align_length_sync_decoding(enc_out) and returns: B, final