manager

Source code: tianshou/data/buffer/manager.py

---

class HERReplayBufferManager(buffer_list: list[HERReplayBuffer])

HERReplayBufferManager contains a list of HERReplayBuffer with exactly the same configuration.

These replay buffers have contiguous memory layout, and the storage space each buffer has is a shallow copy of the topmost memory.

Parameters:

buffer_list – a list of HERReplayBuffer needed to be handled.

See also

Please refer to ReplayBuffer for other APIs’ usage.

add(batch: RolloutBatchProtocol, buffer_ids: ndarray | list[int] | None = None) → tuple[ndarray, ndarray, ndarray, ndarray]

Add a batch of data into ReplayBufferManager.

Each of the data’s length (first dimension) must equal to the length of buffer_ids. By default buffer_ids is [0, 1, …, buffer_num - 1].

Return (current_index, episode_reward, episode_length, episode_start_index). If the episode is not finished, the return value of episode_length and episode_reward is 0.

save_hdf5(path: str, compression: str | None = None) → None

Save replay buffer to HDF5 file.

set_batch(batch: RolloutBatchProtocol) → None

Manually choose the batch you want the ReplayBuffer to manage.

update(buffer: HERReplayBuffer | ReplayBuffer) → ndarray

The ReplayBufferManager cannot be updated by any buffer.

class PrioritizedReplayBufferManager(buffer_list: Sequence[PrioritizedReplayBuffer])

PrioritizedReplayBufferManager contains a list of PrioritizedReplayBuffer with exactly the same configuration.

These replay buffers have contiguous memory layout, and the storage space each buffer has is a shallow copy of the topmost memory.

Parameters:

buffer_list – a list of PrioritizedReplayBuffer needed to be handled.

See also

Please refer to ReplayBuffer for other APIs’ usage.

class ReplayBufferManager(buffer_list: list[ReplayBuffer] | list[HERReplayBuffer])

ReplayBufferManager contains a list of ReplayBuffer with exactly the same configuration.

These replay buffers have contiguous memory layout, and the storage space each buffer has is a shallow copy of the topmost memory.

Parameters:

buffer_list – a list of ReplayBuffer needed to be handled.

See also

Please refer to ReplayBuffer for other APIs’ usage.

add(batch: RolloutBatchProtocol, buffer_ids: ndarray | list[int] | None = None) → tuple[ndarray, ndarray, ndarray, ndarray]

Add a batch of data into ReplayBufferManager.

Each of the data’s length (first dimension) must equal to the length of buffer_ids. By default buffer_ids is [0, 1, …, buffer_num - 1].

Return (current_index, episode_reward, episode_length, episode_start_index). If the episode is not finished, the return value of episode_length and episode_reward is 0.

next(index: int | ndarray) → ndarray

Return the index of next transition.

The index won’t be modified if it is the end of an episode.

prev(index: int | ndarray) → ndarray

Return the index of previous transition.

The index won’t be modified if it is the beginning of an episode.

reset(keep_statistics: bool = False) → None

Clear all the data in replay buffer and episode statistics.

sample_indices(batch_size: int | None) → ndarray

Get a random sample of index with size = batch_size.

Return all available indices in the buffer if batch_size is 0; return an empty numpy array if batch_size < 0 or no available index can be sampled.

Parameters:

batch_size – the number of indices to be sampled. If None, it will be set to the length of the buffer (i.e. return all available indices in a random order).

set_batch(batch: RolloutBatchProtocol) → None

Manually choose the batch you want the ReplayBuffer to manage.

unfinished_index() → ndarray

Return the index of unfinished episode.

update(buffer: ReplayBuffer) → ndarray

The ReplayBufferManager cannot be updated by any buffer.