Технології AI Написано практикуючими розробниками

Continual Learning: як навчати AI, не стираючи йому пам'ять

Оновлено: 12 хв читання 4 переглядів

Уяви: навчив модель розпізнавати котів. Ідеально. Потім вирішив додати собак. Навчив. А тепер вона забула котів.


Уяви: навчив модель розпізнавати котів. Ідеально. Потім вирішив додати собак. Навчив. А тепер вона забула котів.

Це не жарт. Це катастрофічне забування (catastrophic forgetting) — одна з найбільших невирішених проблем у машинному навчанні. Нейронні мережі, на відміну від людського мозку, не вміють накопичувати знання. Вони вміють тільки переписувати старе новим.

І якщо ти шукаєш PhD-рівня тему, яка реально важлива для майбутнього AI — ось вона. Тому що без continual learning не буде ні AGI, ні автономних агентів, ні систем, які можуть працювати роками без перенавчання з нуля.


Проблема: чому нейронні мережі забувають

Нейронні мережі не вміють вчитися поступово. Вони вміють або:

  • Вивчити все одразу (batch learning) — коли весь датасет доступний
  • Або перенавчитися на нові дані, катастрофічно забувши старі

Математично проблема виглядає так:

Коли ми оптимізуємо loss на задачі B:

θ_B* = argmin L_B(θ)

Градієнтний спуск рухає параметри в напрямку, який мінімізує L_B. Але цей напрямок може бути ортогональним або навіть протилежним до напрямку, який був оптимальним для задачі A. Результат — ваги, критичні для A, перезаписуються.

Людський мозок працює інакше:

  • Ти вивчив їздити на велосипеді — і не забув, як ходити
  • Вивчив Python — не забув English
  • Вивчив нову пісню — не забув старі

Як? Мозок має механізми консолідації пам'яті, нейрогенез, sparse representations. AI поки що так не може.


Чому це критично для агентних систем

Агенти повинні навчатися онлайн. В реальному часі. На нових даних. Без зупинки системи на перенавчання.

Приклад 1: Customer support агент

Понеділок: навчився на FAQ v1.0 (100 питань)
Вівторок: компанія випустила новий продукт, FAQ v2.0
Середа: агент повинен знати:
  - Старе (для legacy клієнтів, які ще користуються v1)
  - Нове (для клієнтів нового продукту)
  - Зв'язки між версіями (міграційні питання)

Якщо перенавчити на v2.0 — забуде v1.0. Якщо не навчати — не знатиме нового. Патова ситуація.

Приклад 2: Автономний робот

Місяць 1: навчився орієнтуватися в офісі
Місяць 2: офіс переїхав, новий layout
Місяць 3: потрібно знати обидва офіси (старий для архівних задач)

Приклад 3: Medical AI

2023: навчений на симптомах COVID-19
2024: з'явився новий штам з іншими симптомами
2025: потрібно діагностувати всі варіанти

Три сценарії Continual Learning

Van de Ven & Tolias (2019) визначили три фундаментальні сценарії:

1. Task-Incremental Learning (Task-IL)

- Модель знає, яку задачу виконує
- На inference передається task ID
- Найпростіший сценарій
- Приклад: окремі heads для кожної задачі

2. Domain-Incremental Learning (Domain-IL)

- Структура задач однакова, змінюється domain
- Task ID не потрібен
- Приклад: класифікація емоцій в різних мовах

3. Class-Incremental Learning (Class-IL)

- Нові класи додаються поступово
- Модель повинна розрізняти ВСІ бачені класи
- Найскладніший сценарій
- Приклад: розпізнавання 10 нових об'єктів щомісяця
# Візуалізація сценаріїв
class ContinualScenarios:
    """Три сценарії continual learning."""

    def task_incremental(self, x, task_id):
        """Task-IL: знаємо task_id."""
        features = self.shared_encoder(x)
        head = self.task_heads[task_id]  # Вибираємо правильний head
        return head(features)

    def domain_incremental(self, x):
        """Domain-IL: один output space, різні domains."""
        # Той самий head для всіх domains
        return self.model(x)

    def class_incremental(self, x):
        """Class-IL: зростаючий output space."""
        features = self.encoder(x)
        # Потрібно класифікувати серед ВСІХ бачених класів
        return self.growing_classifier(features)

