25189796363

Committed 30 Apr 2026 09:17PM UTC coverage: 75.524% (+0.6%) from 74.909%

Build # 25189796363

Build Type

Pull #355

github

Committed by

web-flow

Commit Message

Merge e4643751b into 2ef7fe73e

Pull Request Pull Request #355: feat(ocl): add LWF and DER

Coverage Stats

366 of 372 new or added lines in 19 files covered. (98.39%)

6 existing lines in 4 files now uncovered.

7452 of 9867 relevant lines covered (75.52%)

0.76 hits per line

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

95.83

/src/capymoa/ocl/strategy/_der.py

from typing import Callable

import torch
from torch import Tensor, nn

from capymoa.base import BatchClassifier
from capymoa.ocl.util._replay import ReservoirSampler
from capymoa.stream._stream import Schema


class DER(BatchClassifier, nn.Module):
    """Dark Experience Replay.

    Dark Experience Replay (DER) [#f1]_ is a replay-based continual learning
    strategy that stores model logits for replay samples and regularises new
    predictions with an MSE loss on those stored logits.

    ..  [#f1] Buzzega, Pietro, Matteo Boschini, Angelo Porrello, Davide Abati,
        and SIMONE CALDERARA. “Dark Experience for General Continual Learning:
        A Strong, Simple Baseline.” Advances in Neural Information Processing
        Systems 33 (2020): 15920–30.
        https://papers.nips.cc/paper/2020/hash/b704ea2c39778f07c617f6b7ce480e9e-Abstract.html.

    """

    def __init__(
        self,
        schema: Schema,
        model: torch.nn.Module,
        optimiser: torch.optim.Optimizer,
        augment: Callable[[Tensor], Tensor],
        alpha: float = 0.5,
        buffer_size: int = 200,
        seed: int = 0,
        substeps: int = 1,
        device: torch.device = torch.device("cpu"),
    ) -> None:
        """Construct a DER learner.

        :param schema: Stream schema used by the classifier interface.
        :param model: Torch model that outputs class logits.
        :param optimiser: Optimiser used to update ``model`` parameters.
        :param alpha: Weight of the DER replay-logit loss term.
        :param buffer_size: Number of replay samples to retain.
        :param augment: Data augmentation function that takes a batch of
            examples.
        :param substeps: Number of optimization steps to take per batch. Each
            step will use a different random augmentation of the batch and
            replay samples.
        :param seed: Random seed for reproducibility.
        :param device: Compute device.
        """
        super().__init__(schema, 0)
        nn.Module.__init__(self)
        if alpha < 0:
            raise ValueError("alpha must be non-negative.")
        if buffer_size <= 0:
            raise ValueError("buffer_size must be greater than zero.")

        self.device = device
        self._augment = augment
        self._alpha = alpha
        self._substeps = substeps
        self._model = model
        self._optimiser = optimiser
        self._criterion = torch.nn.CrossEntropyLoss()
        self._logit_loss = torch.nn.MSELoss()
        self._buffer = ReservoirSampler(
            capacity=buffer_size,
            buffers=dict(
                x=(schema.shape, torch.float32),
                z=((schema.get_num_classes(),), torch.float32),
                y=((), torch.long),
            ),
            rng=torch.Generator().manual_seed(seed),
        )
        self.to(device)

    def _train_step(self, x: Tensor, y: Tensor, update_buffer: bool) -> None:
        self._optimiser.zero_grad()

        n = x.shape[0]

        # Update Buffer
        x_t = self._augment(x)  # $x_t$
        z = self._model(x_t)  # $z$
        if update_buffer:
            self._buffer.update(x=x, z=z, y=y)

        # Sample buffer and augment
        xp, zp, _ = self._buffer.sample(n).values()  # $x'$, $z'$, $y'$
        x_tp = self._augment(xp)  # $x_t'$

        reg = self._alpha * self._logit_loss(zp, self._model(x_tp))  # Line 7
        loss = self._criterion(z, y) + reg  # Line 8
        loss.backward()
        self._optimiser.step()

    def batch_train(self, x: Tensor, y: Tensor) -> None:
        """Implementes Algorithm 1 from the DER paper."""
        self._model.train()
        for i in range(self._substeps):
            self._train_step(x, y, update_buffer=i == 0)

    @torch.no_grad()
    def batch_predict_proba(self, x: Tensor) -> Tensor:
        self._model.eval()
        y_hat = self._model(x)
        return torch.softmax(y_hat, dim=1)

    def __str__(self) -> str:
        return f"DER(alpha={self._alpha}, buffer_size={self._buffer._capacity})"

