class_docstr: Megatron NMT training
class_docstr_tok: ["Megatron","NMT","training"]
code: def eval_epoch_end(self, outputs, mode):
      if not outputs:
          return
  
      if isinstance(outputs[0], dict):
          outputs = [outputs]
  
      loss_list = []
      bleu_score_list = []
      for dataloader_idx, output in enumerate(outputs):
          if parallel_state.is_pipeline_last_stage():
              # only the last pipeline parallel stages return loss
              averaged_loss = torch.stack(
                  [x["loss"] for x in output]
              ).mean()
          else:
              averaged_loss = torch.tensor(0.0).to(self.device)
  
          # we can only log on one rank if it is rank zero so we broadcast from last rank
          torch.distributed.broadcast(averaged_loss, get_last_rank())
  
          # averaged_loss = average_losses_across_data_parallel_group([x['loss'] for x in output])
          inputs = list(itertools.chain(*[x["inputs"] for x in output]))
          translations = list(
              itertools.chain(*[x["translations"] for x in output])
          )
          ground_truths = list(
              itertools.chain(*[x["ground_truths"] for x in output])
          )
          assert len(translations) == len(inputs)
          assert len(translations) == len(ground_truths)
  
          # Gather translations and ground truths from all workers
          tr_gt_inp = [
              None
              for _ in range(
                  parallel_state.get_data_parallel_world_size()
              )
          ]
          # we also need to drop pairs where ground truth is an empty string
          torch.distributed.all_gather_object(
              tr_gt_inp,
              [
                  (t, g, i)
                  for (t, g, i) in zip(
                      translations, ground_truths, inputs
                  )
              ],
              group=parallel_state.get_data_parallel_group(),
          )
          # if parallel_state.get_data_parallel_rank() == 0:
          if self.global_rank == 0:
              _translations = []
              _ground_truths = []
              _inputs = []
  
              # Deduplicate sentences that may have been distributed across multiple data parallel ranks.
              gt_inp_set = set()
              for rank in range(
                  0, parallel_state.get_data_parallel_world_size()
              ):
                  for t, g, i in tr_gt_inp[rank]:
                      if g + i not in gt_inp_set:
                          gt_inp_set.add(g + i)
                          _translations.append(t)
                          _ground_truths.append(g)
                          _inputs.append(i)
  
              if self.tgt_language in ["ja"]:
                  sacre_bleu = corpus_bleu(
                      _translations,
                      [_ground_truths],
                      tokenize="ja-mecab",
                  )
              elif self.tgt_language in ["zh"]:
                  sacre_bleu = corpus_bleu(
                      _translations, [_ground_truths], tokenize="zh"
                  )
              else:
                  sacre_bleu = corpus_bleu(
                      _translations, [_ground_truths], tokenize="13a"
                  )
  
              bleu_score = sacre_bleu.score
  
              dataset_name = "Validation" if mode == "val" else "Test"
              logging.info(
                  f"{dataset_name}, Dataloader index: {dataloader_idx}, Set size: {len(_translations)}"
              )
              logging.info(
                  f"{dataset_name}, Dataloader index: {dataloader_idx}, SacreBLEU = {bleu_score}"
              )
              logging.info(
                  f"{dataset_name}, Dataloader index: {dataloader_idx}, Translation Examples:"
              )
              logging.info(
                  "============================================================"
              )
              for example_idx in range(0, 3):
                  random_index = random.randint(
                      0, len(_translations) - 1
                  )
                  logging.info(
                      "    " + "\u0332".join(f"Example {example_idx}:")
                  )
                  logging.info(
                      f"    Input:        {_inputs[random_index]}"
                  )
                  logging.info(
                      f"    Prediction:   {_translations[random_index]}"
                  )
                  logging.info(
                      f"    Ground Truth: {_ground_truths[random_index]}"
                  )
                  logging.info(
                      "============================================================"
                  )
  
          else:
              bleu_score = 0.0
  
          bleu_score = torch.FloatTensor([bleu_score]).to(self.device)
          # BLEU score is computed on global rank 0 only and then broadcasted to other ranks.
          torch.distributed.all_reduce(
              bleu_score, op=torch.distributed.ReduceOp.SUM
          )
          bleu_score = bleu_score.cpu().item()
          loss_list.append(averaged_loss.cpu().numpy())
          bleu_score_list.append(bleu_score)
          if dataloader_idx == 0:
              if self.multilingual:
                  self._log_multilingual_bleu_and_loss(
                      dataloader_idx, bleu_score, averaged_loss, mode
                  )
              else:
                  self.log(
                      f"{mode}_sacreBLEU", bleu_score, batch_size=1
                  )
                  self.log(
                      f"{mode}_loss",
                      averaged_loss,
                      prog_bar=True,
                      batch_size=1,
                  )
          else:
              if self.multilingual:
                  self._log_multilingual_bleu_and_loss(
                      dataloader_idx, bleu_score, averaged_loss, mode
                  )
              else:
                  self.log(
                      f"{mode}_sacreBLEU_dl_index_{dataloader_idx}",
                      bleu_score,
                      batch_size=1,
                  )
                  self.log(
                      f"{mode}_loss_dl_index_{dataloader_idx}",
                      averaged_loss,
                      prog_bar=False,
                      batch_size=1,
                  )
          outputs[dataloader_idx].clear()  # free memory
  
      if len(loss_list) > 1:
          self.log(
              f"{mode}_loss_avg",
              np.mean(loss_list),
              sync_dist=True,
              batch_size=1,
          )
          self.log(
              f"{mode}_sacreBLEU_avg",
              np.mean(bleu_score_list),
              batch_size=1,
          )
  
      app_state = AppState()
      if hasattr(self, "_train_ds"):
          _reconfigure_microbatch_calculator(
              rank=app_state.global_rank,
              rampup_batch_size=None,
              global_batch_size=self._cfg.train_ds.global_batch_size,
              micro_batch_size=self._cfg.train_ds.micro_batch_size,
              data_parallel_size=parallel_state.get_data_parallel_world_size(),
          )
  
