class_docstr: Short benchmarks (~110 steps) for DQN CartPole environment.
class_docstr_tok: ["Short","benchmarks","(","~110","steps",")","for","DQN","CartPole","environment","."]
docstr: Runs Dqn CartPole environment.
  
  Args:
    strategy: Strategy to use, None is a valid value.
    batch_size: Total batch size to use for the run.
    tf_function: If True tf.function is used.
    replay_buffer_max_length: Max length of the replay buffer.
    train_steps: Number of steps to run.
    log_steps: How often to log step statistics, e.g. step time.
docstr_tok: ["Runs","Dqn","CartPole","environment",".","Args",":","strategy",":","Strategy","to","use",",","None","is","a","valid","value",".","batch_size",":","Total","batch","size","to","use","for","the","run",".","tf_function",":","If","True","tf.function","is","used",".","replay_buffer_max_length",":","Max","length","of","the","replay","buffer",".","train_steps",":","Number","of","steps","to","run",".","log_steps",":","How","often","to","log","step","statistics",",","e.g",".","step","time","."]
code: def _run(
      self,
      strategy,
      batch_size=64,
      tf_function=True,
      replay_buffer_max_length=1000,
      train_steps=110,
      log_steps=10,
  ):
      """Runs Dqn CartPole environment.
  
      Args:
        strategy: Strategy to use, None is a valid value.
        batch_size: Total batch size to use for the run.
        tf_function: If True tf.function is used.
        replay_buffer_max_length: Max length of the replay buffer.
        train_steps: Number of steps to run.
        log_steps: How often to log step statistics, e.g. step time.
      """
      obs_spec = array_spec.BoundedArraySpec(
          [
              4,
          ],
          np.float32,
          -4.0,
          4.0,
      )
      action_spec = array_spec.BoundedArraySpec((), np.int64, 0, 1)
  
      py_env = random_py_environment.RandomPyEnvironment(
          obs_spec,
          action_spec,
          batch_size=1,
          reward_fn=lambda *_: np.random.randint(1, 10, 1),
      )
      env = tf_py_environment.TFPyEnvironment(py_env)
  
      policy = random_tf_policy.RandomTFPolicy(
          env.time_step_spec(), env.action_spec()
      )
  
      with distribution_strategy_utils.strategy_scope_context(strategy):
          q_net = q_network.QNetwork(
              env.time_step_spec().observation,
              env.action_spec(),
              fc_layer_params=(100,),
          )
  
          tf_agent = dqn_agent.DqnAgent(
              env.time_step_spec(),
              env.action_spec(),
              q_network=q_net,
              optimizer=tf.keras.optimizers.Adam(),
              td_errors_loss_fn=common.element_wise_squared_loss,
          )
          tf_agent.initialize()
          print(q_net.summary())
  
      replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
          data_spec=tf_agent.collect_data_spec,
          batch_size=1,
          max_length=replay_buffer_max_length,
      )
  
      driver = dynamic_step_driver.DynamicStepDriver(
          env, policy, [replay_buffer.add_batch]
      )
      if tf_function:
          driver.run = common.function(driver.run)
  
      for _ in range(replay_buffer_max_length):
          driver.run()
  
      check_values = ["QNetwork/EncodingNetwork/dense/bias:0"]
      initial_values = utils.get_initial_values(tf_agent, check_values)
  
      with distribution_strategy_utils.strategy_scope_context(strategy):
          dataset = replay_buffer.as_dataset(
              num_parallel_calls=tf.data.experimental.AUTOTUNE,
              sample_batch_size=batch_size,
              num_steps=2,
          )
          if strategy:
              iterator = iter(
                  strategy.experimental_distribute_dataset(dataset)
              )
          else:
              iterator = iter(dataset)
  
          def train_step(inputs):
              experience, _ = inputs
              return tf_agent.train(experience)
  
          if tf_function:
              train_step = common.function(train_step)
          self.run_and_report(
              train_step,
              strategy,
              batch_size,
              train_steps=train_steps,
              log_steps=log_steps,
              iterator=iterator,
          )
  
      utils.check_values_changed(tf_agent, initial_values, check_values)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import unittest
  import numpy
  from six.moves import range
  import tensorflow
  from tf_agents.agents.dqn import dqn_agent
  from tf_agents.benchmark import distribution_strategy_utils
  from tf_agents.benchmark import utils
  from tf_agents.drivers import dynamic_step_driver
  from tf_agents.environments import random_py_environment
  from tf_agents.environments import tf_py_environment
  from tf_agents.networks import q_network
  from tf_agents.policies import random_tf_policy
  from tf_agents.replay_buffers import tf_uniform_replay_buffer
  from tf_agents.specs import array_spec
  from tf_agents.utils import common
  from tensorflow.python import tf2
label_desc: Use image node_id 1 for calling the DqnCartPoleAgentBenchmark obj's underlying member method code with example usage: obj._run(strategy, batch_size, tf_function, replay_buffer_max_length, train_steps, log_steps) and returns: tf_agent