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

93.55

/src/capymoa/ocl/strategy/_experience_replay.py

import torch
from torch import Tensor

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


class ExperienceReplay(BatchClassifier, TrainTaskAware, TestTaskAware):
    """Experience Replay.

    Experience Replay (ER) [#f0]_ 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 [#f1]_.
    * 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 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}%")
    32.5%

    .. [#f0] `Rolnick, D., Ahuja, A., Schwarz, J., Lillicrap, T., & Wayne, G. (2019).
              Experience replay for continual learning. Advances in neural information
              processing systems, 32. <https://arxiv.org/abs/1811.11682>`_
    .. [#f1] `Jeffrey S. Vitter. 1985. Random sampling with a reservoir. ACM Trans. Math.
              Softw. 11, 1 (March 1985), 37–57. <https://doi.org/10.1145/3147.3165>`_
    """

    def __init__(
        self, learner: BatchClassifier, buffer_size: int = 200, repeat: int = 1
    ) -> None:
        """Initialize the Experience Replay strategy.

        :param learner: The learner to be wrapped for experience replay.
        :param buffer_size: The size of the replay buffer, defaults to 200.
        :param repeat: The number of times to repeat the training data in each batch,
            defaults to 1.
        """
        super().__init__(learner.schema, learner.random_seed)
        #: The wrapped learner to be trained with experience replay.
        self.learner = learner
        self._buffer = ReservoirSampler(
            capacity=buffer_size,
            features=self.schema.get_num_attributes(),
            rng=torch.Generator().manual_seed(learner.random_seed),
        )
        self.repeat = repeat

    def batch_train(self, x: Tensor, y: Tensor) -> None:
        # update the buffer with the new data
        self._buffer.update(x, y)

        for _ in range(self.repeat):
            # sample from the buffer and construct training batch
            replay_x, replay_y = self._buffer.sample(x.shape[0])
            train_x = torch.cat((x, replay_x), dim=0)
            train_y = torch.cat((y, replay_y), dim=0)
            train_x = train_x.to(self.learner.device, dtype=self.learner.x_dtype)
            train_y = train_y.to(self.learner.device, dtype=self.learner.y_dtype)
            self.learner.batch_train(train_x, train_y)

    def batch_predict_proba(self, x: Tensor) -> Tensor:
        x = x.to(self.learner.device, dtype=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)

    def __str__(self) -> str:
        return f"ExperienceReplay(buffer_size={self._buffer.capacity})"

1	import torch	1✔
2	from torch import Tensor	1✔
3
4	from capymoa.base import BatchClassifier	1✔
5	from capymoa.ocl.base import TrainTaskAware, TestTaskAware	1✔
6	from capymoa.ocl.util._replay import ReservoirSampler	1✔
7
8
9	class ExperienceReplay(BatchClassifier, TrainTaskAware, TestTaskAware):	1✔
10	"""Experience Replay.
11
12	Experience Replay (ER) [#f0]_ is a replay continual learning strategy.
13
14	* Uses a replay buffer to store past experiences and samples from it during training
15	to mitigate catastrophic forgetting.
16	* The replay buffer is implemented using reservoir sampling, which allows for
17	uniform sampling over the entire stream [#f1]_.
18	* Not :class:`capymoa.ocl.base.TrainTaskAware` or
19	:class:`capymoa.ocl.base.TestTaskAware`, but will proxy it to the wrapped learner.
20
21	>>> from capymoa.ann import Perceptron
22	>>> from capymoa.classifier import Finetune
23	>>> from capymoa.ocl.strategy import ExperienceReplay
24	>>> from capymoa.ocl.datasets import TinySplitMNIST
25	>>> from capymoa.ocl.evaluation import ocl_train_eval_loop
26	>>> import torch
27	>>> _ = torch.manual_seed(0)
28	>>> scenario = TinySplitMNIST()
29	>>> model = Perceptron(scenario.schema)
30	>>> learner = ExperienceReplay(Finetune(scenario.schema, model))
31	>>> results = ocl_train_eval_loop(
32	... learner,
33	... scenario.train_loaders(32),
34	... scenario.test_loaders(32),
35	... )
36	>>> print(f"{results.accuracy_final*100:.1f}%")
37	32.5%
38
39	.. [#f0] `Rolnick, D., Ahuja, A., Schwarz, J., Lillicrap, T., & Wayne, G. (2019).
40	Experience replay for continual learning. Advances in neural information
41	processing systems, 32. <https://arxiv.org/abs/1811.11682>`_
42	.. [#f1] `Jeffrey S. Vitter. 1985. Random sampling with a reservoir. ACM Trans. Math.
43	Softw. 11, 1 (March 1985), 37–57. <https://doi.org/10.1145/3147.3165>`_
44	"""
45
46	def __init__(	1✔
47	self, learner: BatchClassifier, buffer_size: int = 200, repeat: int = 1
48	) -> None:
49	"""Initialize the Experience Replay strategy.
50
51	:param learner: The learner to be wrapped for experience replay.
52	:param buffer_size: The size of the replay buffer, defaults to 200.
53	:param repeat: The number of times to repeat the training data in each batch,
54	defaults to 1.
55	"""
56	super().__init__(learner.schema, learner.random_seed)	1✔
57	#: The wrapped learner to be trained with experience replay.
58	self.learner = learner	1✔
59	self._buffer = ReservoirSampler(	1✔
60	capacity=buffer_size,
61	features=self.schema.get_num_attributes(),
62	rng=torch.Generator().manual_seed(learner.random_seed),
63	)
64	self.repeat = repeat	1✔
65
66	def batch_train(self, x: Tensor, y: Tensor) -> None:	1✔
67	# update the buffer with the new data
68	self._buffer.update(x, y)	1✔
69
70	for _ in range(self.repeat):	1✔
71	# sample from the buffer and construct training batch
72	replay_x, replay_y = self._buffer.sample(x.shape[0])	1✔
73	train_x = torch.cat((x, replay_x), dim=0)	1✔
74	train_y = torch.cat((y, replay_y), dim=0)	1✔
75	train_x = train_x.to(self.learner.device, dtype=self.learner.x_dtype)	1✔
76	train_y = train_y.to(self.learner.device, dtype=self.learner.y_dtype)	1✔
77	self.learner.batch_train(train_x, train_y)	1✔
78
79	def batch_predict_proba(self, x: Tensor) -> Tensor:	1✔
80	x = x.to(self.learner.device, dtype=self.learner.x_dtype)	1✔
81	return self.learner.batch_predict_proba(x)	1✔
82
83	def on_test_task(self, task_id: int):	1✔
84	if isinstance(self.learner, TestTaskAware):	1✔
UNCOV 85	self.learner.on_test_task(task_id)	×
86
87	def on_train_task(self, task_id: int):	1✔
88	if isinstance(self.learner, TrainTaskAware):	1✔
UNCOV 89	self.learner.on_train_task(task_id)	×
90
91	def __str__(self) -> str:	1✔
92	return f"ExperienceReplay(buffer_size={self._buffer.capacity})"	1✔

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