class_docstr: CQL (derived from SAC).
class_docstr_tok: ["CQL","(","derived","from","SAC",")","."]
code: def training_step(self) -> ResultDict:
      # Collect SampleBatches from sample workers.
      with self._timers[SAMPLE_TIMER]:
          train_batch = synchronous_parallel_sample(
              worker_set=self.workers
          )
      train_batch = train_batch.as_multi_agent()
      self._counters[
          NUM_AGENT_STEPS_SAMPLED
      ] += train_batch.agent_steps()
      self._counters[NUM_ENV_STEPS_SAMPLED] += train_batch.env_steps()
  
      # Postprocess batch before we learn on it.
      post_fn = self.config.get("before_learn_on_batch") or (
          lambda b, *a: b
      )
      train_batch = post_fn(train_batch, self.workers, self.config)
  
      # Learn on training batch.
      # Use simple optimizer (only for multi-agent or tf-eager; all other
      # cases should use the multi-GPU optimizer, even if only using 1 GPU)
      if self.config.get("simple_optimizer") is True:
          train_results = train_one_step(self, train_batch)
      else:
          train_results = multi_gpu_train_one_step(self, train_batch)
  
      # Update target network every `target_network_update_freq` training steps.
      cur_ts = self._counters[
          NUM_AGENT_STEPS_TRAINED
          if self.config.count_steps_by == "agent_steps"
          else NUM_ENV_STEPS_TRAINED
      ]
      last_update = self._counters[LAST_TARGET_UPDATE_TS]
      if cur_ts - last_update >= self.config.target_network_update_freq:
          with self._timers[TARGET_NET_UPDATE_TIMER]:
              to_update = (
                  self.workers.local_worker().get_policies_to_train()
              )
              self.workers.local_worker().foreach_policy_to_train(
                  lambda p, pid: pid in to_update and p.update_target()
              )
          self._counters[NUM_TARGET_UPDATES] += 1
          self._counters[LAST_TARGET_UPDATE_TS] = cur_ts
  
      # Update remote workers's weights after learning on local worker
      # (only those policies that were actually trained).
      if self.workers.num_remote_workers() > 0:
          with self._timers[SYNCH_WORKER_WEIGHTS_TIMER]:
              self.workers.sync_weights(
                  policies=list(train_results.keys())
              )
  
      # Return all collected metrics for the iteration.
      return train_results
  
imports: import logging
  from typing import Optional, Type
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig, NotProvided
  from ray.rllib.algorithms.cql.cql_tf_policy import CQLTFPolicy
  from ray.rllib.algorithms.cql.cql_torch_policy import CQLTorchPolicy
  from ray.rllib.algorithms.sac.sac import SAC, SACConfig
  from ray.rllib.execution.rollout_ops import synchronous_parallel_sample
  from ray.rllib.execution.train_ops import multi_gpu_train_one_step, train_one_step
  from ray.rllib.policy.policy import Policy
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.deprecation import DEPRECATED_VALUE, deprecation_warning
  from ray.rllib.utils.framework import try_import_tf, try_import_tfp
  from ray.rllib.utils.metrics import LAST_TARGET_UPDATE_TS, NUM_AGENT_STEPS_SAMPLED, NUM_AGENT_STEPS_TRAINED, NUM_ENV_STEPS_SAMPLED, NUM_ENV_STEPS_TRAINED, NUM_TARGET_UPDATES, TARGET_NET_UPDATE_TIMER, SYNCH_WORKER_WEIGHTS_TIMER, SAMPLE_TIMER
  from ray.rllib.utils.typing import ResultDict
label_desc: Use image node_id 3 for calling the CQL obj's underlying member method code with example usage: obj.training_step() and returns: train_results