Catastrophic Forgetting › Implementing Forgetting Measurement

import warnings

warnings.filterwarnings("ignore")

# Create a simple dataset for demonstration
from torch.utils.data import Dataset


class SentimentDataset(Dataset):
    def __init__(self, tokenizer, num_samples=200):
        self.tokenizer = tokenizer
        # Simple positive/negative examples
        positive = [
            "This movie was absolutely wonderful and amazing.",
            "I loved every minute of this fantastic film.",
            "Brilliant performances and excellent direction.",
            "A masterpiece of modern cinema.",
            "Outstanding work from the entire cast.",
        ]
        negative = [
            "This was the worst movie I have ever seen.",
            "Terrible acting and boring plot throughout.",
            "I regret wasting my time on this film.",
            "Completely disappointing and poorly made.",
            "A total disaster from start to finish.",
        ]

        # Expand dataset
        self.texts = (positive * (num_samples // 10)) + (
            negative * (num_samples // 10)
        )
        self.labels = [1] * (num_samples // 2) + [0] * (num_samples // 2)

    def __len__(self):
        return len(self.texts)

    def __getitem__(self, idx):
        encoding = self.tokenizer(
            self.texts[idx],
            truncation=True,
            padding="max_length",
            max_length=64,
            return_tensors="pt",
        )
        return {
            "input_ids": encoding["input_ids"].squeeze(),
            "attention_mask": encoding["attention_mask"].squeeze(),
            "labels": torch.tensor(self.labels[idx]),
        }

def measure_language_modeling_capability(model, tokenizer, test_texts):
    """Measure model's language modeling capability via pseudo-perplexity."""
    model.eval()
    total_loss = 0
    total_tokens = 0

    device = next(model.parameters()).device

    with torch.no_grad():
        for text in test_texts:
            encodings = tokenizer(
                text, return_tensors="pt", truncation=True, max_length=128
            )
            input_ids = encodings["input_ids"].to(device)

            # For classification model, use embedding similarity as proxy
            # In practice, you'd use a separate LM head or the original model
            outputs = model.base_model(input_ids)
            hidden_states = outputs.last_hidden_state

            # Measure representation quality via hidden state statistics
            total_loss += hidden_states.var().item()
            total_tokens += input_ids.size(1)

    return total_loss / len(test_texts)

def train_epoch_with_monitoring(
    model, dataloader, optimizer, device, tokenizer, lm_test_texts
):
    """Train one epoch and return metrics."""
    model.train()
    total_loss = 0
    correct = 0
    total = 0

    for batch in dataloader:
        input_ids = batch["input_ids"].to(device)
        attention_mask = batch["attention_mask"].to(device)
        labels = batch["labels"].to(device)

        optimizer.zero_grad()
        outputs = model(
            input_ids=input_ids, attention_mask=attention_mask, labels=labels
        )
        loss = outputs.loss
        loss.backward()
        optimizer.step()

        total_loss += loss.item()
        preds = outputs.logits.argmax(dim=-1)
        correct += (preds == labels).sum().item()
        total += labels.size(0)

    train_acc = correct / total
    avg_loss = total_loss / len(dataloader)

    # Measure language modeling capability
    lm_score = measure_language_modeling_capability(
        model, tokenizer, lm_test_texts
    )

    return avg_loss, train_acc, lm_score

import torch
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer

# Use a small model for demonstration
model_name = "distilbert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(
    model_name, num_labels=2
)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)

# Create dataset and dataloader
dataset = SentimentDataset(tokenizer, num_samples=200)
dataloader = DataLoader(dataset, batch_size=16, shuffle=True)

# Test texts for monitoring language modeling capability
lm_test_texts = [
    "The quick brown fox jumps over the lazy dog.",
    "Natural language processing enables computers to understand human language.",
    "Machine learning algorithms improve through experience and data.",
    "The weather forecast predicts rain tomorrow afternoon.",
    "Scientists discovered a new species in the Amazon rainforest.",
]

# Measure baseline
baseline_lm_score = measure_language_modeling_capability(
    model, tokenizer, lm_test_texts
)

# Fine-tune and track forgetting
optimizer = torch.optim.AdamW(model.parameters(), lr=3e-5)
num_epochs = 15

history = {
    "train_loss": [],
    "train_acc": [],
    "lm_score": [],
    "lm_score_change": [],
}

for epoch in range(num_epochs):
    train_loss, train_acc, lm_score = train_epoch_with_monitoring(
        model, dataloader, optimizer, device, tokenizer, lm_test_texts
    )

    history["train_loss"].append(train_loss)
    history["train_acc"].append(train_acc)
    history["lm_score"].append(lm_score)
    history["lm_score_change"].append(
        (lm_score - baseline_lm_score) / baseline_lm_score * 100
    )