1	from typing import Callable	1✔
2
3	import torch	1✔
4	from torch import Tensor, nn	1✔
5
6	from capymoa.base import BatchClassifier	1✔
7	from capymoa.ocl.util._replay import ReservoirSampler	1✔
8	from capymoa.stream._stream import Schema	1✔
9
10
11	class DER(BatchClassifier, nn.Module):	1✔
12	"""Dark Experience Replay.
13
14	Dark Experience Replay (DER) [#f1]_ is a replay-based continual learning
15	strategy that stores model logits for replay samples and regularises new
16	predictions with an MSE loss on those stored logits.
17
18	.. [#f1] Buzzega, Pietro, Matteo Boschini, Angelo Porrello, Davide Abati,
19	and SIMONE CALDERARA. “Dark Experience for General Continual Learning:
20	A Strong, Simple Baseline.” Advances in Neural Information Processing
21	Systems 33 (2020): 15920–30.
22	https://papers.nips.cc/paper/2020/hash/b704ea2c39778f07c617f6b7ce480e9e-Abstract.html.
23
24	"""
25
26	def __init__(	1✔
27	self,
28	schema: Schema,
29	model: torch.nn.Module,
30	optimiser: torch.optim.Optimizer,
31	augment: Callable[[Tensor], Tensor],
32	alpha: float = 0.5,
33	buffer_size: int = 200,
34	seed: int = 0,
35	substeps: int = 1,
36	device: torch.device = torch.device("cpu"),
37	) -> None:
38	"""Construct a DER learner.
39
40	:param schema: Stream schema used by the classifier interface.
41	:param model: Torch model that outputs class logits.
42	:param optimiser: Optimiser used to update ``model`` parameters.
43	:param alpha: Weight of the DER replay-logit loss term.
44	:param buffer_size: Number of replay samples to retain.
45	:param augment: Data augmentation function that takes a batch of
46	examples.
47	:param substeps: Number of optimization steps to take per batch. Each
48	step will use a different random augmentation of the batch and
49	replay samples.
50	:param seed: Random seed for reproducibility.
51	:param device: Compute device.
52	"""
53	super().__init__(schema, 0)	1✔
54	nn.Module.__init__(self)	1✔
55	if alpha < 0:	1✔
NEW 56	raise ValueError("alpha must be non-negative.")	×
57	if buffer_size <= 0:	1✔
NEW 58	raise ValueError("buffer_size must be greater than zero.")	×
59
60	self.device = device	1✔
61	self._augment = augment	1✔
62	self._alpha = alpha	1✔
63	self._substeps = substeps	1✔
64	self._model = model	1✔
65	self._optimiser = optimiser	1✔
66	self._criterion = torch.nn.CrossEntropyLoss()	1✔
67	self._logit_loss = torch.nn.MSELoss()	1✔
68	self._buffer = ReservoirSampler(	1✔
69	capacity=buffer_size,
70	buffers=dict(
71	x=(schema.shape, torch.float32),
72	z=((schema.get_num_classes(),), torch.float32),
73	y=((), torch.long),
74	),
75	rng=torch.Generator().manual_seed(seed),
76	)
77	self.to(device)	1✔
78
79	def _train_step(self, x: Tensor, y: Tensor, update_buffer: bool) -> None:	1✔
80	self._optimiser.zero_grad()	1✔
81
82	n = x.shape[0]	1✔
83
84	# Update Buffer
85	x_t = self._augment(x) # $x_t$	1✔
86	z = self._model(x_t) # $z$	1✔
87	if update_buffer:	1✔
88	self._buffer.update(x=x, z=z, y=y)	1✔
89
90	# Sample buffer and augment
91	xp, zp, _ = self._buffer.sample(n).values() # $x'$, $z'$, $y'$	1✔
92	x_tp = self._augment(xp) # $x_t'$	1✔
93
94	reg = self._alpha * self._logit_loss(zp, self._model(x_tp)) # Line 7	1✔
95	loss = self._criterion(z, y) + reg # Line 8	1✔
96	loss.backward()	1✔
97	self._optimiser.step()	1✔
98
99	def batch_train(self, x: Tensor, y: Tensor) -> None:	1✔
100	"""Implementes Algorithm 1 from the DER paper."""
101	self._model.train()	1✔
102	for i in range(self._substeps):	1✔
103	self._train_step(x, y, update_buffer=i == 0)	1✔
104
105	@torch.no_grad()	1✔
106	def batch_predict_proba(self, x: Tensor) -> Tensor:	1✔
107	self._model.eval()	1✔
108	y_hat = self._model(x)	1✔
109	return torch.softmax(y_hat, dim=1)	1✔
110
111	def __str__(self) -> str:	1✔
112	return f"DER(alpha={self._alpha}, buffer_size={self._buffer._capacity})"	1✔

adaptive-machine-learning / CapyMOA / 25189796363

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