Методи: Regularization-based

Ідея: «штрафувати» модель за зміну важливих ваг. Якщо вага критична для старих задач — не дозволяти їй сильно змінюватися.

EWC (Elastic Weight Consolidation) — Kirkpatrick et al., 2017

import torch
import torch.nn as nn
from torch.nn import functional as F

class EWC:
    """Elastic Weight Consolidation для continual learning."""

    def __init__(self, model, dataloader, lambda_ewc=1000):
        self.model = model
        self.lambda_ewc = lambda_ewc

        # Зберігаємо параметри після навчання на задачі
        self.saved_params = {}
        for name, param in model.named_parameters():
            self.saved_params[name] = param.clone().detach()

        # Обчислюємо Fisher Information Matrix (діагональ)
        self.fisher = self._compute_fisher(dataloader)

    def _compute_fisher(self, dataloader):
        """Обчислює діагональ Fisher Information Matrix."""
        fisher = {}
        for name, param in self.model.named_parameters():
            fisher[name] = torch.zeros_like(param)

        self.model.eval()
        for x, y in dataloader:
            self.model.zero_grad()
            output = self.model(x)
            # Використовуємо log-likelihood
            loss = F.cross_entropy(output, y)
            loss.backward()

            for name, param in self.model.named_parameters():
                if param.grad is not None:
                    # Fisher ≈ E[grad²]
                    fisher[name] += param.grad.data ** 2

        # Нормалізуємо
        for name in fisher:
            fisher[name] /= len(dataloader)

        return fisher

    def penalty(self):
        """Обчислює EWC penalty для поточних параметрів."""
        loss = 0
        for name, param in self.model.named_parameters():
            if name in self.fisher:
                # Штраф пропорційний важливості (Fisher) та відхиленню
                loss += (self.fisher[name] *
                        (param - self.saved_params[name]) ** 2).sum()
        return self.lambda_ewc * loss

# Використання
def train_with_ewc(model, task1_loader, task2_loader, epochs=10):
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

    # Навчаємо на задачі 1
    for epoch in range(epochs):
        for x, y in task1_loader:
            optimizer.zero_grad()
            loss = F.cross_entropy(model(x), y)
            loss.backward()
            optimizer.step()

    # Створюємо EWC після задачі 1
    ewc = EWC(model, task1_loader)

    # Навчаємо на задачі 2 з EWC penalty
    for epoch in range(epochs):
        for x, y in task2_loader:
            optimizer.zero_grad()
            ce_loss = F.cross_entropy(model(x), y)
            ewc_loss = ewc.penalty()  # Додаємо EWC regularization
            total_loss = ce_loss + ewc_loss
            total_loss.backward()
            optimizer.step()

Synaptic Intelligence (SI) — Zenke et al., 2017

class SynapticIntelligence:
    """SI трекає важливість ваг під час навчання."""

    def __init__(self, model, lambda_si=1.0, epsilon=1e-3):
        self.model = model
        self.lambda_si = lambda_si
        self.epsilon = epsilon

        # Накопичена важливість
        self.omega = {}
        # Траєкторія градієнтів
        self.W = {}
        # Попередні значення параметрів
        self.prev_params = {}

        for name, param in model.named_parameters():
            self.omega[name] = torch.zeros_like(param)
            self.W[name] = torch.zeros_like(param)
            self.prev_params[name] = param.clone().detach()

    def update_omega(self):
        """Оновлює omega після завершення задачі."""
        for name, param in self.model.named_parameters():
            delta = param.detach() - self.prev_params[name]
            # omega += W / (delta² + ε)
            self.omega[name] += self.W[name] / (delta ** 2 + self.epsilon)
            # Reset W
            self.W[name].zero_()
            self.prev_params[name] = param.clone().detach()

    def accumulate_gradients(self):
        """Накопичує градієнти для обчислення важливості."""
        for name, param in self.model.named_parameters():
            if param.grad is not None:
                # W accumulates -grad * delta
                delta = param.detach() - self.prev_params[name]
                self.W[name] -= param.grad.detach() * delta

    def penalty(self):
        """SI penalty."""
        loss = 0
        for name, param in self.model.named_parameters():
            diff = param - self.prev_params[name]
            loss += (self.omega[name] * diff ** 2).sum()
        return self.lambda_si * loss

