OakEmu-RL v0.1.0

Appearance

RewardShaper ​

python
RewardShaper(context: RewardShapingContext) -> None

Base class for trainer-owned reward shapers.

Initialize the reward shaper.

Parameters:

  • context (RewardShapingContext) – Shared shaping runtime context.

Source

Properties ​

context ​

python
context: RewardShapingContext

Get the shaping runtime context.

Returns:

  • RewardShapingContext – The shared shaping context.

Source

metric_namespace ​

python
metric_namespace: str

Get the metric namespace used for this shaper's custom metrics.

Returns:

  • str – The metric namespace.

Source

Methods ​

add_averaged_metric ​

python
add_averaged_metric(name: str, value: float) -> None

Log a shaper-specific averaged metric.

The value is buffered and written out at the step passed to the next flush_metrics.

Parameters:

  • name (str) – The metric name suffix.
  • value (float) – The metric value.

Source

add_episode_metric ​

python
add_episode_metric(name: str, env_id: int, value: float) -> None

Accumulate a shaper-specific metric to be logged once per episode.

Parameters:

  • name (str) – The episode metric name suffix.
  • env_id (int) – The environment ID.
  • value (float) – The value to add to the episode total.

Source

add_scalar_metric ​

python
add_scalar_metric(name: str, value: float) -> None

Log a shaper-specific scalar metric.

The value is buffered and written out at the step passed to the next flush_metrics.

Parameters:

  • name (str) – The metric name suffix.
  • value (float) – The metric value.

Source

flush_episode_metrics ​

python
flush_episode_metrics(env_id: int) -> None

Emit and clear accumulated episode metrics for one environment.

Parameters:

  • env_id (int) – The environment ID.

Source

flush_metrics ​

python
flush_metrics(step: int) -> None

Emit any throttled scalar and averaged metrics accumulated since the last flush.

Parameters:

  • step (int) – The step at which to log.

Source

learn ​

python
learn(batch: RewardShapingBatch) -> None

Update learner-owned shaper state from a batch.

Stateless or actor-only shapers can keep the default no-op implementation.

Parameters:

  • batch (RewardShapingBatch) – The learner batch to consume.

Source

load_state_dict ​

python
load_state_dict(state: dict[str, Any]) -> None

Restore shaper state from a checkpoint dictionary.

Parameters:

  • state (dict[str, Any]) – The serialised state dictionary.

Source

metadata_key ​

python
metadata_key(name: str) -> str

Build a namespaced learner-metadata key for this shaper.

Parameters:

  • name (str) – The metadata name suffix.

Returns:

  • str – The full metadata key.

Source

on_episode_end ​

python
on_episode_end(env_id: int, step: int) -> None

Notify the shaper that an environment episode has ended.

Parameters:

  • env_id (int) – The environment ID.
  • step (int) – The global step at episode end.

Source

on_episode_start ​

python
on_episode_start(env_id: int, initial_observation: np.ndarray) -> None

Notify the shaper that an environment has started a fresh episode.

Parameters:

  • env_id (int) – The environment ID.
  • initial_observation (ndarray) – The reset observation for the new episode.

Source

reset_episode_metrics ​

python
reset_episode_metrics(env_id: int) -> None

Clear any accumulated episode metrics for one environment.

Parameters:

  • env_id (int) – The environment ID.

Source

shape ​

python
shape(step: RewardShapingStep) -> float

Shape a reward for one environment step.

Parameters:

  • step (RewardShapingStep) – The step information, including observations and reward.

Returns:

  • float – The new reward value after this shaper has run.

Source

shape_batch ​

python
shape_batch(steps: Sequence[RewardShapingStep]) -> list[float]

Shape a batch of environment steps.

The default implementation delegates to shape one item at a time. Shapers with efficient batched inference can override this.

Parameters:

  • steps (Sequence[RewardShapingStep]) – The steps to shape.

Returns:

  • list[float] – A shaped reward for each input step.

Source

shape_batch_with_metadata ​

python
shape_batch_with_metadata(steps: Sequence[RewardShapingStep]) -> list[RewardShapingOutput]

Shape a batch of environment steps and emit learner metadata for each step.

Shapers that need learner-side routing information can override this instead of shape_batch.

Parameters:

  • steps (Sequence[RewardShapingStep]) – The steps to shape.

Returns:

  • list[RewardShapingOutput] – A shaped reward plus optional learner metadata for each step.

Source

state_dict ​

python
state_dict() -> dict[str, Any]

Capture serializable shaper state for checkpoints.

Returns:

  • dict[str, Any] – A serializable state dictionary.

Source

Values ​

metric_namespace_override ​

python
metric_namespace_override: str | None = None

Source

reward_stream ​

python
reward_stream: RewardStream = RewardStream.EXTRINSIC

Source