class_docstr: The Recurrent Neural Network Grammar (RNNG) parser from
  Dyer et al.: https://arxiv.org/abs/1602.07776 and
  Gupta et al.: https://arxiv.org/abs/1810.07942d.
  RNNG is a neural constituency parsing algorithm that
  explicitly models compositional structure of a sentence.
  It is able to learn about hierarchical relationship among the words and
  phrases in a given sentence thereby learning the underlying tree structure.
  The paper proposes generative as well as discriminative approaches.
  In PyText we have implemented the discriminative approach for modeling
  intent slot models.
  It is a top-down shift-reduce parser than can output
  trees with non-terminals (intent and slot labels) and terminals (tokens)
docstr: RNNG forward function.
  
  Args:
      tokens (torch.Tensor): list of tokens
      seq_lens (torch.Tensor): list of sequence lengths
      dict_feat (Optional[Tuple[torch.Tensor, ...]]): dictionary or gazetteer
          features for each token
      actions (Optional[List[List[int]]]): Used only during training.
          Oracle actions for the instances.
  
  Returns:
      list of top k tuple of predicted actions tensor and corresponding scores tensor.
      Tensor shape:
      (batch_size, action_length)
      (batch_size, action_length, number_of_actions)
code: def forward(
      self,
      tokens: torch.Tensor,
      seq_lens: torch.Tensor,
      dict_feat: Optional[Tuple[torch.Tensor, ...]] = None,
      actions: Optional[List[List[int]]] = None,
      contextual_token_embeddings: Optional[torch.Tensor] = None,
      beam_size=1,
      top_k=1,
  ) -> List[Tuple[torch.Tensor, torch.Tensor]]:
      """RNNG forward function.
  
      Args:
          tokens (torch.Tensor): list of tokens
          seq_lens (torch.Tensor): list of sequence lengths
          dict_feat (Optional[Tuple[torch.Tensor, ...]]): dictionary or gazetteer
              features for each token
          actions (Optional[List[List[int]]]): Used only during training.
              Oracle actions for the instances.
  
      Returns:
          list of top k tuple of predicted actions tensor and corresponding scores tensor.
          Tensor shape:
          (batch_size, action_length)
          (batch_size, action_length, number_of_actions)
      """
  
      if self.stage != Stage.TEST:
          beam_size = 1
          top_k = 1
  
      if self.training:
          assert (
              actions is not None
          ), "actions must be provided for training"
          actions_idx_rev = list(reversed(actions[0]))
      else:
          torch.manual_seed(0)
  
      beam_size = max(beam_size, 1)
  
      # Reverse the order of input tokens.
      tokens_list_rev = torch.flip(tokens, [len(tokens.size()) - 1])
  
      # Aggregate inputs for embedding module.
      embedding_input = [tokens]
      if dict_feat is not None:
          embedding_input.append(dict_feat)
      if contextual_token_embeddings is not None:
          embedding_input.append(contextual_token_embeddings)
  
      # Embed and reverse the order of tokens.
      token_embeddings = self.embedding(*embedding_input)
      token_embeddings = torch.flip(
          token_embeddings, [len(tokens.size()) - 1]
      )
  
      # Batch size is always = 1. So we squeeze the batch_size dimension.
      token_embeddings = token_embeddings.squeeze(0)
      tokens_list_rev = tokens_list_rev.squeeze(0)
  
      initial_state = ParserState(self)
      for i in range(token_embeddings.size()[0]):
          token_embedding = token_embeddings[i].unsqueeze(0)
          tok = tokens_list_rev[i]
          initial_state.buffer_stackrnn.push(
              token_embedding, Element(tok)
          )
  
      beam = [initial_state]
      while beam and any(not state.finished() for state in beam):
          # Stores plans for expansion as (score, state, action)
          plans: List[Tuple[float, ParserState, int]] = []
          # Expand current beam states
          for state in beam:
              # Keep terminal states
              if state.finished():
                  plans.append((state.neg_prob, state, -1))
                  continue
  
              #  translating Expression p_t = affine_transform({pbias, S,
              #  stack_summary, B, buffer_summary, A, action_summary});
              stack = state.stack_stackrnn
              stack_summary = stack.embedding()
              action_summary = state.action_stackrnn.embedding()
              buffer_summary = state.buffer_stackrnn.embedding()
              if self.dropout_layer.p > 0:
                  stack_summary = self.dropout_layer(stack_summary)
                  action_summary = self.dropout_layer(action_summary)
                  buffer_summary = self.dropout_layer(buffer_summary)
  
              # feature for index of last open non-terminal
              last_open_NT_feature = torch.zeros(
                  len(self.actions_vocab)
              )
              open_NT_exists = state.num_open_NT > 0
  
              if (
                  len(stack) > 0
                  and open_NT_exists
                  and self.ablation_use_last_open_NT_feature
              ):
                  last_open_NT = None
                  try:
                      open_NT = state.is_open_NT[::-1].index(True)
                      last_open_NT = stack.element_from_top(open_NT)
                  except ValueError:
                      pass
                  if last_open_NT:
                      last_open_NT_feature[last_open_NT.node] = 1.0
              last_open_NT_feature = last_open_NT_feature.unsqueeze(0)
  
              summaries = []
              if self.ablation_use_buffer:
                  summaries.append(buffer_summary)
              if self.ablation_use_stack:
                  summaries.append(stack_summary)
              if self.ablation_use_action:
                  summaries.append(action_summary)
              if self.ablation_use_last_open_NT_feature:
                  summaries.append(last_open_NT_feature)
  
              state.action_p = self.action_linear(
                  torch.cat(summaries, dim=1)
              )
  
              log_probs = F.log_softmax(state.action_p, dim=1)[0]
  
              for action in self.valid_actions(state):
                  plans.append(
                      (
                          state.neg_prob - log_probs[action].item(),
                          state,
                          action,
                      )
                  )
  
          beam = []
          # Take actions to regenerate the beam
          for neg_prob, state, predicted_action_idx in sorted(plans)[
              :beam_size
          ]:
              # Skip terminal states
              if state.finished():
                  beam.append(state)
                  continue
  
              # Only branch out states when needed
              if beam_size > 1:
                  state = state.copy()
  
              state.predicted_actions_idx.append(predicted_action_idx)
  
              target_action_idx = predicted_action_idx
              if self.training:
                  assert (
                      len(actions_idx_rev) > 0
                  ), "Actions and tokens may not be in sync."
                  target_action_idx = actions_idx_rev[-1]
                  actions_idx_rev = actions_idx_rev[:-1]
  
              if (
                  self.constraints_ignore_loss_for_unsupported
                  and state.found_unsupported
              ):
                  pass
              else:
                  state.action_scores.append(state.action_p)
  
              self.push_action(state, target_action_idx)
  
              state.neg_prob = neg_prob
              beam.append(state)
          # End for
      # End while
      assert len(beam) > 0, "How come beam is empty?"
      assert (
          len(state.stack_stackrnn) == 1
      ), "How come stack len is " + str(len(state.stack_stackrnn))
      assert (
          len(state.buffer_stackrnn) == 0
      ), "How come buffer len is " + str(len(state.buffer_stackrnn))
  
      # Unsqueeze to add batch dimension before returning.
      return [
          (
              cuda_utils.LongTensor(
                  state.predicted_actions_idx
              ).unsqueeze(0),
              torch.cat(state.action_scores).unsqueeze(0),
          )
          for state in sorted(beam)[:top_k]
      ]
  
imports: from enum import Enum
  from typing import Dict, List, Optional, Tuple
  import numpy
  import pytext.utils.cuda
  import torch
  import torch.nn
  import torch.nn.functional
  from pytext.common.constants import Stage
  from pytext.config import ConfigBase
  from pytext.config.component import ComponentType
  from pytext.data import CommonMetadata
  from pytext.data.tensorizers import AnnotationNumberizer, Tensorizer, TokenTensorizer
  from pytext.data.utils import pad_and_tensorize
  from pytext.models import BaseModel, Model
  from pytext.models.embeddings import EmbeddingList
  from pytext.models.embeddings.word_embedding import WordEmbedding
  from pytext.models.module import create_module
  from pytext.models.representations.bilstm import BiLSTM
  from pytext.models.semantic_parsers.rnng.rnng_data_structures import CompositionalNN, CompositionalSummationNN, Element, ParserState
  from pytext.utils.usage import log_class_usage
label_desc: Use image node_id 3 for calling the RNNGParserBase obj's underlying member method code with example usage: obj.forward(tokens, seq_lens, dict_feat, actions, contextual_token_embeddings, beam_size, top_k) and returns: unknown