code: def test_steps_are_reproducible(self):
      def propose_and_update_log_weights_fn(
          _, weighted_particles, seed=None
      ):
          proposed_particles = normal.Normal(
              loc=weighted_particles.particles, scale=1.0
          ).sample(seed=seed)
          return WeightedParticles(
              particles=proposed_particles,
              log_weights=weighted_particles.log_weights
              + normal.Normal(loc=-2.6, scale=0.1).log_prob(
                  proposed_particles
              ),
          )
  
      num_particles = 16
      initial_state = self.evaluate(
          WeightedParticles(
              particles=tf.random.normal(
                  [num_particles], seed=test_util.test_seed()
              ),
              log_weights=tf.fill(
                  [num_particles], -tf.math.log(float(num_particles))
              ),
          )
      )
  
      # Run a couple of steps.
      seeds = samplers.split_seed(
          test_util.test_seed(sampler_type="stateless"), n=2
      )
      kernel = SequentialMonteCarlo(
          propose_and_update_log_weights_fn=propose_and_update_log_weights_fn,
          resample_fn=weighted_resampling.resample_systematic,
          resample_criterion_fn=sequential_monte_carlo_kernel.ess_below_threshold,
      )
      state, results = kernel.one_step(
          state=initial_state,
          kernel_results=kernel.bootstrap_results(initial_state),
          seed=seeds[0],
      )
      state, results = kernel.one_step(
          state=state, kernel_results=results, seed=seeds[1]
      )
      state, results = self.evaluate(
          (
              tf.nest.map_structure(tf.convert_to_tensor, state),
              tf.nest.map_structure(tf.convert_to_tensor, results),
          )
      )
  
      # Re-initialize and run the same steps with the same seed.
      kernel2 = SequentialMonteCarlo(
          propose_and_update_log_weights_fn=propose_and_update_log_weights_fn,
          resample_fn=weighted_resampling.resample_systematic,
          resample_criterion_fn=sequential_monte_carlo_kernel.ess_below_threshold,
      )
      state2, results2 = kernel2.one_step(
          state=initial_state,
          kernel_results=kernel2.bootstrap_results(initial_state),
          seed=seeds[0],
      )
      state2, results2 = kernel2.one_step(
          state=state2, kernel_results=results2, seed=seeds[1]
      )
      state2, results2 = self.evaluate(
          (
              tf.nest.map_structure(tf.convert_to_tensor, state2),
              tf.nest.map_structure(tf.convert_to_tensor, results2),
          )
      )
  
      # Results should match.
      self.assertAllCloseNested(state, state2)
      self.assertAllCloseNested(results, results2)
  
file_path: probability/tensorflow_probability/python/experimental/mcmc/sequential_monte_carlo_kernel_test.py
imports: import functools
  import numpy
  import tensorflow.compat.v2
  from tensorflow_probability.python.distributions import normal
  from tensorflow_probability.python.experimental.mcmc import sequential_monte_carlo_kernel
  from tensorflow_probability.python.experimental.mcmc.sequential_monte_carlo_kernel import SequentialMonteCarlo
  from tensorflow_probability.python.experimental.mcmc.sequential_monte_carlo_kernel import weighted_resampling
  from tensorflow_probability.python.experimental.mcmc.sequential_monte_carlo_kernel import WeightedParticles
  from tensorflow_probability.python.internal import samplers
  from tensorflow_probability.python.internal import test_util
  from tensorflow_probability.python.math import gradient
label_desc: Use image node_id 1 for calling the _SequentialMonteCarloTest obj's underlying member method code with example usage: obj.test_steps_are_reproducible() and returns: WeightedParticles