class_docstr: Transformer-based (duplex) decoder model for TN/ITN.
docstr: Called at the end of validation to aggregate outputs.
  
  Args:
      outputs: list of individual outputs of each validation step.
code: def multi_validation_epoch_end(
      self, outputs: List, dataloader_idx=0, split="val"
  ):
      """
      Called at the end of validation to aggregate outputs.
  
      Args:
          outputs: list of individual outputs of each validation step.
      """
      avg_loss = torch.stack(
          [x[f"{split}_loss"] for x in outputs]
      ).mean()
  
      # create a dictionary to store all the results
      results = {}
      directions = (
          [constants.TN_MODE, constants.ITN_MODE]
          if self.mode == constants.JOINT_MODE
          else [self.mode]
      )
      for class_name in self._val_class_to_id[dataloader_idx]:
          for direction in directions:
              results[f"correct_{class_name}_{direction}"] = 0
              results[f"total_{class_name}_{direction}"] = 0
  
      for key in results:
          count = [x[key] for x in outputs if key in x]
          count = (
              torch.stack(count).sum()
              if len(count) > 0
              else torch.tensor(0).to(self.device)
          )
          results[key] = count
  
      all_results = defaultdict(list)
  
      if torch.distributed.is_initialized():
          world_size = torch.distributed.get_world_size()
          for ind in range(world_size):
              for key, v in results.items():
                  all_results[key].append(torch.empty_like(v))
          for key, v in results.items():
              torch.distributed.all_gather(all_results[key], v)
      else:
          for key, v in results.items():
              all_results[key].append(v)
  
      if (
          not torch.distributed.is_initialized()
          or torch.distributed.get_rank() == 0
      ):
          if split == "test":
              val_name = self._test_names[dataloader_idx].upper()
          else:
              val_name = self._validation_names[dataloader_idx].upper()
          final_results = defaultdict(int)
          for key, v in all_results.items():
              for _v in v:
                  final_results[key] += _v.item()
  
          accuracies = defaultdict(dict)
          for key, value in final_results.items():
              if "total_" in key:
                  _, class_name, mode = key.split("_")
                  correct = final_results[
                      f"correct_{class_name}_{mode}"
                  ]
                  if value == 0:
                      accuracies[mode][class_name] = (0, correct, value)
                  else:
                      acc = round(correct / value * 100, 3)
                      accuracies[mode][class_name] = (
                          acc,
                          correct,
                          value,
                      )
  
          for mode, values in accuracies.items():
              report = f"Accuracy {mode.upper()} task {val_name}:\n"
              report += "\n".join(
                  [
                      get_formatted_string(
                          (class_name, f"{v[0]}% ({v[1]}/{v[2]})"),
                          str_max_len=24,
                      )
                      for class_name, v in values.items()
                  ]
              )
              # calculate average across all classes
              all_total = 0
              all_correct = 0
              for _, class_values in values.items():
                  _, correct, total = class_values
                  all_correct += correct
                  all_total += total
              all_acc = (
                  round((all_correct / all_total) * 100, 3)
                  if all_total > 0
                  else 0
              )
              report += "\n" + get_formatted_string(
                  ("AVG", f"{all_acc}% ({all_correct}/{all_total})"),
                  str_max_len=24,
              )
              logging.info(report)
              accuracies[mode]["AVG"] = [all_acc]
  
      self.log(f"{split}_loss", avg_loss)
      if self.trainer.is_global_zero:
          for mode in accuracies:
              for class_name, values in accuracies[mode].items():
                  self.log(
                      f"{val_name}_{mode.upper()}_acc_{class_name.upper()}",
                      values[0],
                      rank_zero_only=True,
                  )
      return {
          f"{split}_loss": avg_loss,
      }
  
imports: import json
  import os
  from collections import defaultdict
  from math import ceil
  from typing import Dict, List, Optional, Union
  import torch
  from omegaconf import DictConfig
  from pytorch_lightning import Trainer
  from transformers import AutoModelForSeq2SeqLM, AutoTokenizer, DataCollatorForSeq2Seq
  import nemo.collections.nlp.data.text_normalization.constants
  from nemo.collections.common.tokenizers.moses_tokenizers import MosesProcessor
  from nemo.collections.nlp.data.text_normalization import TextNormalizationTestDataset
  from nemo.collections.nlp.data.text_normalization.decoder_dataset import TarredTextNormalizationDecoderDataset, TextNormalizationDecoderDataset
  from nemo.collections.nlp.models.duplex_text_normalization.utils import get_formatted_string
  from nemo.collections.nlp.models.nlp_model import NLPModel
  from nemo.core.classes.common import PretrainedModelInfo, typecheck
  from nemo.core.neural_types import ChannelType, LabelsType, LossType, MaskType, NeuralType
  from nemo.utils import logging
label_desc: Use image node_id 8 for calling the DuplexDecoderModel obj's underlying member method code with example usage: obj.multi_validation_epoch_end(outputs, dataloader_idx, split) and returns: dict