25192586337

Committed 30 Apr 2026 10:31PM UTC coverage: 75.462% (+0.6%) from 74.909%

Build # 25192586337

Build Type

Pull #355

github

Committed by

web-flow

Commit Message

Merge 287330aab into 2ef7fe73e

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

Coverage Stats

341 of 347 new or added lines in 18 files covered. (98.27%)

6 existing lines in 4 files now uncovered.

7427 of 9842 relevant lines covered (75.46%)

0.75 hits per line

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

94.34

/src/capymoa/ocl/strategy/_lwf.py

from copy import deepcopy
from typing import Optional

import torch
from torch import Tensor, nn

from capymoa.base import BatchClassifier
from capymoa.ocl.base import TrainTaskAware
from capymoa.ocl.util.functional import hinton_distillation_loss
from capymoa.stream._stream import Schema


class LWF(BatchClassifier, nn.Module, TrainTaskAware):
    """Learning Without Forgetting (LwF).

    LwF [#f1]_ is a regularisation-based continual learning strategy that distils
    predictions from a frozen teacher snapshot of the previous task while learning the
    current task.

    ..  [#f1] Li, Z., & Hoiem, D. (2016). Learning without forgetting. CoRR,
        abs/1606.09282. http://arxiv.org/abs/1606.09282
    """

    def __init__(
        self,
        schema: Schema,
        model: torch.nn.Module,
        optimiser: torch.optim.Optimizer,
        alpha: float = 1.0,
        temperature: float = 2.0,
        device: torch.device = torch.device("cpu"),
    ) -> None:
        """Construct an LWF 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 distillation loss term.
        :param temperature: Distillation temperature.
        :param device: Compute device.
        """
        super().__init__(schema, 0)
        nn.Module.__init__(self)
        if alpha < 0:
            raise ValueError("alpha must be non-negative.")
        if temperature <= 0:
            raise ValueError("temperature must be greater than zero.")

        self.device = device

        self._alpha = alpha
        self._temperature = temperature

        self._optimiser = optimiser
        self._model = model
        self._criterion = torch.nn.CrossEntropyLoss()

        self._teacher: Optional[torch.nn.Module] = None
        self._train_task = 0

    def batch_train(self, x: Tensor, y: Tensor) -> None:
        self._model.train()
        self._optimiser.zero_grad()

        student_logits = self._model(x)
        task_loss = self._criterion(student_logits, y)
        total_loss = task_loss + self._alpha * self._distillation_loss(
            x, student_logits
        )

        total_loss.backward()
        self._optimiser.step()

    @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 on_train_task(self, task_id: int) -> None:
        if task_id > 0:
            self._teacher = (
                deepcopy(self._model).to(self.device).eval().requires_grad_(False)
            )
        self._train_task = task_id

    @torch.no_grad()
    def _teacher_forward(self, x: Tensor) -> Tensor:
        if self._teacher is None:
            raise RuntimeError("Teacher model is not available before task 1.")
        return self._teacher(x)

    def _distillation_loss(self, x: Tensor, student_logits: Tensor) -> Tensor:
        if self._teacher is None:
            return torch.tensor(0.0, device=self.device)

        teacher_logits = self._teacher_forward(x)

        return hinton_distillation_loss(
            teacher_logits=teacher_logits,
            student_logits=student_logits,
            temperature=self._temperature,
        )

    def __str__(self) -> str:
        return f"LWF(alpha={self._alpha}, temperature={self._temperature})"

1	from copy import deepcopy	1✔
2	from typing import Optional	1✔
3
4	import torch	1✔
5	from torch import Tensor, nn	1✔
6
7	from capymoa.base import BatchClassifier	1✔
8	from capymoa.ocl.base import TrainTaskAware	1✔
9	from capymoa.ocl.util.functional import hinton_distillation_loss	1✔
10	from capymoa.stream._stream import Schema	1✔
11
12
13	class LWF(BatchClassifier, nn.Module, TrainTaskAware):	1✔
14	"""Learning Without Forgetting (LwF).
15
16	LwF [#f1]_ is a regularisation-based continual learning strategy that distils
17	predictions from a frozen teacher snapshot of the previous task while learning the
18	current task.
19
20	.. [#f1] Li, Z., & Hoiem, D. (2016). Learning without forgetting. CoRR,
21	abs/1606.09282. http://arxiv.org/abs/1606.09282
22	"""
23
24	def __init__(	1✔
25	self,
26	schema: Schema,
27	model: torch.nn.Module,
28	optimiser: torch.optim.Optimizer,
29	alpha: float = 1.0,
30	temperature: float = 2.0,
31	device: torch.device = torch.device("cpu"),
32	) -> None:
33	"""Construct an LWF learner.
34
35	:param schema: Stream schema used by the classifier interface.
36	:param model: Torch model that outputs class logits.
37	:param optimiser: Optimiser used to update ``model`` parameters.
38	:param alpha: Weight of the distillation loss term.
39	:param temperature: Distillation temperature.
40	:param device: Compute device.
41	"""
42	super().__init__(schema, 0)	1✔
43	nn.Module.__init__(self)	1✔
44	if alpha < 0:	1✔
NEW 45	raise ValueError("alpha must be non-negative.")	×
46	if temperature <= 0:	1✔
NEW 47	raise ValueError("temperature must be greater than zero.")	×
48
49	self.device = device	1✔
50
51	self._alpha = alpha	1✔
52	self._temperature = temperature	1✔
53
54	self._optimiser = optimiser	1✔
55	self._model = model	1✔
56	self._criterion = torch.nn.CrossEntropyLoss()	1✔
57
58	self._teacher: Optional[torch.nn.Module] = None	1✔
59	self._train_task = 0	1✔
60
61	def batch_train(self, x: Tensor, y: Tensor) -> None:	1✔
62	self._model.train()	1✔
63	self._optimiser.zero_grad()	1✔
64
65	student_logits = self._model(x)	1✔
66	task_loss = self._criterion(student_logits, y)	1✔
67	total_loss = task_loss + self._alpha * self._distillation_loss(	1✔
68	x, student_logits
69	)
70
71	total_loss.backward()	1✔
72	self._optimiser.step()	1✔
73
74	@torch.no_grad()	1✔
75	def batch_predict_proba(self, x: Tensor) -> Tensor:	1✔
76	self._model.eval()	1✔
77	y_hat = self._model(x)	1✔
78	return torch.softmax(y_hat, dim=1)	1✔
79
80	def on_train_task(self, task_id: int) -> None:	1✔
81	if task_id > 0:	1✔
82	self._teacher = (	1✔
83	deepcopy(self._model).to(self.device).eval().requires_grad_(False)
84	)
85	self._train_task = task_id	1✔
86
87	@torch.no_grad()	1✔
88	def _teacher_forward(self, x: Tensor) -> Tensor:	1✔
89	if self._teacher is None:	1✔
NEW 90	raise RuntimeError("Teacher model is not available before task 1.")	×
91	return self._teacher(x)	1✔
92
93	def _distillation_loss(self, x: Tensor, student_logits: Tensor) -> Tensor:	1✔
94	if self._teacher is None:	1✔
95	return torch.tensor(0.0, device=self.device)	1✔
96
97	teacher_logits = self._teacher_forward(x)	1✔
98
99	return hinton_distillation_loss(	1✔
100	teacher_logits=teacher_logits,
101	student_logits=student_logits,
102	temperature=self._temperature,
103	)
104
105	def __str__(self) -> str:	1✔
106	return f"LWF(alpha={self._alpha}, temperature={self._temperature})"	1✔

adaptive-machine-learning / CapyMOA / 25192586337

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