Методи: Replay-based

Ідея: зберігати частину старих даних і «перегравати» їх разом з новими. Найпростіший і часто найефективніший підхід.

Experience Replay з резервуаром

import random
from collections import deque

class ReplayBuffer:
    """Reservoir sampling для збалансованого буфера."""

    def __init__(self, max_size=1000):
        self.buffer = []
        self.max_size = max_size
        self.seen_samples = 0

    def add(self, x, y, task_id=None):
        """Додає sample з reservoir sampling."""
        self.seen_samples += 1

        if len(self.buffer) < self.max_size:
            self.buffer.append((x.clone(), y.clone(), task_id))
        else:
            # Reservoir sampling: замінюємо з ймовірністю max_size/seen
            idx = random.randint(0, self.seen_samples - 1)
            if idx < self.max_size:
                self.buffer[idx] = (x.clone(), y.clone(), task_id)

    def sample(self, batch_size):
        """Повертає batch з буфера."""
        if len(self.buffer) < batch_size:
            return None
        samples = random.sample(self.buffer, batch_size)
        xs, ys, tasks = zip(*samples)
        return torch.stack(xs), torch.stack(ys)

class ReplayTrainer:
    """Тренер з experience replay."""

    def __init__(self, model, buffer_size=1000, replay_batch_size=32):
        self.model = model
        self.buffer = ReplayBuffer(buffer_size)
        self.replay_batch_size = replay_batch_size
        self.optimizer = torch.optim.Adam(model.parameters())

    def train_step(self, x, y, current_task):
        """Один крок навчання з replay."""
        self.optimizer.zero_grad()

        # Loss на нових даних
        output = self.model(x)
        new_loss = F.cross_entropy(output, y)

        # Loss на replay даних
        replay_batch = self.buffer.sample(self.replay_batch_size)
        if replay_batch is not None:
            replay_x, replay_y = replay_batch
            replay_output = self.model(replay_x)
            replay_loss = F.cross_entropy(replay_output, replay_y)
            total_loss = new_loss + replay_loss
        else:
            total_loss = new_loss

        total_loss.backward()
        self.optimizer.step()

        # Додаємо нові дані в буфер
        for i in range(x.size(0)):
            self.buffer.add(x[i], y[i], current_task)

        return total_loss.item()

Generative Replay — Shin et al., 2017

class GenerativeReplay:
    """Генеративний replay замість зберігання реальних даних."""

    def __init__(self, classifier, generator):
        self.classifier = classifier
        self.generator = generator  # GAN або VAE
        self.old_generator = None
        self.old_classifier = None

    def train_on_new_task(self, new_data_loader, epochs=10):
        """Навчання з генеративним replay."""
        optimizer_c = torch.optim.Adam(self.classifier.parameters())
        optimizer_g = torch.optim.Adam(self.generator.parameters())

        for epoch in range(epochs):
            for x_new, y_new in new_data_loader:
                # 1. Генеруємо псевдо-дані від старих задач
                if self.old_generator is not None:
                    z = torch.randn(x_new.size(0), self.generator.latent_dim)
                    x_old = self.old_generator.generate(z)
                    with torch.no_grad():
                        y_old = self.old_classifier(x_old).argmax(dim=1)
                else:
                    x_old, y_old = None, None

                # 2. Навчаємо classifier на нових + згенерованих даних
                optimizer_c.zero_grad()
                loss_new = F.cross_entropy(self.classifier(x_new), y_new)
                if x_old is not None:
                    loss_old = F.cross_entropy(self.classifier(x_old), y_old)
                    loss_c = loss_new + loss_old
                else:
                    loss_c = loss_new
                loss_c.backward()
                optimizer_c.step()

                # 3. Навчаємо generator на нових даних
                # (+ можливо distillation від старого generator)
                optimizer_g.zero_grad()
                loss_g = self.generator.training_loss(x_new)
                loss_g.backward()
                optimizer_g.step()

        # Зберігаємо копії для наступної задачі
        self.old_generator = copy.deepcopy(self.generator)
        self.old_classifier = copy.deepcopy(self.classifier)

Методи: Architecture-based

Ідея: виділяти окремі частини мережі для різних задач. Замість боротися з interference — уникати його структурно.

