code: def _all_tower_reduce(path, *tower_data):
      """Reduces stats across towers based on their stats-dict paths."""
      # TD-errors: Need to stay per batch item in order to be able to update
      # each item's weight in a prioritized replay buffer.
      if len(path) == 1 and path[0] == "td_error":
          return np.concatenate(tower_data, axis=0)
      elif tower_data[0] is None:
          return None
  
      if isinstance(path[-1], str):
          # TODO(sven): We need to fix this terrible dependency on `str.starts_with`
          #  for determining, how to aggregate these stats! As "num_..." might
          #  be a good indicator for summing, it will fail if the stats is e.g.
          #  `num_samples_per_sec" :)
          # Counter stats: Reduce sum.
          # if path[-1].startswith("num_"):
          #    return np.nansum(tower_data)
          # Min stats: Reduce min.
          if path[-1].startswith("min_"):
              return np.nanmin(tower_data)
          # Max stats: Reduce max.
          elif path[-1].startswith("max_"):
              return np.nanmax(tower_data)
      if np.isnan(tower_data).all():
          return np.nan
      # Everything else: Reduce mean.
      return np.nanmean(tower_data)
  
file_path: ray/rllib/utils/metrics/learner_info.py
imports: from collections import defaultdict
  import numpy
  import tree
  from typing import Dict
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID
  from ray.rllib.utils.typing import PolicyID
label_desc: Use image node_id 1 for calling a global function with example usage: _all_tower_reduce(path) and returns: np, np, np, None, np, np