code: def aggregate(self, phase):
      input, target = [], None
      for cid, ndarray_map in self.client_data[phase].items():
          for k, v in ndarray_map.items():
              if k == "input":
                  input.append((cid, tf.convert_to_tensor(v)))
              elif k == "target":
                  target = tf.convert_to_tensor(v)
              else:
                  invalidInputError(
                      False,
                      f"Invalid type of tensor map key: {k}, should be input/target",
                  )
      # TODO: to be consistent with Pytorch, custom API
  
      def sort_by_key(kv_tuple):
          return kv_tuple[0]
  
      input.sort(key=sort_by_key)
      tensor_list = []
      for cid, input_tensor in input:
          input_tensor.requires_grad = True
          tensor_list.append(input_tensor)
  
      if phase == "train":
          with tf.GradientTape(persistent=True) as tape:
              for tensor in tensor_list:
                  tape.watch(tensor)
              pred = self.model(tensor_list)
              loss = self.loss_fn(target, pred)
          gradients = tape.gradient(
              loss, self.model.trainable_variables
          )
          if self.optimizer is not None:
              self.optimizer.apply_gradients(
                  zip(gradients, self.model.trainable_variables)
              )
  
          for cid, input_tensor in input:
              x_grad = tape.gradient(loss, input_tensor)
              grad_map = {
                  "grad": x_grad.numpy(),
                  "loss": np.array(loss.numpy()),
              }
              self.server_data["train"][
                  cid
              ] = ndarray_map_to_tensor_map(grad_map)
  
          del tape  # manually delete the persistent GradientTape
      elif phase == "eval":
          pass
      elif phase == "pred":
          pred = self.model(tensor_list)
          for cid, input_tensor in input:
              pred_map = {"pred": pred.numpy()}
              self.server_data["pred"][cid] = ndarray_map_to_tensor_map(
                  pred_map
              )
      else:
          invalidInputError(
              False,
              f"Invalid phase: {phase}, should be train/eval/pred",
          )
  
file_path: BigDL/python/ppml/src/bigdl/ppml/fl/nn/tensorflow/aggregator.py
imports: import os
  from bigdl.ppml.utils.safepickle import SafePickle
  import logging
  import threading
  from bigdl.dllib.utils.log4Error import invalidInputError, invalidOperationError
  from bigdl.ppml.fl.nn.utils import ndarray_map_to_tensor_map
  from threading import Condition
  import numpy
  import tensorflow
label_desc: Use image node_id 6 for calling the Aggregator obj's underlying member method code with example usage: obj.aggregate(phase) and returns: kv_tuple