class_docstr: A training engine hybrid pipeline, data, and model parallel training.
  
  This engine is created by ``deepspeed.initialize()`` when a :class:`PipelineModule`
  is provided.
class_docstr_tok: ["A","training","engine","hybrid","pipeline",",","data",",","and","model","parallel","training",".","This","engine","is","created","by","``","deepspeed.initialize","(",")","``","when","a",":","class",":","`","PipelineModule","`","is","provided","."]
code: def _reduce_outputs(self, outputs, reduce="avg", reduce_dp=True):
      if reduce is None:
          return outputs
  
      if reduce.lower() == "avg":
          # first sum over all microbatches
          if torch.is_tensor(outputs[0]):
              reduced = sum(outputs)
          else:
              assert isinstance(outputs, (list, tuple))
              reduced = [torch.zeros_like(o) for o in outputs[0]]
              for idx, out in outputs:
                  reduced[idx] += out
  
          # Average over the microbatches
          reduced = self._scale_loss_by_gas(reduced)
  
          # Average over DP groups
          if reduce_dp and self.is_data_parallel:
              if torch.is_tensor(reduced):
                  dist.all_reduce(
                      reduced, group=self.mpu.get_data_parallel_group()
                  )
                  reduced /= self.dp_world_size
              else:
                  for idx in range(len(reduced)):
                      dist.all_reduce(
                          reduced[idx],
                          group=self.mpu.get_data_parallel_group(),
                      )
                      reduced[idx] /= self.dp_world_size
  
          return reduced
      else:
          raise NotImplementedError(
              f"reduction type {reduce} not supported."
          )
  
imports: from types import MethodType
  import torch
  from deepspeed import comm
  from deepspeed.utils import logger
  from deepspeed.utils.timer import ThroughputTimer
  from deepspeed.accelerator import get_accelerator
  from deepspeed.runtime.bf16_optimizer import BF16_Optimizer
  from ..engine import DeepSpeedEngine, MEMORY_OPT_ALLREDUCE_SIZE
  from deepspeed.utils.timer import FORWARD_MICRO_TIMER, FORWARD_GLOBAL_TIMER, BACKWARD_MICRO_TIMER, BACKWARD_GLOBAL_TIMER, BACKWARD_INNER_MICRO_TIMER, BACKWARD_INNER_GLOBAL_TIMER, BACKWARD_REDUCE_MICRO_TIMER, BACKWARD_REDUCE_GLOBAL_TIMER, STEP_MICRO_TIMER, STEP_GLOBAL_TIMER
  from ..utils import PartitionedTensor
  from ..dataloader import RepeatingLoader
  from ..zero.config import ZeroStageEnum
  from ..activation_checkpointing import checkpointing
  from .module import PipelineModule, PipelineError
  from .None import p2p
  from .None import schedule
label_desc: Use image node_id 14 for calling the PipelineEngine obj's underlying member method code with example usage: obj._reduce_outputs(outputs, reduce, reduce_dp) and returns: outputs, reduced