MAS#

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

Bases: BatchClassifier, Module, Handler

Memory Aware Synapses learner.

Memory Aware Synapses (MAS) is a regularisation-based continual learning strategy that estimates per-parameter importance from the sensitivity of the model’s output to small parameter perturbations, then penalises changes to parameters that were important for previous tasks [1].

Unlike EWC and SI, MAS estimates importance from the squared L2 norm of the model’s output rather than the task loss, so importance can be estimated without labels. We use a replay buffer to approximate the active task distribution when estimating importance.

Alternative implementations:

__init__(
schema: Schema,
model: Module,
optimiser: Optimizer,
lambda_: float,
alpha: float = 0.5,
buffer_capacity: int = 256,
importance_batch_size: int = 32,
device: device | None = None,
mask_test: bool = False,
mask_train: bool = False,
task_mask: Tensor | None = None,
) → None[source]#

Construct a MAS 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 MAS regularisation term.

  • alpha – EMA factor weighting the old importance estimate (alpha=1.0 means the importance never updates past its initial zero value; alpha=0.0 keeps only the most recent estimate). RWalk’s alpha weights the new estimate instead.

  • buffer_capacity – Replay window size used to estimate importance.

  • importance_batch_size – Mini-batch size used when estimating importance.

  • 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], or task-specific masking is requested without task_mask.

attach_with(
source: Dispatcher,
) → MAS[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.