class_docstr: Tests insertion and (weighted) sampling of the PrioritizedReplayBuffer.
class_docstr_tok: ["Tests","insertion","and","(","weighted",")","sampling","of","the","PrioritizedReplayBuffer","."]
docstr: Tests adding, sampling and eviction of sequences.
docstr_tok: ["Tests","adding",",","sampling","and","eviction","of","sequences","."]
code: def test_sequences_unit(self):
      """Tests adding, sampling and eviction of sequences."""
      # We do not test the mathematical correctness of our prioritization
      # here but rather if it works together with sequence mode at all
  
      buffer = PrioritizedReplayBuffer(
          capacity=10, storage_unit="sequences"
      )
  
      batches = [
          SampleBatch(
              {
                  SampleBatch.T: i * [np.random.random((4,))],
                  SampleBatch.ACTIONS: i * [np.random.choice([0, 1])],
                  SampleBatch.REWARDS: i * [np.random.rand()],
                  SampleBatch.TERMINATEDS: i
                  * [np.random.choice([False, True])],
                  SampleBatch.TRUNCATEDS: i
                  * [np.random.choice([False, True])],
                  SampleBatch.SEQ_LENS: [i],
                  "batch_id": i * [i],
              }
          )
          for i in range(1, 4)
      ]
  
      # Add some batches with sequences of low priority
      for batch in batches:
          buffer.add(batch, weight=0.01)
  
      # Add two high priority sequences
      buffer.add(
          SampleBatch(
              {
                  SampleBatch.T: 4 * [np.random.random((4,))],
                  SampleBatch.ACTIONS: 4 * [np.random.choice([0, 1])],
                  SampleBatch.REWARDS: 4 * [np.random.rand()],
                  SampleBatch.TERMINATEDS: 4
                  * [np.random.choice([False, True])],
                  SampleBatch.TRUNCATEDS: 4
                  * [np.random.choice([False, True])],
                  SampleBatch.SEQ_LENS: [2, 2],
                  "batch_id": 4 * [4],
              }
          ),
          weight=1,
      )
  
      num_sampled_dict = {_id: 0 for _id in range(1, 5)}
      num_samples = 200
      for i in range(num_samples):
          sample = buffer.sample(1, beta=self.beta)
          _id = sample["batch_id"][0]
          assert len(sample[SampleBatch.SEQ_LENS]) == 1
          num_sampled_dict[_id] += 1
  
      # Out of five sequences, we want to sequences from the last batch to
      # be sampled almost always
      assert np.allclose(
          np.array(list(num_sampled_dict.values())) / num_samples,
          [0.1, 0.1, 0.1, 0.8],
          atol=0.2,
      )
  
      # Add another batch to evict
      buffer.add(
          SampleBatch(
              {
                  SampleBatch.T: 5 * [np.random.random((4,))],
                  SampleBatch.ACTIONS: 5 * [np.random.choice([0, 1])],
                  SampleBatch.REWARDS: 5 * [np.random.rand()],
                  SampleBatch.TERMINATEDS: 5
                  * [np.random.choice([False, True])],
                  SampleBatch.TRUNCATEDS: 5
                  * [np.random.choice([False, True])],
                  SampleBatch.SEQ_LENS: [5],
                  "batch_id": 5 * [5],
              }
          ),
          weight=1,
      )
  
      # After adding 1 more batch, eviction has started with 15
      # timesteps added in total
      assert len(buffer) == 5
      assert buffer._num_timesteps_added == sum(range(1, 6))
      assert buffer._num_timesteps_added_wrap == 5
      assert buffer._next_idx == 1
      assert buffer._eviction_started is True
  
      num_sampled_dict = {_id: 0 for _id in range(1, 6)}
      num_samples = 200
      for i in range(num_samples):
          sample = buffer.sample(1, beta=self.beta)
          _id = sample["batch_id"][0]
          assert len(sample[SampleBatch.SEQ_LENS]) == 1
          num_sampled_dict[_id] += 1
  
      # Out of all six sequences, we want sequences from batches 4 and 5
      # to be sampled with equal probability
      assert np.allclose(
          np.array(list(num_sampled_dict.values())) / num_samples,
          [0, 0, 0, 0.5, 0.5],
          atol=0.25,
      )
  
file_path: ray/python/ray/rllib/utils/replay_buffers/tests/test_prioritized_replay_buffer_replay_buffer_api.py
imports: from collections import Counter
  import numpy
  import unittest
  from ray.rllib.utils.replay_buffers.prioritized_replay_buffer import PrioritizedReplayBuffer
  from ray.rllib.policy.sample_batch import SampleBatch, MultiAgentBatch, concat_samples
  from ray.rllib.utils.test_utils import check
label_desc: Use image node_id 7 for calling the TestPrioritizedReplayBuffer obj's underlying member method code with example usage: obj.test_sequences_unit() without return types