Source code for avalanche.training.supervised.strategy_wrappers

################################################################################
# Copyright (c) 2021 ContinualAI.                                              #
# Copyrights licensed under the MIT License.                                   #
# See the accompanying LICENSE file for terms.                                 #
#                                                                              #
# Date: 01-12-2020                                                             #
# Author(s): Antonio Carta, Andrea Cossu                                       #
# E-mail: contact@continualai.org                                              #
# Website: avalanche.continualai.org                                           #
################################################################################
from typing import Callable, Optional, Sequence, List, Union
import torch
from torch.nn.parameter import Parameter

from torch import sigmoid
from torch.nn import Module, CrossEntropyLoss
from torch.optim import Optimizer
from avalanche.models.packnet import PackNetModel, PackNetModule, PackNetPlugin

from avalanche.models.pnn import PNN
from avalanche.training.plugins.evaluation import (
    default_evaluator,
    default_loggers,
)
from avalanche.training.plugins import (
    SupervisedPlugin,
    CWRStarPlugin,
    ReplayPlugin,
    GenerativeReplayPlugin,
    TrainGeneratorAfterExpPlugin,
    GDumbPlugin,
    LwFPlugin,
    AGEMPlugin,
    GEMPlugin,
    EWCPlugin,
    EvaluationPlugin,
    SynapticIntelligencePlugin,
    CoPEPlugin,
    GSS_greedyPlugin,
    LFLPlugin,
    MASPlugin,
    BiCPlugin,
    MIRPlugin,
    FromScratchTrainingPlugin,
)
from avalanche.training.templates.base import BaseTemplate
from avalanche.training.templates import SupervisedTemplate
from avalanche.evaluation.metrics import loss_metrics
from avalanche.models.generator import MlpVAE, VAE_loss
from avalanche.models.expert_gate import AE_loss
from avalanche.logging import InteractiveLogger
from avalanche.training.templates.strategy_mixin_protocol import CriterionType


