ExperienceReplay#

class capymoa.ocl.strategy.ExperienceReplay[source]#

Experience Replay (ER) strategy for continual learning.

Uses a replay buffer to store past experiences and samples from it during training to mitigate catastrophic forgetting.
The replay buffer is implemented using reservoir sampling, which allows for uniform sampling over the entire stream [vitter1985].
Not capymoa.ocl.base.TrainTaskAware or capymoa.ocl.base.TestTaskAware, but will proxy it to the wrapped learner.

Jeffrey S. Vitter. 1985. Random sampling with a reservoir. ACM Trans. Math. Softw. 11, 1 (March 1985), 37–57. https://doi.org/10.1145/3147.3165

__init__( learner: BatchClassifier, buffer_size: int = 200, repeat: int = 1, ) → None[source]#

Initialize the Experience Replay strategy.

Parameters:

learner – The learner to be wrapped for experience replay.
buffer_size – The size of the replay buffer, defaults to 200.
repeat – The number of times to repeat the training data in each batch, defaults to 1.

batch_predict(x: Tensor) → Tensor[source]#

Predict the labels for a batch of instances.

Parameters:: x – Batch of x_dtype valued feature vectors (batch_size, num_features)
Returns:: Predicted batch of y_dtype valued labels (batch_size,).

batch_predict_proba(x: Tensor) → Tensor[source]#

Predict the probabilities of the classes for a batch of instances.

Parameters:: x – Batch of x_dtype valued feature vectors (batch_size, num_features)
Returns:: Batch of x_dtype valued predicted probabilities (batch_size, num_classes).

batch_train(x: Tensor, y: Tensor) → None[source]#

Train with a batch of instances.

Parameters:

on_test_task(task_id: int)[source]#: Called when testing on a task starts.

on_train_task(task_id: int)[source]#: Called when a new training task starts.

predict(instance: Instance) → int | None[source]#

Predict the label of an instance.

The base implementation calls predict_proba() and returns the label with the highest probability.

Parameters:: instance – The instance to predict the label for.
Returns:: The predicted label or None if the classifier is unable to make a prediction.

predict_proba( instance: Instance, ) → ndarray[Any, dtype[float64]] | None[source]#: Calls batch_predict_proba() with a batch of size 1.

train(instance: LabeledInstance) → None[source]#: Calls batch_train() with a batch of size 1.

device: torch.device = device(type='cpu')#: Device on which the batch will be processed.

random_seed: int#

The random seed for reproducibility.

When implementing a classifier ensure random number generators are seeded.

x_dtype: torch.dtype = torch.float32[source]#: Data type for the input features.

y_dtype: torch.dtype = torch.int64[source]#: Data type for the target value/labels.