code: def _compute_validation_metrics(self) -> workload.Response:
      self.context.reset_reducers()
      # Set the behavior of certain layers (e.g., dropout) that are
      # different between training and inference.
      for model in self.context.models:
          model.eval()
  
      step_start_time = time.time()
  
      for callback in self.callbacks.values():
          callback.on_validation_start()
  
      num_inputs = 0
      keys = None
      batch_metrics = []
  
      for callback in self.callbacks.values():
          callback.on_validation_epoch_start()
  
      validation_iterator = (
          iter(self.validation_loader)
          if self.validation_loader
          else None
      )
      for idx in range(cast(int, self.num_validation_batches)):
          num_inputs += cast(int, self.validation_batch_size)
          # Note that when using pipeline parallelism, each call to evaluate_batch will request
          # self.context.num_micro_batches_per_slot batches from the validation iterator.
          # This is why we set self.num_validation_batches differently for pipeline parallel
          # and no pipeline parallel when building the data loaders.
          vld_metrics = self.trial.evaluate_batch(
              validation_iterator, idx
          )
          if self.context._mpu.should_report_metrics:
              if not isinstance(vld_metrics, dict):
                  raise det.errors.InvalidExperimentException(
                      "evaluate_batch must return a dictionary of string names "
                      "to Tensor metrics",
                  )
              # Verify validation metric names are the same across batches.
              if keys is None:
                  keys = vld_metrics.keys()
              else:
                  if keys != vld_metrics.keys():
                      raise det.errors.InvalidExperimentException(
                          "Validation metric names must match across all batches of data.",
                      )
              # TODO: For performance perform -> cpu() only at the end of validation.
              batch_metrics.append(
                  pytorch._convert_metrics_to_numpy(vld_metrics)
              )
          if self.env.test_mode:
              break
  
      # keys and list(keys) does not satisfy all cases because it will return dict_keys type if
      # keys is an empty dict. this will then break when passed to zmq_broadcast since it does
      # not know how to serialize dict_keys type.
      all_keys = self.context.distributed.gather(
          keys if keys is None else list(keys)
      )
      if self.is_chief:
          all_keys = [k for k in all_keys if k is not None]
          keys = all_keys[0]
      keys = self.context.distributed.broadcast(keys)
  
      for callback in self.callbacks.values():
          callback.on_validation_epoch_end(batch_metrics)
  
      metrics = pytorch._reduce_metrics(
          self.context.distributed,
          batch_metrics=batch_metrics,
          keys=keys,
          metrics_reducers=pytorch._prepare_metrics_reducers(
              pytorch.Reducer.AVG, keys=keys
          ),
      )
      metrics.update(
          pytorch._convert_metrics_to_numpy(
              self.context.reduce_metrics(for_training=False)
          )
      )
  
      if self.context.distributed.size > 1 and any(
          util.is_overridden(
              c.on_validation_end, pytorch.PyTorchCallback
          )
          for c in self.callbacks.values()
      ):
          logger.debug(
              "Broadcasting metrics to all worker processes to execute a "
              "validation step end callback"
          )
          metrics = self.context.distributed.broadcast(metrics)
  
      for callback in self.callbacks.values():
          callback.on_validation_end(metrics)
  
      if self.is_chief:
          num_inputs *= self.context._mpu.data_parallel_world_size
          step_duration = time.time() - step_start_time
          logger.info(
              det.util.make_timing_log(
                  "validated",
                  step_duration,
                  num_inputs,
                  cast(int, self.num_validation_batches),
              )
          )
  
          if self.context.get_enable_tensorboard_logging():
              det.pytorch._log_tb_metrics(
                  self.context.get_tensorboard_writer(),
                  "val",
                  self.steps_completed,
                  metrics,
              )
  
      if not self.is_chief:
          return {}
  
      return {"num_inputs": num_inputs, "validation_metrics": metrics}
  
imports: import abc
  import contextlib
  import logging
  import os
  import pathlib
  import pickle
  import random
  import time
  import warnings
  from typing import Any, Callable, Dict, Iterator, List, Optional, Type, Union, cast
  import deepspeed
  import numpy
  import torch
  from deepspeed.runtime import dataloader
  import determined
  from determined import layers, pytorch, util, workload
  from determined.pytorch import deepspeed
  from determined.pytorch import dsat
label_desc: Use image node_id 9 for calling the DeepSpeedTrialController obj's underlying member method code with example usage: obj._compute_validation_metrics() and returns: dict, dict