class_docstr: Behavioral Cloning (derived from MARWIL).
  
  Simply uses MARWIL with beta force-set to 0.0.
class_docstr_tok: ["Behavioral","Cloning","(","derived","from","MARWIL",")",".","Simply","uses","MARWIL","with","beta","force-set","to","0.0","."]
code: def training_step(self) -> ResultDict:
      if not self.config._enable_new_api_stack:
          # Using ModelV2.
          return super().training_step()
      else:
          # Implement logic using RLModule and Learner API.
          # TODO (sven): Remove RolloutWorkers/EnvRunners for
          # datasets. Use RolloutWorker/EnvRunner only for
          # env stepping.
          # TODO (simon): Take care of sampler metrics: right
          # now all rewards are `nan`, which possibly confuses
          # the user that sth. is not right, although it is as
          # we do not step the env.
          with self._timers[SAMPLE_TIMER]:
              # Sampling from offline data.
              # TODO (simon): We have to remove the `RolloutWorker`
              # here and just use the already distributed `dataset`
              # for sampling. Only in online evaluation
              # `RolloutWorker/EnvRunner` should be used.
              if self.config.count_steps_by == "agent_steps":
                  train_batch = synchronous_parallel_sample(
                      worker_set=self.workers,
                      max_agent_steps=self.config.train_batch_size,
                  )
              else:
                  train_batch = synchronous_parallel_sample(
                      worker_set=self.workers,
                      max_env_steps=self.config.train_batch_size,
                  )
  
              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()
  
          # Updating the policy.
          train_results = self.learner_group.update_from_batch(
              batch=train_batch
          )
  
          # Synchronize weights.
          # As the results contain for each policy the loss and in addition the
          # total loss over all policies is returned, this total loss has to be
          # removed.
          policies_to_update = set(train_results.keys()) - {ALL_MODULES}
  
          global_vars = {
              "timestep": self._counters[NUM_AGENT_STEPS_SAMPLED],
          }
  
          with self._timers[SYNCH_WORKER_WEIGHTS_TIMER]:
              if self.workers.num_remote_workers() > 0:
                  self.workers.sync_weights(
                      from_worker_or_learner_group=self.learner_group,
                      policies=policies_to_update,
                      global_vars=global_vars,
                  )
              # Get weights from Learner to local worker.
              else:
                  self.workers.local_worker().set_weights(
                      self.learner_group.get_weights()
                  )
  
          return train_results
  
file_path: ray/rllib/algorithms/bc/bc.py
imports: from typing import Type, TYPE_CHECKING, Union
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  from ray.rllib.algorithms.bc.bc_catalog import BCCatalog
  from ray.rllib.algorithms.marwil.marwil import MARWIL, MARWILConfig
  from ray.rllib.core.rl_module.rl_module import SingleAgentRLModuleSpec
  from ray.rllib.execution.rollout_ops import synchronous_parallel_sample
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.metrics import ALL_MODULES, NUM_AGENT_STEPS_SAMPLED, NUM_ENV_STEPS_SAMPLED, SAMPLE_TIMER, SYNCH_WORKER_WEIGHTS_TIMER
  from ray.rllib.utils.typing import RLModuleSpec, ResultDict
label_desc: Use image node_id 2 for calling the BC obj's underlying member method code with example usage: obj.training_step() and returns: super, train_results