OakEmu-RL v0.1.0

Appearance

ReplayBuffer ​

Abstract base class for experience replay buffers.

Replay buffers store pre-computed n-step transitions as flat tuples rather than raw sequences. The NStepBuffer in the Actor accumulates single-step transitions and produces n-step transitions with already-discounted rewards. Each stored transition contains (state, action, reward, next_state, done, n_steps) where:

  • state: The starting state of the n-step window.
  • action: The action taken at the starting state.
  • reward: The accumulated discounted reward (r₀ + γr₁ + γ²r₂ + ...).
  • next_state: The state n steps ahead, used for bootstrapping.
  • done: Whether bootstrapping should happen (True for terminated, False for truncated).
  • n_steps: The actual number of steps accumulated, used by the Learner to apply the correct γⁿ discount when computing Q-targets.

Source

Properties ​

observation_dtype ​

python
observation_dtype: np.dtype

Get the dtype used for stored observations.

Source

observation_shape ​

python
observation_shape: tuple[int, ...]

Get the shape of observations stored in the buffer.

Returns:

  • tuple[int, ...] – The observation shape tuple.

Source

reward_shaping_metadata_spec ​

python
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.

Source

Methods ​

__len__ ​

python
__len__() -> int

Get the current number of transitions in the buffer.

Returns:

  • int – The current size of the buffer.

Source

add ​

python
add(state: list[np.ndarray], action: list[int], reward: list[float], next_state: list[np.ndarray], done: list[bool], n_steps: list[int], reward_shaping_metadata: Mapping[str, np.ndarray | int | float | list[int | float]] | None = None) -> None

Add a batch of transitions to the buffer.

Parameters:

  • state (list[ndarray]) – The starting states of the n-step windows.
  • action (list[int]) – The actions taken at the starting states.
  • reward (list[float]) – The accumulated discounted n-step rewards.
  • next_state (list[ndarray]) – The states n steps ahead for bootstrapping.
  • done (list[bool]) – Whether each episode terminated (True prevents bootstrapping).
  • n_steps (list[int]) – The actual number of steps in each transition.
  • reward_shaping_metadata (Mapping[str, ndarray | int | float | list[int | float]] | None) – Optional compact numeric metadata columns that must travel with the sampled transition for learner-side reward-shaper updates.

Source

load ​

python
load(path: Path) -> None

Load the buffer state from a file.

Parameters:

  • path (Path) – Path to the save file.

Source

sample_into ​

python
sample_into(batch_size: int, *, arrays_out: HostBatchArrays, reward_shaping_metadata_out: dict[str, np.ndarray] | None = None, alpha: float = 0.6, beta: float = 0.4) -> tuple[np.ndarray | None, np.ndarray | None]

Sample a batch of transitions directly into caller-provided output arrays.

Implementations should fill the provided arrays in-place. Prioritized replay buffers also populate arrays_out.weights and return sampled indices/versions so the learner can update priorities later.

Parameters:

  • batch_size (int) – Number of transitions to sample.
  • arrays_out (HostBatchArrays) – Output arrays for the sampled batch, shaped (batch_size, ...).
  • reward_shaping_metadata_out (dict[str, ndarray] | None) – Optional output arrays for compact reward-shaping metadata.
  • alpha (float) – Prioritization exponent for prioritized buffers.
  • beta (float) – Importance-sampling exponent for prioritized buffers.

Returns:

  • tuple[ndarray | None, ndarray | None] – (indices, versions) for prioritized buffers, otherwise (None, None).

Source

save ​

python
save(path: Path) -> None

Save the buffer state to a file.

Parameters:

  • path (Path) – Path to the save file.

Source