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.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) [#f1]_ .

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