class_docstr: Program definition for D4PG.
code: def __init__(
      self,
      environment_factory: Callable[[bool], dm_env.Environment],
      network_factory: Callable[
          [specs.BoundedArray], Dict[str, snt.Module]
      ],
      accelerator: Optional[str] = None,
      num_actors: int = 1,
      num_caches: int = 0,
      environment_spec: Optional[specs.EnvironmentSpec] = None,
      batch_size: int = 256,
      prefetch_size: int = 4,
      min_replay_size: int = 1000,
      max_replay_size: int = 1000000,
      samples_per_insert: Optional[float] = 32.0,
      n_step: int = 5,
      sigma: float = 0.3,
      clipping: bool = True,
      discount: float = 0.99,
      policy_optimizer: Optional[snt.Optimizer] = None,
      critic_optimizer: Optional[snt.Optimizer] = None,
      target_update_period: int = 100,
      variable_update_period: int = 1000,
      max_actor_steps: Optional[int] = None,
      log_every: float = 10.0,
  ):
      if accelerator is not None and accelerator not in _ACCELERATORS:
          raise ValueError(
              f"Accelerator must be one of {_ACCELERATORS}, "
              f'not "{accelerator}".'
          )
  
      if not environment_spec:
          environment_spec = specs.make_environment_spec(
              environment_factory(False)
          )
  
      # TODO(mwhoffman): Make network_factory directly return the struct.
      # TODO(mwhoffman): Make the factory take the entire spec.
      def wrapped_network_factory(action_spec):
          networks_dict = network_factory(action_spec)
          networks = agent.D4PGNetworks(
              policy_network=networks_dict.get("policy"),
              critic_network=networks_dict.get("critic"),
              observation_network=networks_dict.get(
                  "observation", tf.identity
              ),
          )
          return networks
  
      self._environment_factory = environment_factory
      self._network_factory = wrapped_network_factory
      self._environment_spec = environment_spec
      self._sigma = sigma
      self._num_actors = num_actors
      self._num_caches = num_caches
      self._max_actor_steps = max_actor_steps
      self._log_every = log_every
      self._accelerator = accelerator
      self._variable_update_period = variable_update_period
  
      self._builder = agent.D4PGBuilder(
          # TODO(mwhoffman): pass the config dataclass in directly.
          # TODO(mwhoffman): use the limiter rather than the workaround below.
          agent.D4PGConfig(
              accelerator=accelerator,
              discount=discount,
              batch_size=batch_size,
              prefetch_size=prefetch_size,
              target_update_period=target_update_period,
              variable_update_period=variable_update_period,
              policy_optimizer=policy_optimizer,
              critic_optimizer=critic_optimizer,
              min_replay_size=min_replay_size,
              max_replay_size=max_replay_size,
              samples_per_insert=samples_per_insert,
              n_step=n_step,
              sigma=sigma,
              clipping=clipping,
          )
      )
  
imports: import copy
  from typing import Callable, Dict, Optional
  import acme
  from acme import specs
  from acme.agents.tf.d4pg import agent
  from acme.tf import savers
  from acme.utils import counting
  from acme.utils import loggers
  from acme.utils import lp_utils
  import dm_env
  import launchpad
  import reverb
  import sonnet
  import tensorflow
label_desc: Use image node_id 1 to create a new DistributedD4PG object  with example: obj = DistributedD4PG(environment_factory, network_factory, accelerator, num_actors, num_caches, environment_spec, batch_size, prefetch_size, min_replay_size, max_replay_size, samples_per_insert, n_step, sigma, clipping, discount, policy_optimizer, critic_optimizer, target_update_period, variable_update_period, max_actor_steps, log_every)