Progressive Neural Networks — Rusu et al., 2016

class ProgressiveNetwork(nn.Module):
    """Progressive Networks: нова колонка для кожної задачі."""

    def __init__(self, input_dim, hidden_dim, output_dim):
        super().__init__()
        self.columns = nn.ModuleList()
        self.laterals = nn.ModuleList()
        self.input_dim = input_dim
        self.hidden_dim = hidden_dim
        self.output_dim = output_dim

    def add_column(self):
        """Додає нову колонку для нової задачі."""
        n_cols = len(self.columns)

        # Нова колонка
        new_column = nn.ModuleList([
            nn.Linear(self.input_dim, self.hidden_dim),
            nn.Linear(self.hidden_dim, self.hidden_dim),
            nn.Linear(self.hidden_dim, self.output_dim)
        ])
        self.columns.append(new_column)

        # Lateral connections від попередніх колонок
        if n_cols > 0:
            laterals_for_column = nn.ModuleList()
            for layer_idx in range(2):  # Для кожного hidden layer
                lateral = nn.Linear(self.hidden_dim * n_cols, self.hidden_dim)
                laterals_for_column.append(lateral)
            self.laterals.append(laterals_for_column)

        # Заморожуємо попередні колонки
        for col_idx in range(n_cols):
            for param in self.columns[col_idx].parameters():
                param.requires_grad = False

    def forward(self, x, task_id):
        """Forward pass для конкретної задачі."""
        col_outputs = []

        for col_idx, column in enumerate(self.columns):
            h = x

            for layer_idx, layer in enumerate(column[:-1]):
                h = F.relu(layer(h))

                # Додаємо lateral connections
                if col_idx > 0 and layer_idx < 2:
                    # Збираємо outputs від попередніх колонок
                    prev_outputs = torch.cat(col_outputs[:col_idx], dim=1)
                    lateral_idx = col_idx - 1
                    lateral_contrib = self.laterals[lateral_idx][layer_idx](prev_outputs)
                    h = h + lateral_contrib

                if layer_idx < len(column) - 2:
                    col_outputs.append(h)

            # Final layer
            output = column[-1](h)

            if col_idx == task_id:
                return output

        raise ValueError(f"Task {task_id} not found")

PackNet — Mallya & Lazebnik, 2018

class PackNet:
    """PackNet: pruning + reuse для continual learning."""

    def __init__(self, model, prune_percentage=0.5):
        self.model = model
        self.prune_percentage = prune_percentage
        self.masks = {}  # Маски для кожної задачі

    def train_and_prune(self, task_id, train_loader, epochs=10):
        """Навчаємо на задачі, потім pruning."""
        optimizer = torch.optim.Adam(
            [p for p in self.model.parameters() if p.requires_grad]
        )

        # 1. Навчання
        for epoch in range(epochs):
            for x, y in train_loader:
                optimizer.zero_grad()
                loss = F.cross_entropy(self.model(x), y)
                loss.backward()
                optimizer.step()

        # 2. Pruning: залишаємо тільки top-k% важливих ваг
        self.masks[task_id] = {}
        for name, param in self.model.named_parameters():
            if 'weight' in name:
                # Визначаємо threshold для pruning
                importance = param.abs()
                k = int(importance.numel() * self.prune_percentage)
                threshold = importance.flatten().topk(k).values[-1]

                # Створюємо маску
                mask = (importance >= threshold).float()
                self.masks[task_id][name] = mask

                # Заморожуємо важливі ваги
                param.data *= mask

    def get_available_mask(self):
        """Повертає маску вільних ваг для нової задачі."""
        if not self.masks:
            return None

        available = {}
        for name, param in self.model.named_parameters():
            if 'weight' in name:
                used = sum(self.masks[t][name] for t in self.masks)
                available[name] = (used == 0).float()

        return available

Методи для LLM: Prompt-based Continual Learning

Для великих мовних моделей зміна ваг — дорого і ризиковано. Альтернатива — навчати prompts.

