class_docstr: Coordinator of n (possibly remote) Learner workers.
  
  Each Learner worker has a copy of the RLModule, the loss function(s), and
  one or more optimizers.
docstr: Initializes a LearnerGroup instance.
  
  Args:
      config: The AlgorithmConfig object to use to configure this LearnerGroup.
          Call the `resources(num_learner_workers=...)` method on your config to
          specify the number of learner workers to use.
          Call the same method with arguments `num_cpus_per_learner_worker` and/or
          `num_gpus_per_learner_worker` to configure the compute used by each
          Learner worker in this LearnerGroup.
          Call the `training(learner_class=...)` method on your config to specify,
          which exact Learner class to use.
          Call the `rl_module(rl_module_spec=...)` method on your config to set up
          the specifics for your RLModule to be used in each Learner.
      module_spec: If not already specified in `config`, a separate overriding
          RLModuleSpec may be provided via this argument.
      max_queue_len: The maximum number of batches to queue up if doing
          async_update. If the queue is full it will evict the oldest batch first.
docstr_tok: ["Initializes","a","LearnerGroup","instance",".","Args",":","config",":","The","AlgorithmConfig","object","to","use","to","configure","this","LearnerGroup",".","Call","the","`","resources","(","num_learner_workers=","...",")","`","method","on","your","config","to","specify","the","number","of","learner","workers","to","use",".","Call","the","same","method","with","arguments","`","num_cpus_per_learner_worker","`","and\/or","`","num_gpus_per_learner_worker","`","to","configure","the","compute","used","by","each","Learner","worker","in","this","LearnerGroup",".","Call","the","`","training","(","learner_class=","...",")","`","method","on","your","config","to","specify",",","which","exact","Learner","class","to","use",".","Call","the","`","rl_module","(","rl_module_spec=","...",")","`","method","on","your","config","to","set","up","the","specifics","for","your","RLModule","to","be","used","in","each","Learner",".","module_spec",":","If","not","already","specified","in","`","config","`",",","a","separate","overriding","RLModuleSpec","may","be","provided","via","this","argument",".","max_queue_len",":","The","maximum","number","of","batches","to","queue","up","if","doing","async_update",".","If","the","queue","is","full","it","will","evict","the","oldest","batch","first","."]
code: def __init__(
      self,
      *,
      config: AlgorithmConfig = None,  # TODO (sven): Make this arg mandatory.
      module_spec: Optional[RLModuleSpec] = None,
      max_queue_len: int = 20,
      # Deprecated args.
      learner_spec=None,
  ):
      """Initializes a LearnerGroup instance.
  
      Args:
          config: The AlgorithmConfig object to use to configure this LearnerGroup.
              Call the `resources(num_learner_workers=...)` method on your config to
              specify the number of learner workers to use.
              Call the same method with arguments `num_cpus_per_learner_worker` and/or
              `num_gpus_per_learner_worker` to configure the compute used by each
              Learner worker in this LearnerGroup.
              Call the `training(learner_class=...)` method on your config to specify,
              which exact Learner class to use.
              Call the `rl_module(rl_module_spec=...)` method on your config to set up
              the specifics for your RLModule to be used in each Learner.
          module_spec: If not already specified in `config`, a separate overriding
              RLModuleSpec may be provided via this argument.
          max_queue_len: The maximum number of batches to queue up if doing
              async_update. If the queue is full it will evict the oldest batch first.
      """
      if learner_spec is not None:
          deprecation_warning(
              old="LearnerGroup(learner_spec=...)",
              new="config = AlgorithmConfig().[resources|training|rl_module](...); "
              "LearnerGroup(config=config)",
              error=True,
          )
      if config is None:
          raise ValueError(
              "LearnerGroup constructor must be called with a `config` arg! "
              "Pass in a `ray.rllib.algorithms.algorithm_config::AlgorithmConfig` "
              "object with the proper settings configured."
          )
  
      # scaling_config = learner_spec.learner_group_scaling_config
      self.config = config
  
      learner_class = self.config.learner_class
      module_spec = module_spec or self.config.get_marl_module_spec()
  
      self._learner = None
      self._workers = None
      # If a user calls self.shutdown() on their own then this flag is set to true.
      # When del is called the backend executor isn't shutdown twice if this flag is
      # true. the backend executor would otherwise log a warning to the console from
      # ray train.
      self._is_shut_down = False
  
      # The callable to use to figure out whether a (single-agent) sub-module is
      # trainable (via its ModuleID and the train batch) or not.
      self._should_module_be_updated_fn = (
          _default_should_module_be_updated_fn
      )
  
      # How many timesteps had to be dropped due to a full input queue?
      self._ts_dropped = 0
  
      # A single local Learner.
      if not self.is_remote:
          self._learner = learner_class(
              config=config, module_spec=module_spec
          )
          self._learner.build()
          self._worker_manager = None
      # N remote Learner workers.
      else:
          backend_config = _get_backend_config(learner_class)
          backend_executor = BackendExecutor(
              backend_config=backend_config,
              num_workers=self.config.num_learner_workers,
              # TODO (sven): Cannot set both `num_cpus_per_learner_worker`>1 and
              #  `num_gpus_per_learner_worker`>0! Users must set one or the other due
              #  to issues with placement group fragmentation. See
              #  https://github.com/ray-project/ray/issues/35409 for more details.
              num_cpus_per_worker=(
                  self.config.num_cpus_per_learner_worker
                  if not self.config.num_gpus_per_learner_worker
                  else 0
              ),
              num_gpus_per_worker=self.config.num_gpus_per_learner_worker,
              max_retries=0,
          )
          backend_executor.start(
              train_cls=learner_class,
              train_cls_kwargs={
                  "config": config,
                  "module_spec": module_spec,
              },
          )
          self._backend_executor = backend_executor
  
          self._workers = [
              w.actor for w in backend_executor.worker_group.workers
          ]
  
          # Run the neural network building code on remote workers.
          ray.get([w.build.remote() for w in self._workers])
  
          self._worker_manager = FaultTolerantActorManager(
              self._workers,
              # TODO (sven): This probably works even without any restriction
              #  (allowing for any arbitrary number of requests in-flight). Test with
              #  3 first, then with unlimited, and if both show the same behavior on
              #  an async algo, remove this restriction entirely.
              max_remote_requests_in_flight_per_actor=3,
          )
          # Counters for the tags for asynchronous update requests that are
          # in-flight. Used for keeping trakc of and grouping together the results of
          # requests that were sent to the workers at the same time.
          self._update_request_tags = Counter()
          self._additional_update_request_tags = Counter()
  
imports: from collections import defaultdict, Counter
  from functools import partial
  import pathlib
  from typing import Any, Callable, Dict, List, Optional, Set, Type, Union
  import uuid
  import ray
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  from ray.rllib.core.learner.learner import Learner
  from ray.rllib.core.learner.reduce_result_dict_fn import _reduce_mean_results
  from ray.rllib.core.rl_module.rl_module import SingleAgentRLModuleSpec, RLMODULE_STATE_DIR_NAME
  from ray.rllib.policy.sample_batch import MultiAgentBatch
  from ray.rllib.utils.actor_manager import FaultTolerantActorManager
  from ray.rllib.utils.deprecation import Deprecated, deprecation_warning
  from ray.rllib.utils.minibatch_utils import ShardBatchIterator, ShardEpisodesIterator
  from ray.rllib.utils.numpy import convert_to_numpy
  from ray.rllib.utils.typing import EpisodeType, ModuleID, ResultDict, RLModuleSpec, ShouldModuleBeUpdatedFn
  from ray.train._internal.backend_executor import BackendExecutor
  from ray.tune.utils.file_transfer import sync_dir_between_nodes
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 1 to create a new LearnerGroup object  with example: obj = LearnerGroup()