docstr: Helper function for testing correct variable updating and restoring.
docstr_tok: ["Helper","function","for","testing","correct","variable","updating","and","restoring","."]
code: def VerifyTrainAndRestore(self):
      """Helper function for testing correct variable updating and restoring."""
      batch_size = 2
      seq_len = 2
      observations = tensor_spec.sample_spec_nest(
          self._observation_spec, outer_dims=(batch_size, seq_len)
      )
      actions = tensor_spec.sample_spec_nest(
          self._action_spec, outer_dims=(batch_size, seq_len)
      )
      rewards = tf.constant([[10, 10], [20, 20]], dtype=tf.float32)
      discounts = tf.constant(
          [[0.9, 0.9], [0.9, 0.9]], dtype=tf.float32
      )
      experience = trajectory.first(
          observation=observations,
          action=actions,
          policy_info=(),
          reward=rewards,
          discount=discounts,
      )
      strategy = tf.distribute.get_strategy()
      with strategy.scope():
          q_net = critic_network.CriticNetwork(
              (self._observation_spec, self._action_spec)
          )
          agent = qtopt_agent.QtOptAgent(
              self._time_step_spec,
              self._action_spec,
              q_network=q_net,
              optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
              init_mean_cem=self._mean,
              init_var_cem=self._var,
              num_samples_cem=self._num_samples,
              actions_sampler=self._sampler,
              in_graph_bellman_update=True,
          )
      loss_before_train = agent.loss(experience).loss
      # Check loss is stable.
      self.assertEqual(loss_before_train, agent.loss(experience).loss)
      # Train 1 step, verify that loss is decreased for the same input.
      agent.train(experience)
      loss_after_train = agent.loss(experience).loss
      self.assertLessEqual(loss_after_train, loss_before_train)
      # Assert loss evaluation is still stable, e.g. deterministic.
      self.assertLessEqual(
          loss_after_train, agent.loss(experience).loss
      )
      # Save checkpoint
      ckpt_dir = self.create_tempdir()
      checkpointer = common.Checkpointer(ckpt_dir=ckpt_dir, agent=agent)
      global_step = tf.constant(1)
      checkpointer.save(global_step)
      # Assign all vars to 0.
      for var in tf.nest.flatten(agent.variables):
          var.assign(tf.zeros_like(var))
      loss_after_zero = agent.loss(experience).loss
      self.assertEqual(loss_after_zero, agent.loss(experience).loss)
      self.assertNotEqual(loss_after_zero, loss_after_train)
      # Restore
      checkpointer._checkpoint.restore(
          checkpointer._manager.latest_checkpoint
      )
      loss_after_restore = agent.loss(experience).loss
      self.assertNotEqual(loss_after_restore, loss_after_zero)
      self.assertEqual(loss_after_restore, loss_after_train)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from absl.testing import parameterized
  import numpy
  import tensorflow
  from tf_agents.agents.ddpg import critic_network
  from tf_agents.agents.qtopt import qtopt_agent
  from tf_agents.networks import network
  from tf_agents.policies.samplers import qtopt_cem_actions_sampler_continuous
  from tf_agents.specs import tensor_spec
  from tf_agents.trajectories import policy_step
  from tf_agents.trajectories import test_utils
  from tf_agents.trajectories import time_step
  from tf_agents.trajectories import trajectory
  from tf_agents.utils import common
label_desc: Use image node_id 7 for calling the QtoptAgentTest obj's underlying member method code with example usage: obj.VerifyTrainAndRestore() without return types