class ContinualPromptTuning:
    """Continual learning через prompt tuning."""

    def __init__(self, llm, prompt_length=20, prompt_dim=768):
        self.llm = llm
        self.prompt_length = prompt_length
        self.prompt_dim = prompt_dim

        # Pool of prompt tokens
        self.prompt_pool = nn.ParameterList()
        self.task_to_prompts = {}

    def add_task(self, task_id):
        """Додаємо prompt для нової задачі."""
        prompt = nn.Parameter(
            torch.randn(self.prompt_length, self.prompt_dim) * 0.01
        )
        self.prompt_pool.append(prompt)
        self.task_to_prompts[task_id] = len(self.prompt_pool) - 1

    def forward(self, input_ids, task_id):
        """Forward з task-specific prompt."""
        prompt_idx = self.task_to_prompts[task_id]
        prompt = self.prompt_pool[prompt_idx]

        # Prepend prompt до input embeddings
        input_embeds = self.llm.get_input_embeddings()(input_ids)
        prompt_embeds = prompt.unsqueeze(0).expand(input_embeds.size(0), -1, -1)
        combined = torch.cat([prompt_embeds, input_embeds], dim=1)

        # Forward через LLM (заморожений)
        with torch.no_grad():
            for param in self.llm.parameters():
                param.requires_grad = False

        outputs = self.llm(inputs_embeds=combined)
        return outputs

    def train_prompt(self, task_id, train_loader, epochs=5):
        """Навчаємо тільки prompt для задачі."""
        prompt_idx = self.task_to_prompts[task_id]
        optimizer = torch.optim.Adam([self.prompt_pool[prompt_idx]], lr=1e-3)

        for epoch in range(epochs):
            for input_ids, labels in train_loader:
                optimizer.zero_grad()
                outputs = self.forward(input_ids, task_id)
                loss = F.cross_entropy(
                    outputs.logits[:, self.prompt_length:, :].reshape(-1, outputs.logits.size(-1)),
                    labels.reshape(-1)
                )
                loss.backward()
                optimizer.step()

Benchmark датасети

Для класифікації зображень:

  • Split CIFAR-100: CIFAR-100 розбитий на 10 або 20 задач
  • Split ImageNet: ImageNet на послідовні задачі
  • Permuted MNIST: MNIST з різними перестановками пікселів
  • Rotated MNIST: MNIST з різними кутами повороту

Для NLP:

  • 5-Dataset NLP: AGNews, Yelp, Amazon, DBPedia, Yahoo
  • CLUE Continual: китайський NLU benchmark
  • SuperGLUE Continual: версія SuperGLUE для CL

Для RL та робототехніки:

  • Continual World: MuJoCo tasks в послідовності
  • CORA: continual RL benchmark
  • Meta-World: multi-task RL environment

Метрики оцінки

class ContinualMetrics:
    """Метрики для оцінки continual learning."""

    def __init__(self, n_tasks):
        self.n_tasks = n_tasks
        self.accuracy_matrix = np.zeros((n_tasks, n_tasks))
        # accuracy_matrix[i, j] = accuracy на задачі j після навчання на задачі i

    def update(self, trained_task, test_task, accuracy):
        """Оновлює accuracy після навчання на trained_task."""
        self.accuracy_matrix[trained_task, test_task] = accuracy

    def average_accuracy(self):
        """Середня accuracy по всіх задачах."""
        # Використовуємо тільки нижній трикутник (включно з діагоналлю)
        mask = np.tril(np.ones_like(self.accuracy_matrix))
        return (self.accuracy_matrix * mask).sum() / mask.sum()

    def forgetting(self):
        """Середнє забування."""
        forgetting_per_task = []
        for j in range(self.n_tasks - 1):
            # Максимальна accuracy на задачі j
            max_acc = self.accuracy_matrix[:, j].max()
            # Фінальна accuracy
            final_acc = self.accuracy_matrix[-1, j]
            forgetting_per_task.append(max_acc - final_acc)
        return np.mean(forgetting_per_task) if forgetting_per_task else 0

    def forward_transfer(self):
        """Forward transfer: чи допомагає старе знання новому."""
        # Порівнюємо з baseline (навчання тільки на одній задачі)
        pass

    def backward_transfer(self):
        """Backward transfer: чи покращує нова задача старі."""
        bt_per_task = []
        for j in range(self.n_tasks - 1):
            # Accuracy одразу після навчання vs фінальна
            immediate_acc = self.accuracy_matrix[j, j]
            final_acc = self.accuracy_matrix[-1, j]
            bt_per_task.append(final_acc - immediate_acc)
        return np.mean(bt_per_task) if bt_per_task else 0

