25090331288

Committed 29 Apr 2026 04:05AM UTC coverage: 75.028% (+0.1%) from 74.909%

Build # 25090331288

Build Type

Pull #355

github

Committed by

web-flow

Commit Message

Merge 263e60102 into 2ef7fe73e

Pull Request Pull Request #355: feat(ocl): add lwf

Coverage Stats

57 of 60 new or added lines in 4 files covered. (95.0%)

51 existing lines in 4 files now uncovered.

7247 of 9659 relevant lines covered (75.03%)

0.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.87

/src/capymoa/ocl/strategy/_rar.py

import torch
from torch import Tensor

from capymoa.base import BatchClassifier
from capymoa.ocl.util._replay import ReservoirSampler
from capymoa.ocl.base import TrainTaskAware, TestTaskAware

from typing import Callable


class RAR(BatchClassifier, TrainTaskAware, TestTaskAware):
    """Repeated Augmented Rehearsal.

    Repeated Augmented Rehearsal (RAR) [#f0]_ is a replay continual learning
    strategy that combines data augmentation with repeated training on each
    batch to mitigate catastrophic forgetting.

    * Coreset Selection: Reservoir sampling is used to select a fixed-size
      buffer of past examples.

    * Coreset Retrieval: During training, the learner samples uniformly from the
      buffer of past examples.

    * Coreset Exploitation: The learner trains on the current batch of examples
      and the sampled buffer examples, performing multiple optimization steps
      per-batch using random augmentations of the examples. The original paper uses
      RandAugment [#f1]_ for augmentation but any randomized augmentation can be used.
      But the choice of augmentation is important and should be chosen based on the
      problem domain.

    * Not :class:`~capymoa.ocl.base.TrainTaskAware` or
      :class:`~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 RAR
    >>> from capymoa.ocl.datasets import TinySplitMNIST
    >>> from capymoa.ocl.evaluation import ocl_train_eval_loop
    >>> import torchvision.transforms as T
    >>> import torch
    >>> _ = torch.manual_seed(0)
    >>> scenario = TinySplitMNIST()
    >>> model = Perceptron(scenario.schema)
    >>> # You should use more complex augmentations for more challenging problems.
    >>> augment = T.Compose([
    ...     T.RandomRotation(10),
    ... ])
    >>> learner = RAR(Finetune(scenario.schema, model), augment=augment, repeats=5)
    >>> results = ocl_train_eval_loop(
    ...     learner,
    ...     scenario.train_loaders(32),
    ...     scenario.test_loaders(32),
    ... )
    >>> print(f"{results.accuracy_final*100:.1f}%")
    45.0%

    Usually more complex augmentations are used such as random crops and
    rotations.

    .. [#f0] Zhang, Yaqian, Bernhard Pfahringer, Eibe Frank, Albert Bifet, Nick
       Jin Sean Lim, and Yunzhe Jia. “A Simple but Strong Baseline for Online
       Continual Learning: Repeated Augmented Rehearsal.” In Advances in Neural
       Information Processing Systems 35: Annual Conference on Neural
       Information Processing Systems 2022, NeurIPS 2022, New Orleans, LA, USA,
       November 28 - December 9, 2022, edited by Sanmi Koyejo, S. Mohamed, A.
       Agarwal, Danielle Belgrave, K. Cho, and A. Oh, 2022.
       https://doi.org/10.5555/3600270.3601344.

    .. [#f1] Cubuk, E. D., Zoph, B., Shlens, J., & Le, Q. V. (2020). Randaugment:
       Practical automated data augmentation with a reduced search space. 2020 IEEE/CVF
       Conference on Computer Vision and Pattern Recognition Workshops (CVPRW),
       3008-3017. https://doi.org/10.1109/CVPRW50498.2020.00359
    """

    def __init__(
        self,
        learner: BatchClassifier,
        augment: Callable[[Tensor], Tensor],
        coreset_size: int = 200,
        repeats: int = 1,
    ) -> None:
        """Initialize Repeated Augmented Rehearsal.

        :param learner: Underlying learner to be trained with RAR.
        :param augment: Data augmentation function to apply to the samples. Should take
            a Tensor of shape ``(batch_size, *schema.shape)`` and return a Tensor of the
            same shape. Take a look at the PyTorch torchvision transforms for some
            building blocks for your pipeline (https://docs.pytorch.org/vision/main/transforms.html).
        :param coreset_size: Size of the coreset buffer.
        :param repeats: Number of times to repeat training on each batch, defaults to 1.
        """

        super().__init__(learner.schema)
        num_features = learner.schema.get_num_attributes()
        self.learner = learner
        self.augment = augment
        self.repeats = repeats
        self.coreset = ReservoirSampler(
            coreset_size,
            num_features,
            rng=torch.Generator().manual_seed(learner.random_seed),
        )
        self.shape = learner.schema.shape

    def train_step(self, x_fresh: Tensor, y_fresh: Tensor) -> None:
        # Sample from reservoir and augment the data
        n = x_fresh.shape[0]
        x_replay, y_replay = self.coreset.sample(n)
        x = torch.cat((x_fresh, x_replay), dim=0).to(self.device, self.x_dtype)
        y = torch.cat((y_fresh, y_replay), dim=0).to(self.device, self.y_dtype)
        x = x.view(-1, *self.shape)
        x: Tensor = self.augment(x)

        # Train the learner
        x = x.to(self.learner.device, self.learner.x_dtype)
        y = y.to(self.learner.device, self.learner.y_dtype)
        self.learner.batch_train(x, y)

    def batch_train(self, x: Tensor, y: Tensor) -> None:
        self.coreset.update(x, y)
        for i in range(self.repeats):
            self.train_step(x, y)

    @torch.no_grad()
    def batch_predict_proba(self, x: Tensor) -> Tensor:
        x = x.to(self.learner.device, self.learner.x_dtype)
        return self.learner.batch_predict_proba(x)

    def on_test_task(self, task_id: int):
        if isinstance(self.learner, TestTaskAware):
            self.learner.on_test_task(task_id)

    def on_train_task(self, task_id: int):
        if isinstance(self.learner, TrainTaskAware):
            self.learner.on_train_task(task_id)

