code: def init(self, aggregators):
          
  
          assert len(aggregators) == self.num_aggregation_workers, aggregators
          if len(self.remote_evaluators) < self.num_aggregation_workers:
              raise ValueError(
                  "The number of aggregation workers should not exceed the "
                  "number of total evaluation workers ({} vs {})".format(
                      self.num_aggregation_workers, len(self.remote_evaluators)))
  
          assigned_evaluators = collections.defaultdict(list)
          for i, ev in enumerate(self.remote_evaluators):
              assigned_evaluators[i % self.num_aggregation_workers].append(ev)
  
          self.workers = aggregators
          for i, worker in enumerate(self.workers):
              worker.init.remote(
                  self.broadcasted_weights, assigned_evaluators[i],
                  self.max_sample_requests_in_flight_per_worker,
                  self.replay_proportion, self.replay_buffer_num_slots,
                  self.train_batch_size, self.sample_batch_size)
  
          self.agg_tasks = TaskPool()
          for agg in self.workers:
              agg.set_weights.remote(self.broadcasted_weights)
              self.agg_tasks.add(agg, agg.get_train_batches.remote())
  
          self.initialized = True
docs: Deferred init so that we can pass in previously created workers.