[docs]class Naive(SupervisedTemplate): """Naive finetuning. The simplest (and least effective) Continual Learning strategy. Naive just incrementally fine tunes a single model without employing any method to contrast the catastrophic forgetting of previous knowledge. This strategy does not use task identities. Naive is easy to set up and its results are commonly used to show the worst performing baseline. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType = CrossEntropyLoss(), train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """ Creates an instance of the Naive strategy. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class PNNStrategy(SupervisedTemplate): """Progressive Neural Network strategy. To use this strategy you need to instantiate a PNN model. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion=CrossEntropyLoss(), train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: int = 1, device: Union[str, torch.device] = "cpu", plugins: Optional[Sequence["SupervisedPlugin"]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: PyTorch model. :param optimizer: PyTorch optimizer. :param criterion: loss function. :param train_mb_size: mini-batch size for training. :param train_epochs: number of training epochs. :param eval_mb_size: mini-batch size for eval. :param device: PyTorch device where the model will be allocated. :param plugins: (optional) list of StrategyPlugins. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. None to remove logging. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs ) # Check that the model has the correct architecture. assert isinstance(self.model, PNN), "PNNStrategy requires a PNN model."
class PackNet(SupervisedTemplate): """Task-incremental fixed-network parameter isolation with PackNet. The strategy packs multiple tasks into a single network by pruning parameters that are not important for the current task. This is done by pruning the network after each task. The network is then finetuned for a few epochs to recover the performance. This process is repeated for each task. The supplied model must be wrapped in a :class:`PackNetModel` or implement :class:`PackNetModule`. These wrappers are designed to automatically upgrade most models to support PackNet. Mallya, A., & Lazebnik, S. (2018). PackNet: Adding Multiple Tasks to a Single Network by Iterative Pruning. 2018 IEEE/CVF Conference on Computer Vision and Pattern Recognition, 7765-7773. https://doi.org/10.1109/CVPR.2018.00810 """ def __init__( self, *, model: Union[PackNetModule, PackNetModel], optimizer: Optimizer, post_prune_epochs: int, prune_proportion: float, criterion=CrossEntropyLoss(), train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """ Creates an instance of the Naive strategy. :param model: The model. You can use many modules wrapped in a :class:`avalanche.models.PackNetModel` or your own implementation of :class:`avalanche.models.PackNetModule`. :param optimizer: The optimizer to use. :param post_prune_epochs: The number of epochs to finetune the model after pruning the parameters. Must be less than the number of training epochs. :param prune_proportion: The proportion of parameters to prune each durring each task. Must be between 0 and 1. :param criterion: The loss criterion to use. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=[plugins, PackNetPlugin(post_prune_epochs, prune_proportion)], evaluator=evaluator, eval_every=eval_every, **base_kwargs ) if not isinstance(self.model, PackNetModule): raise ValueError("PackNet requires a model that implements PackNetModule.")
[docs]class CWRStar(SupervisedTemplate): """CWR* Strategy."""
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, cwr_layer_name: str, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param cwr_layer_name: name of the CWR layer. Defaults to None, which means that the last fully connected layer will be used. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ cwsp = CWRStarPlugin( model, cwr_layer_name, freeze_remaining_model=True, ) if plugins is None: plugins = [cwsp] else: plugins.append(cwsp) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class Replay(SupervisedTemplate): """Experience replay strategy. See ReplayPlugin for more details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, mem_size: int = 200, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param mem_size: replay buffer size. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ rp = ReplayPlugin(mem_size) if plugins is None: plugins = [rp] else: plugins.append(rp) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class GenerativeReplay(SupervisedTemplate): """Generative Replay Strategy This implements Deep Generative Replay for a Scholar consisting of a Solver and Generator as described in https://arxiv.org/abs/1705.08690. The model parameter should contain the solver. As an optional input a generator can be wrapped in a trainable strategy and passed to the generator_strategy parameter. By default a simple VAE will be used as generator. For the case where the Generator is the model itself that is to be trained, please simply add the GenerativeReplayPlugin() when instantiating your Generator's strategy. See GenerativeReplayPlugin for more details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType = CrossEntropyLoss(), train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, generator_strategy: Optional[BaseTemplate] = None, replay_size: Optional[int] = None, increasing_replay_size: bool = False, **base_kwargs ): """ Creates an instance of Generative Replay Strategy for a solver-generator pair. :param model: The solver model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param generator_strategy: A trainable strategy with a generative model, which employs GenerativeReplayPlugin. Defaults to None. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ # Check if user inputs a generator model # (which is wrapped in a strategy that can be trained and # uses the GenerativeReplayPlugin; # see 'VAETraining" as an example below.) if generator_strategy is not None: self.generator_strategy = generator_strategy else: # By default we use a fully-connected VAE as the generator. # model: generator = MlpVAE((1, 28, 28), nhid=2, device=device) # optimzer: lr = 0.01 from torch.optim import Adam to_optimize: List[Parameter] = list( filter(lambda p: p.requires_grad, generator.parameters()) ) optimizer_generator = Adam( to_optimize, lr=lr, weight_decay=0.0001, ) # strategy (with plugin): self.generator_strategy = VAETraining( model=generator, optimizer=optimizer_generator, criterion=VAE_loss, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=[ GenerativeReplayPlugin( replay_size=replay_size, increasing_replay_size=increasing_replay_size, ) ], ) rp = GenerativeReplayPlugin( generator_strategy=self.generator_strategy, replay_size=replay_size, increasing_replay_size=increasing_replay_size, ) tgp = TrainGeneratorAfterExpPlugin() if plugins is None: plugins = [tgp, rp] else: plugins.append(tgp) plugins.append(rp) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
class AETraining(SupervisedTemplate): """AETraining class This is the training strategy for the AE model. We make use of the SupervisedTemplate, even though technically this is not a supervised training. However, this reduces the modification to a minimum. We only need to overwrite the criterion function in order to pass all necessary variables to the VAE loss function. Furthermore we remove all metrics from the evaluator. """ ae_evaluator = EvaluationPlugin( loss_metrics(minibatch=False, epoch=True, experience=False, stream=True), loggers=[InteractiveLogger()], ) def __init__( self, *, model: Module, optimizer: Optimizer, device, criterion=AE_loss, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: int = 1, plugins: Optional[List[SupervisedPlugin]] = None, evaluator: EvaluationPlugin = ae_evaluator, eval_every=0, **base_kwargs ): """ Creates an instance of the Naive strategy. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs ) def criterion(self): modified_mb_x = sigmoid(self.model.feature_module(self.mb_x).to(self.device)) return self._criterion(modified_mb_x, self.mb_output) def get_default_vae_logger(): return EvaluationPlugin(loggers=default_loggers) class VAETraining(SupervisedTemplate): """VAETraining class This is the training strategy for the VAE model found in the models directory. We make use of the SupervisedTemplate, even though technically this is not a supervised training. However, this reduces the modification to a minimum. We only need to overwrite the criterion function in order to pass all necessary variables to the VAE loss function. Furthermore we remove all metrics from the evaluator. """ def __init__( self, *, model: Module, optimizer: Optimizer, criterion=VAE_loss, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = get_default_vae_logger, eval_every=-1, **base_kwargs ): """ Creates an instance of the Naive strategy. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs ) def criterion(self): """Adapt input to criterion as needed to compute reconstruction loss and KL divergence. See default criterion VAELoss.""" return self._criterion(self.mb_x, self.mb_output)
[docs]class GSS_greedy(SupervisedTemplate): """Experience replay strategy. See ReplayPlugin for more details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, mem_size: int = 200, mem_strength=1, input_size=[], train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param mem_size: replay buffer size. :param n: memory random set size. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ rp = GSS_greedyPlugin( mem_size=mem_size, mem_strength=mem_strength, input_size=input_size ) if plugins is None: plugins = [rp] else: plugins.append(rp) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class GDumb(SupervisedTemplate): """GDumb strategy. See GDumbPlugin for more details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, mem_size: int = 200, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param mem_size: replay buffer size. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ gdumb = GDumbPlugin(mem_size) if plugins is None: plugins = [gdumb] else: plugins.append(gdumb) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class LwF(SupervisedTemplate): """Learning without Forgetting (LwF) strategy. See LwF plugin for details. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, alpha: Union[float, Sequence[float]], temperature: float, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param alpha: distillation hyperparameter. It can be either a float number or a list containing alpha for each experience. :param temperature: softmax temperature for distillation :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ lwf = LwFPlugin(alpha, temperature) if plugins is None: plugins = [lwf] else: plugins.append(lwf) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class AGEM(SupervisedTemplate): """Average Gradient Episodic Memory (A-GEM) strategy. See AGEM plugin for details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, patterns_per_exp: int, sample_size: int = 64, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param patterns_per_exp: number of patterns per experience in the memory :param sample_size: number of patterns in memory sample when computing reference gradient. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ agem = AGEMPlugin(patterns_per_exp, sample_size) if plugins is None: plugins = [agem] else: plugins.append(agem) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class GEM(SupervisedTemplate): """Gradient Episodic Memory (GEM) strategy. See GEM plugin for details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, patterns_per_exp: int, memory_strength: float = 0.5, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param patterns_per_exp: number of patterns per experience in the memory :param memory_strength: offset to add to the projection direction in order to favour backward transfer (gamma in original paper). :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ gem = GEMPlugin(patterns_per_exp, memory_strength) if plugins is None: plugins = [gem] else: plugins.append(gem) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class EWC(SupervisedTemplate): """Elastic Weight Consolidation (EWC) strategy. See EWC plugin for details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, ewc_lambda: float, mode: str = "separate", decay_factor: Optional[float] = None, keep_importance_data: bool = False, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param ewc_lambda: hyperparameter to weigh the penalty inside the total loss. The larger the lambda, the larger the regularization. :param mode: `separate` to keep a separate penalty for each previous experience. `onlinesum` to keep a single penalty summed over all previous tasks. `onlineweightedsum` to keep a single penalty summed with a decay factor over all previous tasks. :param decay_factor: used only if mode is `onlineweightedsum`. It specify the decay term of the importance matrix. :param keep_importance_data: if True, keep in memory both parameter values and importances for all previous task, for all modes. If False, keep only last parameter values and importances. If mode is `separate`, the value of `keep_importance_data` is set to be True. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ ewc = EWCPlugin(ewc_lambda, mode, decay_factor, keep_importance_data) if plugins is None: plugins = [ewc] else: plugins.append(ewc) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class SynapticIntelligence(SupervisedTemplate): """Synaptic Intelligence strategy. This is the Synaptic Intelligence PyTorch implementation of the algorithm described in the paper "Continuous Learning in Single-Incremental-Task Scenarios" (https://arxiv.org/abs/1806.08568) The original implementation has been proposed in the paper "Continual Learning Through Synaptic Intelligence" (https://arxiv.org/abs/1703.04200). The Synaptic Intelligence regularization can also be used in a different strategy by applying the :class:`SynapticIntelligencePlugin` plugin. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, si_lambda: Union[float, Sequence[float]], eps: float = 0.0000001, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: int = 1, device: Union[str, torch.device] = "cpu", plugins: Optional[Sequence["SupervisedPlugin"]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. Creates an instance of the Synaptic Intelligence strategy. :param model: PyTorch model. :param optimizer: PyTorch optimizer. :param criterion: loss function. :param si_lambda: Synaptic Intelligence lambda term. If list, one lambda for each experience. If the list has less elements than the number of experiences, last lambda will be used for the remaining experiences. :param eps: Synaptic Intelligence damping parameter. :param train_mb_size: mini-batch size for training. :param train_epochs: number of training epochs. :param eval_mb_size: mini-batch size for eval. :param device: PyTorch device to run the model. :param plugins: (optional) list of StrategyPlugins. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ if plugins is None: plugins = [] plugins = list(plugins) # This implementation relies on the S.I. Plugin, which contains the # entire implementation of the strategy! plugins.append(SynapticIntelligencePlugin(si_lambda=si_lambda, eps=eps)) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class CoPE(SupervisedTemplate): """Continual Prototype Evolution strategy. See CoPEPlugin for more details. This strategy does not use task identities during training. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, mem_size: int = 200, n_classes: int = 10, p_size: int = 100, alpha: float = 0.99, T: float = 0.1, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: Loss criterion to use. Standard overwritten by PPPloss (see CoPEPlugin). :param mem_size: replay buffer size. :param n_classes: total number of classes that will be encountered. This is used to output predictions for all classes, with zero probability for unseen classes. :param p_size: The prototype size, which equals the feature size of the last layer. :param alpha: The momentum for the exponentially moving average of the prototypes. :param T: The softmax temperature, used as a concentration parameter. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ copep = CoPEPlugin(mem_size, n_classes, p_size, alpha, T) if plugins is None: plugins = [copep] else: plugins.append(copep) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class LFL(SupervisedTemplate): """Less Forgetful Learning strategy. See LFL plugin for details. Refer Paper: https://arxiv.org/pdf/1607.00122.pdf This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, lambda_e: Union[float, Sequence[float]], train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param lambda_e: euclidean loss hyper parameter. It can be either a float number or a list containing lambda_e for each experience. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :class:`~avalanche.training.BaseTemplate` constructor arguments. """ lfl = LFLPlugin(lambda_e) if plugins is None: plugins = [lfl] else: plugins.append(lfl) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class MAS(SupervisedTemplate): """Memory Aware Synapses (MAS) strategy. See MAS plugin for details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, lambda_reg: float = 1.0, alpha: float = 0.5, verbose: bool = False, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: int = 1, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param lambda_reg: hyperparameter weighting the penalty term in the overall loss. :param alpha: hyperparameter that specifies the weight given to the influence of the previous experience. :param verbose: when True, the computation of the influence of each parameter shows a progress bar. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ # Instantiate plugin mas = MASPlugin(lambda_reg=lambda_reg, alpha=alpha, verbose=verbose) # Add plugin to the strategy if plugins is None: plugins = [mas] else: plugins.append(mas) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class BiC(SupervisedTemplate): """Bias Correction (BiC) strategy. See BiC plugin for details. This strategy does not use task identities. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, mem_size: int = 200, val_percentage: float = 0.1, T: int = 2, stage_2_epochs: int = 200, lamb: float = -1, lr: float = 0.1, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param mem_size: replay buffer size. :param val_percentage: hyperparameter used to set the percentage of exemplars in the val set. :param T: hyperparameter used to set the temperature used in stage 1. :param stage_2_epochs: hyperparameter used to set the amount of epochs of stage 2. :param lamb: hyperparameter used to balance the distilling loss and the classification loss. :param lr: hyperparameter used as a learning rate for the second phase of training. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ # Instantiate plugin bic = BiCPlugin( mem_size=mem_size, val_percentage=val_percentage, T=T, stage_2_epochs=stage_2_epochs, lamb=lamb, lr=lr, ) # Add plugin to the strategy if plugins is None: plugins = [bic] else: plugins.append(bic) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class MIR(SupervisedTemplate): """Maximally Interfered Replay Strategy See ER_MIR plugin for details. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, mem_size: int, subsample: int, batch_size_mem: int = 1, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: int = 1, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param mem_size: Amount of fixed memory to use :param subsample: Size of the initial sample from which to select the replay batch :param batch_size_mem: Size of the replay batch after loss-based selection :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ # Instantiate plugin mir = MIRPlugin( mem_size=mem_size, subsample=subsample, batch_size_mem=batch_size_mem, ) # Add plugin to the strategy if plugins is None: plugins = [mir] else: plugins.append(mir) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
[docs]class FromScratchTraining(SupervisedTemplate): """From scratch training strategy. This strategy trains a model on a stream of experiences, but resets the model's weight initialization and optimizer state after each experience. It is usually used a baseline for comparison with the Naive strategy where the model is fine-tuned to every new experience. See FromScratchTrainingPlugin for more details. """
[docs] def __init__( self, *, model: Module, optimizer: Optimizer, criterion: CriterionType, reset_optimizer: bool = True, train_mb_size: int = 1, train_epochs: int = 1, eval_mb_size: Optional[int] = None, device: Union[str, torch.device] = "cpu", plugins: Optional[List[SupervisedPlugin]] = None, evaluator: Union[ EvaluationPlugin, Callable[[], EvaluationPlugin] ] = default_evaluator, eval_every=-1, **base_kwargs ): """Init. :param model: The model. :param optimizer: The optimizer to use. :param criterion: The loss criterion to use. :param reset_optimizer: If True, optimizer state will be reset after each experience. :param train_mb_size: The train minibatch size. Defaults to 1. :param train_epochs: The number of training epochs. Defaults to 1. :param eval_mb_size: The eval minibatch size. Defaults to 1. :param device: The device to use. Defaults to None (cpu). :param plugins: Plugins to be added. Defaults to None. :param evaluator: (optional) instance of EvaluationPlugin for logging and metric computations. :param eval_every: the frequency of the calls to `eval` inside the training loop. -1 disables the evaluation. 0 means `eval` is called only at the end of the learning experience. Values >0 mean that `eval` is called every `eval_every` epochs and at the end of the learning experience. :param **base_kwargs: any additional :class:`~avalanche.training.BaseTemplate` constructor arguments. """ fstp = FromScratchTrainingPlugin(reset_optimizer=reset_optimizer) if plugins is None: plugins = [fstp] else: plugins.append(fstp) super().__init__( model=model, optimizer=optimizer, criterion=criterion, train_mb_size=train_mb_size, train_epochs=train_epochs, eval_mb_size=eval_mb_size, device=device, plugins=plugins, evaluator=evaluator, eval_every=eval_every, **base_kwargs )
__all__ = [ "Naive", "PNNStrategy", "CWRStar", "Replay", "GenerativeReplay", "AETraining", "VAETraining", "GDumb", "LwF", "AGEM", "GEM", "EWC", "SynapticIntelligence", "GSS_greedy", "CoPE", "LFL", "MAS", "BiC", "MIR", "FromScratchTraining", ]