imports: import enum
  import itertools
  import random
  from typing import List, Optional
  import numpy
  import torch
  from omegaconf.dictconfig import DictConfig
  from omegaconf.listconfig import ListConfig
  from pytorch_lightning.trainer.trainer import Trainer
  from sacrebleu import corpus_bleu
  from nemo.collections.nlp.data.common.sequence_to_sequence_dataset import BinarizedMemmapSequenceToSequenceDataset, TextMemmapSequenceToSequenceDataset
  from nemo.collections.nlp.data.language_modeling.megatron.base_dataset_utils import get_datasets_weights_and_num_samples
  from nemo.collections.nlp.data.language_modeling.megatron.blendable_dataset import BlendableDataset
  from nemo.collections.nlp.data.language_modeling.megatron.megatron_batch_samplers import MegatronPretrainingBatchSampler
  from nemo.collections.nlp.data.language_modeling.megatron.xlm_dataset import BinarizedMemmapCrossLingualMLMAndTranslationDataset, TextMemmapCrossLingualMLMAndTranslationDataset
  from nemo.collections.nlp.models.language_modeling.megatron_lm_encoder_decoder_model import MegatronLMEncoderDecoderModel
  from nemo.collections.nlp.models.language_modeling.megatron_t5_model import MegatronT5Model
  from nemo.collections.nlp.models.machine_translation.mt_enc_dec_model import MTEncDecModel
  from nemo.collections.nlp.modules.common.megatron.megatron_export import DecEmb, EncEmb, TokensHeadEmb
  from nemo.collections.nlp.modules.common.megatron.utils import get_iterator_k_split
  from nemo.collections.nlp.parts.nlp_overrides import GlobalBatchDataFetcher
  from nemo.collections.nlp.parts.utils_funcs import get_last_rank
  from nemo.core.classes import Exportable
  from nemo.utils import AppState, logging, timers
label_desc: Use image node_id 14 for calling the MegatronNMTModel obj's underlying member method code with example usage: obj.eval_epoch_end(outputs, mode) without return types