1	import torch	1✔
2	from torch import Tensor	1✔
3
4	from capymoa.base import BatchClassifier	1✔
5	from capymoa.ocl.util._replay import ReservoirSampler	1✔
6	from capymoa.ocl.base import TrainTaskAware, TestTaskAware	1✔
7
8	from typing import Callable	1✔
9
10
11	class RAR(BatchClassifier, TrainTaskAware, TestTaskAware):	1✔
12	"""Repeated Augmented Rehearsal.
13
14	Repeated Augmented Rehearsal (RAR) [#f0]_ is a replay continual learning
15	strategy that combines data augmentation with repeated training on each
16	batch to mitigate catastrophic forgetting.
17
18	* Coreset Selection: Reservoir sampling is used to select a fixed-size
19	buffer of past examples.
20
21	* Coreset Retrieval: During training, the learner samples uniformly from the
22	buffer of past examples.
23
24	* Coreset Exploitation: The learner trains on the current batch of examples
25	and the sampled buffer examples, performing multiple optimization steps
26	per-batch using random augmentations of the examples. The original paper uses
27	RandAugment [#f1]_ for augmentation but any randomized augmentation can be used.
28	But the choice of augmentation is important and should be chosen based on the
29	problem domain.
30
31	* Not :class:`~capymoa.ocl.base.TrainTaskAware` or
32	:class:`~capymoa.ocl.base.TestTaskAware`, but will proxy it to the wrapped
33	learner.
34
35	>>> from capymoa.ann import Perceptron
36	>>> from capymoa.classifier import Finetune
37	>>> from capymoa.ocl.strategy import RAR
38	>>> from capymoa.ocl.datasets import TinySplitMNIST
39	>>> from capymoa.ocl.evaluation import ocl_train_eval_loop
40	>>> import torchvision.transforms as T
41	>>> import torch
42	>>> _ = torch.manual_seed(0)
43	>>> scenario = TinySplitMNIST()
44	>>> model = Perceptron(scenario.schema)
45	>>> # You should use more complex augmentations for more challenging problems.
46	>>> augment = T.Compose([
47	... T.RandomRotation(10),
48	... ])
49	>>> learner = RAR(Finetune(scenario.schema, model), augment=augment, repeats=5)
50	>>> results = ocl_train_eval_loop(
51	... learner,
52	... scenario.train_loaders(32),
53	... scenario.test_loaders(32),
54	... )
55	>>> print(f"{results.accuracy_final*100:.1f}%")
56	45.0%
57
58	Usually more complex augmentations are used such as random crops and
59	rotations.
60
61	.. [#f0] Zhang, Yaqian, Bernhard Pfahringer, Eibe Frank, Albert Bifet, Nick
62	Jin Sean Lim, and Yunzhe Jia. “A Simple but Strong Baseline for Online
63	Continual Learning: Repeated Augmented Rehearsal.” In Advances in Neural
64	Information Processing Systems 35: Annual Conference on Neural
65	Information Processing Systems 2022, NeurIPS 2022, New Orleans, LA, USA,
66	November 28 - December 9, 2022, edited by Sanmi Koyejo, S. Mohamed, A.
67	Agarwal, Danielle Belgrave, K. Cho, and A. Oh, 2022.
68	https://doi.org/10.5555/3600270.3601344.
69
70	.. [#f1] Cubuk, E. D., Zoph, B., Shlens, J., & Le, Q. V. (2020). Randaugment:
71	Practical automated data augmentation with a reduced search space. 2020 IEEE/CVF
72	Conference on Computer Vision and Pattern Recognition Workshops (CVPRW),
73	3008-3017. https://doi.org/10.1109/CVPRW50498.2020.00359
74	"""
75
76	def __init__(	1✔
77	self,
78	learner: BatchClassifier,
79	augment: Callable[[Tensor], Tensor],
80	coreset_size: int = 200,
81	repeats: int = 1,
82	) -> None:
83	"""Initialize Repeated Augmented Rehearsal.
84
85	:param learner: Underlying learner to be trained with RAR.
86	:param augment: Data augmentation function to apply to the samples. Should take
87	a Tensor of shape ``(batch_size, *schema.shape)`` and return a Tensor of the
88	same shape. Take a look at the PyTorch torchvision transforms for some
89	building blocks for your pipeline (https://docs.pytorch.org/vision/main/transforms.html).
90	:param coreset_size: Size of the coreset buffer.
91	:param repeats: Number of times to repeat training on each batch, defaults to 1.
92	"""
93
94	super().__init__(learner.schema)	1✔
95	num_features = learner.schema.get_num_attributes()	1✔
96	self.learner = learner	1✔
97	self.augment = augment	1✔
98	self.repeats = repeats	1✔
99	self.coreset = ReservoirSampler(	1✔
100	coreset_size,
101	num_features,
102	rng=torch.Generator().manual_seed(learner.random_seed),
103	)
104	self.shape = learner.schema.shape	1✔
105
106	def train_step(self, x_fresh: Tensor, y_fresh: Tensor) -> None:	1✔
107	# Sample from reservoir and augment the data
108	n = x_fresh.shape[0]	1✔
109	x_replay, y_replay = self.coreset.sample(n)	1✔
110	x = torch.cat((x_fresh, x_replay), dim=0).to(self.device, self.x_dtype)	1✔
111	y = torch.cat((y_fresh, y_replay), dim=0).to(self.device, self.y_dtype)	1✔
112	x = x.view(-1, *self.shape)	1✔
113	x: Tensor = self.augment(x)	1✔
114
115	# Train the learner
116	x = x.to(self.learner.device, self.learner.x_dtype)	1✔
117	y = y.to(self.learner.device, self.learner.y_dtype)	1✔
118	self.learner.batch_train(x, y)	1✔
119
120	def batch_train(self, x: Tensor, y: Tensor) -> None:	1✔
121	self.coreset.update(x, y)	1✔
122	for i in range(self.repeats):	1✔
123	self.train_step(x, y)	1✔
124
125	@torch.no_grad()	1✔
126	def batch_predict_proba(self, x: Tensor) -> Tensor:	1✔
127	x = x.to(self.learner.device, self.learner.x_dtype)	1✔
128	return self.learner.batch_predict_proba(x)	1✔
129
130	def on_test_task(self, task_id: int):	1✔
131	if isinstance(self.learner, TestTaskAware):	1✔
UNCOV 132	self.learner.on_test_task(task_id)	×
133
134	def on_train_task(self, task_id: int):	1✔
135	if isinstance(self.learner, TrainTaskAware):	1✔
UNCOV 136	self.learner.on_train_task(task_id)	×

adaptive-machine-learning / CapyMOA / 25090331288

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous