StorageBackedReplayBuffer
StorageBackedReplayBuffer(capacity: int, observation_shape: tuple[int, ...], observation_dtype: np.dtype) -> NoneIntermediate base for replay buffers backed by a TransitionStorage.
Owns the storage and its lock, exposes the storage columns as pass-through properties, and provides the shared row copy that sample_into implementations use to fill the caller-provided output arrays. Sampling policy (uniform vs prioritized) stays in subclasses.
Initialize the storage-backed buffer.
Parameters:
capacity(int) – Maximum number of transitions to store.observation_shape(tuple[int, ...]) – Shape of the observations.observation_dtype(dtype) – The numpy dtype for storing observations.
Properties
actions
actions: np.ndarrayGet the stored actions column.
capacity
capacity: intGet the buffer's maximum number of stored transitions.
dones
dones: np.ndarrayGet the stored done-flags column.
idx
idx: intGet the next circular-buffer write position.
n_steps
n_steps: np.ndarrayGet the stored n-step-count column.
next_states
next_states: np.ndarrayGet the stored next-states column.
observation_dtype
observation_dtype: np.dtypeGet the numpy dtype used for stored observations.
observation_shape
observation_shape: tuple[int, ...]Get the shape of observations stored in the buffer.
Returns:
tuple[int, ...]– The observation shape tuple.
reward_shaping_metadata_spec
reward_shaping_metadata_spec: dict[str, np.dtype]Get the reward-shaping metadata columns stored in the buffer.
Returns:
dict[str, dtype]– A mapping of metadata key to numpy dtype.
rewards
rewards: np.ndarrayGet the stored rewards column.
size
size: intGet the current number of stored transitions.
states
states: np.ndarrayGet the stored states column.
Methods
__len__
__len__() -> intGet the current number of transitions in the buffer.
Returns:
int– The current size of the buffer.
Values
lock
lock = threading.Lock()