code: def my_build_collective_reduce(
      input_tensors,
      num_workers,
      collective_keys,
      reduction_op="Add",
      unary_op="Id",
      communication_hint="auto",
  ):
      """Build a subgraph that does one full all-reduce, using the collective Op.
  
      Args:
        input_tensors: tensors within a single worker graph that are to be reduced
          together; must be one per device.
        num_workers: total number of workers with identical independent graphs that
          will be doing this same reduction.  The reduction will actually include
          the corresponding tensors at all these workers.
        collective_keys: a CollectiveKeys object.
        reduction_op: string naming the reduction op.
        unary_op: string naming the unary final op.
        communication_hint: string providing hint to runtime for choosing collective
          implementation.
  
      Returns:
        An array of final tensors, one per device, computed by the full reduction.
  
      Raises:
        ValueError: There must be at least two tensors over all the workers.
      """
      group_size = len(input_tensors) * num_workers
      if group_size < 2:
          return input_tensors
      devices = [t.device for t in input_tensors]
      num_devices = len(devices)
      group_key = collective_keys.get_group_key(devices)
      instance_key = collective_keys.get_op_instance_key()
      subdiv_offsets = [
          0
      ]  # TODO(tucker): maybe support non-default subdiv spec
  
      def collective_all_reduce():
          """Call collective allreduce."""
          assert not context.executing_eagerly()
          out_tensors = []
          for d in range(num_devices):
              with ops.device(devices[d]):
                  # reduce_op = collective_ops.all_reduce(
                  #     input_tensors[d], group_size, group_key, instance_key, reduction_op,
                  #     unary_op, subdiv_offsets, communication_hint)
                  reduce_op = my_all_reduce(
                      input_tensors[d],
                      group_size,
                      group_key,
                      instance_key,
                      reduction_op,
                      unary_op,
                      subdiv_offsets,
                      communication_hint,
                  )
                  out_tensors.append(reduce_op)
          return out_tensors
  
      if context.executing_eagerly():
          # Collective ops will block unless they are executed concurrently such as in
          # a graph or a defun.
          collective_all_reduce = def_function.function(
              collective_all_reduce
          )
      return collective_all_reduce()
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import collections
  import enum
  import six
  from tensorflow.python.client import device_lib
  from tensorflow.python.distribute import cross_device_utils
  from tensorflow.python.distribute import device_util
  from tensorflow.python.distribute import reduce_util
  from tensorflow.python.distribute import values
  from tensorflow.python.eager import context
  from tensorflow.python.framework import kernels
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor_util
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import resource_variable_ops
  from tensorflow.python.platform import tf_logging
  from tensorflow.python.util.tf_export import tf_export
  from tensorflow.tools.docs import doc_controls
  import tensorflow
  from tensorflow.python.distribute.cross_device_ops import CollectiveAllReduce
  import tensorflow.python.distribute.cross_device_ops
  from byteps.tensorflow.ops import broadcast, _push_pull
  from byteps.tensorflow import local_rank, size
label_desc: Use image node_id 7 for calling a global function with example usage: my_build_collective_reduce(input_tensors, num_workers, collective_keys, reduction_op, unary_op, communication_hint) and returns: collective_all_reduce, input_tensors, out_tensors