OakEmu-RL v0.1.0

Appearance

PrioritizedExperienceReplayBuffer ​

python
PrioritizedExperienceReplayBuffer(capacity: int, observation_shape: tuple[int, ...], observation_dtype: np.dtype, alpha: float = 0.6) -> None

Experience replay buffer with prioritization. Uses a SumTree for efficient sampling based on priorities.

Initialize the PrioritizedExperienceReplayBuffer.

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.
  • alpha (float) – Prioritization exponent.

Source

Methods ​

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 with maximum priority.

Parameters:

  • state (list[ndarray]) – List of states.
  • action (list[int]) – List of actions.
  • reward (list[float]) – List of rewards.
  • next_state (list[ndarray]) – List of next states.
  • done (list[bool]) – List of done flags.
  • n_steps (list[int]) – List of step counts.
  • reward_shaping_metadata (Mapping[str, ndarray | int | float | list[int | float]] | None) – Optional compact reward-shaping metadata columns.

Source

load ​

python
load(path: Path) -> None

Load the buffer state from a file using numpy's compressed format.

Parameters:

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

Source

load_from_path ​

python
load_from_path(path: Path) -> PrioritizedExperienceReplayBuffer

Create a new PrioritizedExperienceReplayBuffer from a saved file using numpy's compressed format.

Parameters:

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

Returns:

  • PrioritizedExperienceReplayBuffer – A new PrioritizedExperienceReplayBuffer instance.

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, np.ndarray]

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

Parameters:

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

Returns:

  • tuple[ndarray, ndarray] – Sampled (indices, versions) for later priority updates.

Source

save ​

python
save(path: Path) -> None

Save the buffer state to a file using numpy's compressed format.

Parameters:

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

Source

update_priorities ​

python
update_priorities(indices: np.ndarray, priorities: np.ndarray, versions: np.ndarray | None = None) -> int

Update priorities for sampled transitions.

Parameters:

  • indices (ndarray) – Indices of transitions to update.
  • priorities (ndarray) – New priorities (usually TD-errors).
  • versions (ndarray | None) – Versions of transitions at the time of sampling. If provided, updates will be skipped if versions don't match.

Returns:

  • int – The number of skipped updates due to version mismatch.

Source

Values ​

alpha ​

python
alpha = alpha

Source

epsilon ​

python
epsilon = 1e-06

Source

tree ​

python
tree = SumTree(capacity)

Source

versions ​

python
versions = np.zeros((capacity,), dtype=np.uint64)

Source