RWalk#

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

Bases: BatchClassifier, Module, Handler

Riemannian Walk (RWalk) learner.

RWalk [1] is a regularisation-based continual learning strategy that, like EWC, augments the task loss with a weighted quadratic penalty on parameter changes. The penalty weights combine an exponential moving average of squared gradients with trajectory scores that estimate how sensitive the loss is to parameter updates, accumulated online between periodic checkpoints.

Alternative implementations:

__init__(
schema: Schema,
model: Module,
optimiser: Optimizer,
lambda_: float,
alpha: float = 0.9,
delta_t: int = 10,
device: device | None = None,
mask_test: bool = False,
mask_train: bool = False,
task_mask: Tensor | None = None,
) → None[source]#

Construct an RWalk learner.

Parameters:

  • schema – Stream schema used by the classifier interface.

  • model – Torch model that outputs class logits.

  • optimiser – Optimiser used to update model parameters.

  • lambda – Weight of the RWalk regularisation term.

  • alpha – EMA decay factor weighting the new gradient estimate (alpha=1.0 keeps only the most recent estimate). MAS’s alpha weights the old estimate instead.

  • delta_t – Number of training steps between score checkpoints.

  • device – Compute device.

  • mask_test – Whether to apply per-task masking during testing. This is a task incremental scenario.

  • mask_train – Whether to apply per-task masking during training. This is also known as the labels trick.

  • task_mask – Optional per-task mask applied to output logits.

Raises:

ValueError – If lambda_ is negative, alpha is outside [0, 1], delta_t is less than 1, or task-specific masking is requested without task_mask.

attach_with(
source: Dispatcher,
) → RWalk[source]#

Attach this sink to an event source.

Implementations should call capymoa.ocl.events.Dispatcher.subscribe() for each event type the sink needs to handle.

Parameters:

dispatcher – The source this sink should subscribe to.

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:

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

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

classmethod from_params(
schema: Any = None,
params: dict[str, Any] | None = None,
random_seed: int = 1,
) → Any[source]#

Construct an instance from parameters produced by get_params.

get_params() → dict[str, Any][source]#

Return the hyper-parameters captured from the constructor.

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[tuple[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.

T_destination = ~T_destination#
call_super_init: bool = False#
device: torch.device = device(type='cpu')#

Device on which the batch will be processed.

dump_patches: bool = False#
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.

training: bool#
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.