code: def _reduce_metrics(
      context: det.core.DistributedContext,
      batch_metrics: List,
      keys: Any,
      metrics_reducers: Dict[str, pytorch.Reducer],
  ) -> Dict[str, Any]:
      metrics = {}
      if len(batch_metrics):
          metrics = {
              name: pytorch._simple_reduce_metrics(
                  reducer=metrics_reducers[name],
                  metrics=np.stack(
                      [b[name] for b in batch_metrics], axis=0
                  ),
                  num_batches=None,
              )
              for name in keys or []
          }
  
      if context.size > 1:
          # If using distributed training, combine metrics across all processes.
          # Only the chief process will receive all the metrics.
          num_batches = len(batch_metrics)
          (
              combined_metrics,
              batches_per_process,
          ) = _combine_metrics_across_processes(
              context, metrics, num_batches
          )
          if context.rank == 0:
              # Only the chief collects all the metrics.
              assert combined_metrics is not None
              combined_metrics = _convert_metrics_to_numpy(
                  combined_metrics
              )
              metrics = {
                  name: pytorch._simple_reduce_metrics(
                      reducer=metrics_reducers[name],
                      metrics=combined_metrics[name],
                      num_batches=batches_per_process,
                  )
                  for name in keys or []
              }
          else:
              return {}
  
      return metrics
  
file_path: determined/harness/determined/pytorch/_metric_utils.py
imports: import logging
  from typing import Any, Dict, List, Optional, Tuple, Union, cast
  import numpy
  import torch
  import determined
  from determined import pytorch, util
label_desc: Use image node_id 7 for calling a global function with example usage: _reduce_metrics(context, batch_metrics, keys, metrics_reducers) and returns: metrics, dict