ExperienceReplay

class capymoa.ocl.strategy.ExperienceReplay

Bases: BatchClassifier, TrainTaskAware, TestTaskAware

Experience Replay.

Experience Replay (ER) [1] is a replay continual learning strategy.

• 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 [2].

• Not capymoa.ocl.base.TrainTaskAware or capymoa.ocl.base.TestTaskAware, but will proxy it to the wrapped learner.

>>> from capymoa.ann import Perceptron
>>> from capymoa.classifier import Finetune
>>> from capymoa.ocl.strategy import ExperienceReplay
>>> from capymoa.ocl.datasets import TinySplitMNIST
>>> from capymoa.ocl.evaluation import ocl_train_eval_loop
>>> import torch
>>> _ = torch.manual_seed(0)
>>> scenario = TinySplitMNIST()
>>> model = Perceptron(scenario.schema)
>>> learner = ExperienceReplay(Finetune(scenario.schema, model))
>>> results = ocl_train_eval_loop(
...     learner,
...     scenario.train_loaders(32),
...     scenario.test_loaders(32),
... )
>>> print(f"{results.accuracy_final*100:.1f}%")
33.0%

[1]

Rolnick, D., Ahuja, A., Schwarz, J., Lillicrap, T., & Wayne, G. (2019). Experience replay for continual learning. Advances in neural information processing systems, 32.

[2]

Jeffrey S. Vitter. 1985. Random sampling with a reservoir. ACM Trans. Math. Softw. 11, 1 (March 1985), 37–57.

__init__(

learner: BatchClassifier,

buffer_size: int = 200,

repeat: int = 1,

) → None

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

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

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

Train with a batch of instances.

Parameters:

• x – Batch of x_dtype valued feature vectors (batch_size, num_features)

• y – Batch of y_dtype valued labels (batch_size,).

on_test_task(task_id: int)

Called when testing on a task starts.

on_train_task(task_id: int)

Called when a new training task starts.

predict(instance: Instance) → int | None

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

Calls batch_predict_proba() with a batch of size 1.

train(instance: LabeledInstance) → None

Calls batch_train() with a batch of size 1.

device: torch.device = device(type='cpu')

Device on which the batch will be processed.

learner

The wrapped learner to be trained with experience replay.

random_seed: int

The random seed for reproducibility.

When implementing a classifier ensure random number generators are seeded.

schema: Schema

The schema representing the instances.

x_dtype: torch.dtype = torch.float32

Data type for the input features.

y_dtype: torch.dtype = torch.int64

Data type for the target value/labels.