Ідеї для дослідження

Для бакалавра:

  • Порівняння EWC vs Replay на Split CIFAR-10
  • Візуалізація забування в нейронних мережах
  • Імплементація базового continual learning pipeline

Для магістра:

  • Порівняння EWC vs Replay vs SI для LLM fine-tuning
  • Оптимізація розміру replay buffer (trade-off memory vs performance)
  • Continual learning для RAG-систем
  • Hybrid методи: комбінація regularization + replay
  • Continual learning для object detection

Для PhD:

  • Теоретичні межі continual learning (information theory bounds)
  • Biologically-inspired методи (як працює гіпокамп, complementary learning systems)
  • Continual learning + federated learning
  • Formal guarantees на forgetting (provable bounds)
  • Meta-learning для швидкої адаптації до нових задач без забування

Тема continual learning ідеальна для академічних робіт — вона актуальна, має чіткі метрики, і дозволяє зробити як теоретичний, так і практичний внесок. Якщо потрібна допомога з вибором конкретного напрямку або реалізацією, звертайтесь до фахівців на skp-degree.com.ua або пишіть у Telegram: @kursovi_diplomy.


Де брати матеріал

Ключові статті:

  • "Overcoming catastrophic forgetting in neural networks" (Kirkpatrick et al., 2017) — EWC
  • "Continual Learning Through Synaptic Intelligence" (Zenke et al., 2017) — SI
  • "Continual Lifelong Learning with Neural Networks: A Review" (Parisi et al., 2019)
  • "Three scenarios for continual learning" (van de Ven & Tolias, 2019)
  • "Progressive Neural Networks" (Rusu et al., 2016)
  • "Continual Learning for Language Models" (recent surveys)

Книги:

  • "Lifelong Machine Learning" (Chen & Liu) — фундаментальна книга

Фреймворки та код:

  • Avalanche: avalanche.continualai.org — найповніший PyTorch framework
  • Continuum: github.com/Continvvm/continuum — датасети для CL
  • CL-Gym: для continual RL
  • github.com/GMvandeVen/continual-learning — репозиторій з імплементаціями

Спільнота:

  • ContinualAI: continualai.org — community з papers, code, benchmarks

Складність: ? PhD-рівень (для фундаментальних досліджень), ? Магістр (для applied)

Чому складно:

  • Немає універсального рішення — кожен метод має trade-offs
  • Потрібне глибоке розуміння оптимізації та теорії навчання
  • Багато математики (Fisher Information, KL divergence, information theory)
  • Експерименти вимагають значних GPU resources
  • Оцінка результатів non-trivial (багато метрик)

Для магістра: можна взяти вужчу підзадачу:

  • Один конкретний метод (EWC або Replay)
  • Один конкретний домен (тільки NLP або тільки CV)
  • Practical application (CL для конкретної бізнес-задачі)

Чому це майбутнє AI

AGI неможливий без continual learning. Система, яка не може навчатися новому без забування старого — це не інтелект. Це калькулятор з фіксованою програмою.

Подумай про це:

  • Людина вчиться все життя, не перезавантажуючи мозок
  • Кожен досвід інтегрується з попереднім
  • Знання накопичуються, а не замінюються

AI поки так не може. Але хто вирішить цю проблему — зробить прорив рівня transformer. Можливо, це будеш ти.

Ключові слова: continual learning, lifelong learning, catastrophic forgetting, EWC, Elastic Weight Consolidation, experience replay, progressive neural networks, prompt tuning, neural networks, machine learning, deep learning, дипломна робота, магістерська, PhD, дослідження AI.

Про автора

Команда SKP-Degree

Верифікований автор

Розробники та дослідники AI · Python, TensorFlow, PyTorch · Досвід у промисловій розробці

Команда SKP-Degree — професійні розробники з досвідом 7+ років у промисловій розробці. Виконали 1000+ проєктів для студентів з України, Польщі та країн Балтії.

Python Django Java ML/AI React C# / .NET JavaScript

Потрібна допомога з роботою?

Замовте курсову чи дипломну роботу з програмування. Оплата після демонстрації!

Без передоплати Відеодемонстрація Автономна робота 24/7
Написати в Telegram