WHAT IS CURRICULUM LEARNING?
Curriculum learning is a training strategy in machine learning where you teach a model by presenting training examples in a meaningful order, starting from easier examples and gradually progressing to more difficult ones. This approach mimics how humans learn, where we typically master simple concepts before tackling complex ones.
Imagine teaching a child mathematics. You would not start with calculus. Instead, you would begin with counting, then addition, then subtraction, and slowly build up to more advanced topics. Curriculum learning applies this same principle to training artificial intelligence systems.
The fundamental idea is that the order in which a machine learning model sees training data can significantly impact how well and how quickly it learns. By carefully organizing training examples from simple to complex, we can often achieve better final performance, faster convergence, and improved generalization compared to randomly shuffling all training data.
WHY USE CURRICULUM LEARNING?
Traditional machine learning training typically shuffles all available training data and presents it randomly to the model. While this works, it has limitations. When a model encounters extremely difficult examples early in training, it may struggle to make progress because it has not yet developed the foundational patterns needed to understand these complex cases.
Curriculum learning addresses several important challenges. First, it can help models converge faster by building knowledge incrementally. Second, it can lead to better final performance by ensuring the model develops robust foundational representations before tackling edge cases. Third, it can improve training stability by avoiding overwhelming the model with complexity too early.
Consider training a computer vision model to recognize objects in cluttered scenes. If we start with clear, well-lit images of single objects, the model can learn basic shape and color patterns. Once it masters these fundamentals, we can introduce images with multiple objects, then images with occlusions, and finally images with poor lighting and heavy clutter. This progression allows the model to build capabilities systematically.
THE CORE COMPONENTS OF CURRICULUM LEARNING
A curriculum learning system consists of several essential components that work together to implement the progressive training strategy.
The first component is the difficulty measurer. This component assigns a difficulty score to each training example. The difficulty measure can be based on various factors such as the complexity of the input, the rarity of the label, or even the model's current performance on similar examples.
The second component is the pacing function. This determines how quickly the curriculum should progress from easy to hard examples. Some curricula use a fixed schedule, while others adapt based on the model's learning progress.
The third component is the data scheduler. This takes the difficulty scores and pacing function to decide which examples should be presented to the model at each training step. The scheduler might use strategies like filtering out examples above a certain difficulty threshold or adjusting the sampling probability based on difficulty.
The fourth component is the training loop itself, which integrates these elements with the standard model training process. The training loop must coordinate between curriculum progression and model optimization.
MEASURING EXAMPLE DIFFICULTY
Determining which examples are easy and which are hard is a crucial challenge in curriculum learning. There are several approaches to measuring difficulty, each with different strengths and use cases.
One straightforward approach is to use predefined heuristics based on domain knowledge. For instance, in language learning, sentence length might serve as a difficulty proxy. Shorter sentences are generally easier to process than longer ones. In image classification, images with clear backgrounds might be considered easier than those with cluttered scenes.
Here is a simple example of a heuristic-based difficulty scorer for text data:
class TextDifficultyScorer:
def __init__(self, vocab_frequency):
# vocab_frequency is a dictionary mapping words to their corpus frequency
self.vocab_frequency = vocab_frequency
def score(self, text):
# Tokenize the text into words
words = text.lower().split()
# Calculate difficulty based on length and rare words
length_score = len(words) / 100.0 # Normalize by expected max length
# Calculate rarity score (inverse of average word frequency)
rarity_scores = []
for word in words:
freq = self.vocab_frequency.get(word, 0.0001) # Default for unknown words
rarity_scores.append(1.0 / (freq + 0.0001))
avg_rarity = sum(rarity_scores) / max(len(rarity_scores), 1)
# Combine scores (you can adjust weights)
difficulty = 0.3 * length_score + 0.7 * avg_rarity
return difficulty
This scorer combines two factors: the length of the text and the rarity of words it contains. Longer texts with rare words receive higher difficulty scores.
Another approach is to use model-based difficulty estimation. In this method, we train a separate model or use the current model's predictions to estimate difficulty. Examples where the model makes confident correct predictions are considered easy, while examples where the model is uncertain or incorrect are considered hard.
A third approach is to use loss-based difficulty. We can measure how much loss the model incurs on each example. Examples with high loss are difficult, while those with low loss are easy. This approach has the advantage of being adaptive: as the model learns, what was once difficult may become easy.
Now let us learn an example of a loss-based difficulty tracker:
import numpy as np
class LossBasedDifficultyTracker:
def __init__(self, smoothing_factor=0.9):
# Store moving average of loss for each example
self.example_losses = {}
self.smoothing_factor = smoothing_factor
def update(self, example_id, loss_value):
# Update the moving average loss for this example
if example_id not in self.example_losses:
self.example_losses[example_id] = loss_value
else:
# Exponential moving average
old_loss = self.example_losses[example_id]
new_loss = self.smoothing_factor * old_loss + (1 - self.smoothing_factor) * loss_value
self.example_losses[example_id] = new_loss
def get_difficulty(self, example_id):
# Return the current difficulty estimate
return self.example_losses.get(example_id, float('inf'))
def get_difficulty_percentile(self, example_id):
# Return what percentile this example falls into
if example_id not in self.example_losses:
return 1.0 # Treat unseen examples as hardest
all_losses = list(self.example_losses.values())
example_loss = self.example_losses[example_id]
# Calculate percentile
percentile = sum(1 for loss in all_losses if loss <= example_loss) / len(all_losses)
return percentile
This tracker maintains a moving average of the loss for each training example. As training progresses, it updates these estimates, allowing the curriculum to adapt to the model's changing capabilities.
PACING STRATEGIES FOR CURRICULUM PROGRESSION
Once we have difficulty scores for our training examples, we need to decide how to pace the curriculum. The pacing strategy determines how quickly we transition from easy to hard examples.
The simplest approach is a fixed linear schedule. We might start by only showing examples in the easiest twenty percent, then after a certain number of training steps, expand to the easiest forty percent, and so on until we are using all examples.
An implementation of a fixed linear pacing function:
class LinearPacingFunction:
def __init__(self, total_steps, start_percentile=0.2, end_percentile=1.0):
# total_steps: how many training steps until we use all data
# start_percentile: what fraction of easiest data to start with
# end_percentile: what fraction to end with (usually 1.0 for all data)
self.total_steps = total_steps
self.start_percentile = start_percentile
self.end_percentile = end_percentile
def get_difficulty_threshold(self, current_step):
# Calculate what percentile of data we should include at this step
if current_step >= self.total_steps:
return self.end_percentile
progress = current_step / self.total_steps
threshold = self.start_percentile + progress * (self.end_percentile - self.start_percentile)
return threshold
This pacing function starts by allowing only the easiest twenty percent of examples and linearly increases this threshold until all examples are included after the specified number of training steps.
A more sophisticated approach is self-paced learning, where the curriculum adapts based on the model's performance. If the model is learning quickly and achieving low loss, the curriculum might accelerate and introduce harder examples sooner. If the model struggles, the curriculum might slow down and spend more time on easier examples.
Let us view a self-paced learning implementation:
class SelfPacedCurriculum:
def __init__(self, initial_threshold=0.2, growth_rate=0.01, performance_window=100):
# initial_threshold: starting difficulty percentile
# growth_rate: how much to increase threshold when performing well
# performance_window: how many recent steps to consider for performance
self.current_threshold = initial_threshold
self.growth_rate = growth_rate
self.performance_window = performance_window
self.recent_losses = []
self.loss_trend = None
def update(self, current_loss):
# Track recent losses to determine if model is improving
self.recent_losses.append(current_loss)
# Keep only the most recent losses
if len(self.recent_losses) > self.performance_window:
self.recent_losses.pop(0)
# Calculate loss trend (negative means improving)
if len(self.recent_losses) >= 2:
recent_avg = np.mean(self.recent_losses[-20:]) if len(self.recent_losses) >= 20 else self.recent_losses[-1]
older_avg = np.mean(self.recent_losses[:20]) if len(self.recent_losses) >= 40 else self.recent_losses[0]
self.loss_trend = recent_avg - older_avg
def get_difficulty_threshold(self):
# Adjust threshold based on learning progress
if self.loss_trend is not None and self.loss_trend < 0:
# Model is improving, increase difficulty
self.current_threshold = min(1.0, self.current_threshold + self.growth_rate)
elif self.loss_trend is not None and self.loss_trend > 0:
# Model is struggling, slow down or maintain current difficulty
self.current_threshold = max(0.1, self.current_threshold - self.growth_rate * 0.5)
return self.current_threshold
This self-paced curriculum monitors the model's recent loss trend. When the model is improving, it increases the difficulty threshold to introduce harder examples. When the model struggles, it reduces the threshold or maintains the current difficulty level.
IMPLEMENTING THE DATA SCHEDULER
The data scheduler is responsible for selecting which training examples to present at each step based on the difficulty scores and pacing function. There are several strategies for implementing this selection.
One approach is hard filtering, where we completely exclude examples above the current difficulty threshold. This ensures the model only sees examples it is ready for, but it can be wasteful because we discard potentially useful data.
Another approach is soft filtering or importance sampling, where we adjust the probability of sampling each example based on its difficulty. Easier examples get higher sampling probability, but harder examples are not completely excluded. This allows the model to occasionally encounter challenging examples while focusing primarily on appropriate difficulty levels.
Here comes an implementation of a curriculum data scheduler with both hard and soft filtering options:
import random
class CurriculumDataScheduler:
def __init__(self, dataset, difficulty_scorer, pacing_function, mode='soft'):
# dataset: list of training examples with unique IDs
# difficulty_scorer: object that can score example difficulty
# pacing_function: object that determines current difficulty threshold
# mode: 'hard' for filtering, 'soft' for importance sampling
self.dataset = dataset
self.difficulty_scorer = difficulty_scorer
self.pacing_function = pacing_function
self.mode = mode
# Pre-compute difficulty scores for all examples
self.difficulty_scores = {}
for example in dataset:
example_id = example['id']
self.difficulty_scores[example_id] = difficulty_scorer.score(example)
# Normalize scores to [0, 1] range
max_score = max(self.difficulty_scores.values())
min_score = min(self.difficulty_scores.values())
score_range = max_score - min_score
if score_range > 0:
for example_id in self.difficulty_scores:
normalized = (self.difficulty_scores[example_id] - min_score) / score_range
self.difficulty_scores[example_id] = normalized
def get_batch(self, batch_size, current_step):
# Get current difficulty threshold from pacing function
threshold = self.pacing_function.get_difficulty_threshold(current_step)
if self.mode == 'hard':
# Hard filtering: only include examples below threshold
eligible_examples = [
ex for ex in self.dataset
if self.difficulty_scores[ex['id']] <= threshold
]
if len(eligible_examples) < batch_size:
# Not enough easy examples, use what we have
batch = eligible_examples
else:
# Randomly sample from eligible examples
batch = random.sample(eligible_examples, batch_size)
else: # soft mode
# Soft filtering: sample with probability inversely proportional to difficulty
sampling_weights = []
for example in self.dataset:
difficulty = self.difficulty_scores[example['id']]
# Examples at or below threshold get full weight
# Examples above threshold get reduced weight
if difficulty <= threshold:
weight = 1.0
else:
# Exponentially decay weight for harder examples
excess_difficulty = difficulty - threshold
weight = np.exp(-5.0 * excess_difficulty)
sampling_weights.append(weight)
# Normalize weights to probabilities
total_weight = sum(sampling_weights)
probabilities = [w / total_weight for w in sampling_weights]
# Sample according to these probabilities
indices = np.random.choice(
len(self.dataset),
size=batch_size,
replace=False,
p=probabilities
)
batch = [self.dataset[i] for i in indices]
return batch
This scheduler can operate in two modes. In hard filtering mode, it completely excludes examples above the difficulty threshold. In soft filtering mode, it uses importance sampling to preferentially select easier examples while still occasionally including harder ones.
INTEGRATING CURRICULUM LEARNING INTO TRAINING
Now that we have all the components, we need to integrate them into the actual training loop. The training loop must coordinate between curriculum progression, batch selection, model updates, and difficulty tracking.
Here is a skeleton of a curriculum learning training loop:
def train_with_curriculum(model, optimizer, dataset, difficulty_scorer,
pacing_function, num_epochs, batch_size):
# Initialize the curriculum scheduler
scheduler = CurriculumDataScheduler(
dataset,
difficulty_scorer,
pacing_function,
mode='soft'
)
# Track training progress
global_step = 0
for epoch in range(num_epochs):
epoch_loss = 0.0
num_batches = len(dataset) // batch_size
for batch_idx in range(num_batches):
# Get a curriculum-based batch
batch = scheduler.get_batch(batch_size, global_step)
# Forward pass
predictions = model(batch)
loss = compute_loss(predictions, batch)
# Backward pass and optimization
optimizer.zero_grad()
loss.backward()
optimizer.step()
# Update difficulty scores if using adaptive scoring
if hasattr(difficulty_scorer, 'update'):
for example in batch:
example_loss = compute_example_loss(model, example)
difficulty_scorer.update(example['id'], example_loss)
# Update pacing function if using self-paced learning
if hasattr(pacing_function, 'update'):
pacing_function.update(loss.item())
epoch_loss += loss.item()
global_step += 1
avg_epoch_loss = epoch_loss / num_batches
print(f"Epoch {epoch + 1}/{num_epochs}, Loss: {avg_epoch_loss:.4f}")
return model
This training loop integrates all curriculum learning components. It uses the scheduler to get appropriately difficult batches, performs standard model training, and updates both the difficulty scorer and pacing function based on training progress.
PRACTICAL CONSIDERATIONS AND BEST PRACTICES
When implementing curriculum learning in practice, several important considerations can affect success.
First, the choice of difficulty measure is crucial and domain-dependent. For some tasks, simple heuristics work well. For others, adaptive loss-based measures are necessary. It is often beneficial to experiment with multiple difficulty measures and compare their effectiveness.
Second, the pacing schedule requires careful tuning. If the curriculum progresses too quickly, the model may not have time to master easier concepts before encountering harder ones. If it progresses too slowly, training time increases without corresponding benefits. Self-paced learning can help automate this tuning but introduces its own hyperparameters.
Third, curriculum learning interacts with other training techniques. When using techniques like data augmentation, learning rate schedules, or regularization, these must be coordinated with the curriculum. For example, you might want to increase data augmentation as the curriculum introduces harder examples.
Fourth, not all tasks benefit equally from curriculum learning. Tasks with clear difficulty hierarchies and where foundational concepts enable learning of advanced concepts tend to benefit most. Tasks where examples are relatively uniform in difficulty may see little benefit.
Fifth, evaluation is important. You should compare curriculum learning against standard random shuffling on your specific task. Measure not just final performance but also convergence speed and training stability.
FULL PRODUCTION-READY RUNNING EXAMPLE
Now I will present a complete, production-ready implementation of curriculum learning for a text classification task. This implementation includes all necessary components and can be adapted to various use cases.
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from collections import defaultdict
import random
from typing import List, Dict, Tuple, Optional, Callable
import json
class TextDataset(Dataset):
"""
A dataset class for text classification that supports curriculum learning.
Each example has a unique ID for tracking difficulty scores.
"""
def __init__(self, texts: List[str], labels: List[int], vocab: Dict[str, int]):
"""
Initialize the dataset.
Args:
texts: List of text strings
labels: List of integer labels
vocab: Dictionary mapping words to integer indices
"""
self.texts = texts
self.labels = labels
self.vocab = vocab
self.max_length = 100
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
text = self.texts[idx]
label = self.labels[idx]
# Tokenize and convert to indices
tokens = text.lower().split()
indices = [self.vocab.get(token, self.vocab['<UNK>']) for token in tokens]
# Pad or truncate to max_length
if len(indices) < self.max_length:
indices = indices + [self.vocab['<PAD>']] * (self.max_length - len(indices))
else:
indices = indices[:self.max_length]
return {
'id': idx,
'text': text,
'indices': torch.tensor(indices, dtype=torch.long),
'label': torch.tensor(label, dtype=torch.long)
}
class TextClassifier(nn.Module):
"""
A simple LSTM-based text classifier.
"""
def __init__(self, vocab_size: int, embedding_dim: int, hidden_dim: int,
num_classes: int, dropout: float = 0.3):
"""
Initialize the classifier.
Args:
vocab_size: Size of the vocabulary
embedding_dim: Dimension of word embeddings
hidden_dim: Dimension of LSTM hidden state
num_classes: Number of output classes
dropout: Dropout probability
"""
super(TextClassifier, self).__init__()
self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=0)
self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True,
num_layers=2, dropout=dropout, bidirectional=True)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_dim * 2, num_classes)
def forward(self, indices):
"""
Forward pass.
Args:
indices: Tensor of shape (batch_size, max_length) containing word indices
Returns:
Tensor of shape (batch_size, num_classes) containing class logits
"""
# Embed the input
embedded = self.embedding(indices) # (batch_size, max_length, embedding_dim)
# Pass through LSTM
lstm_out, (hidden, cell) = self.lstm(embedded)
# Use the final hidden states from both directions
# hidden shape: (num_layers * num_directions, batch_size, hidden_dim)
forward_hidden = hidden[-2, :, :]
backward_hidden = hidden[-1, :, :]
combined_hidden = torch.cat([forward_hidden, backward_hidden], dim=1)
# Apply dropout and final linear layer
dropped = self.dropout(combined_hidden)
logits = self.fc(dropped)
return logits
class HeuristicDifficultyScorer:
"""
Scores text difficulty based on heuristics like length and word rarity.
"""
def __init__(self, vocab_frequency: Dict[str, float]):
"""
Initialize the scorer.
Args:
vocab_frequency: Dictionary mapping words to their frequency in corpus
"""
self.vocab_frequency = vocab_frequency
def score(self, example: Dict) -> float:
"""
Compute difficulty score for an example.
Args:
example: Dictionary containing 'text' key
Returns:
Float difficulty score (higher = more difficult)
"""
text = example['text']
words = text.lower().split()
if len(words) == 0:
return 0.0
# Length component (normalized)
length_score = min(len(words) / 50.0, 1.0)
# Rarity component (average inverse frequency)
rarity_scores = []
for word in words:
freq = self.vocab_frequency.get(word, 0.00001)
rarity_scores.append(1.0 / (freq + 0.00001))
avg_rarity = np.mean(rarity_scores)
# Normalize rarity score
rarity_score = min(avg_rarity / 1000.0, 1.0)
# Combine scores
difficulty = 0.4 * length_score + 0.6 * rarity_score
return difficulty
class AdaptiveDifficultyScorer:
"""
Scores difficulty based on model's current loss on each example.
Adapts as the model learns.
"""
def __init__(self, smoothing_factor: float = 0.9, initial_difficulty: float = 0.5):
"""
Initialize the scorer.
Args:
smoothing_factor: Factor for exponential moving average (0-1)
initial_difficulty: Initial difficulty for unseen examples
"""
self.smoothing_factor = smoothing_factor
self.initial_difficulty = initial_difficulty
self.example_losses = {}
def update(self, example_id: int, loss_value: float):
"""
Update the difficulty estimate for an example.
Args:
example_id: Unique identifier for the example
loss_value: Current loss value on this example
"""
if example_id not in self.example_losses:
self.example_losses[example_id] = loss_value
else:
old_loss = self.example_losses[example_id]
new_loss = (self.smoothing_factor * old_loss +
(1 - self.smoothing_factor) * loss_value)
self.example_losses[example_id] = new_loss
def score(self, example: Dict) -> float:
"""
Get difficulty score for an example.
Args:
example: Dictionary containing 'id' key
Returns:
Float difficulty score (higher = more difficult)
"""
example_id = example['id']
if example_id not in self.example_losses:
return self.initial_difficulty
# Normalize loss to [0, 1] range
all_losses = list(self.example_losses.values())
if len(all_losses) == 0:
return self.initial_difficulty
max_loss = max(all_losses)
min_loss = min(all_losses)
if max_loss == min_loss:
return 0.5
example_loss = self.example_losses[example_id]
normalized_difficulty = (example_loss - min_loss) / (max_loss - min_loss)
return normalized_difficulty
class FixedPacingFunction:
"""
Fixed linear pacing schedule that increases difficulty threshold over time.
"""
def __init__(self, total_steps: int, start_percentile: float = 0.2,
end_percentile: float = 1.0):
"""
Initialize the pacing function.
Args:
total_steps: Number of steps to reach end_percentile
start_percentile: Initial difficulty threshold (0-1)
end_percentile: Final difficulty threshold (0-1)
"""
self.total_steps = total_steps
self.start_percentile = start_percentile
self.end_percentile = end_percentile
def get_difficulty_threshold(self, current_step: int) -> float:
"""
Get the current difficulty threshold.
Args:
current_step: Current training step
Returns:
Float threshold value (0-1)
"""
if current_step >= self.total_steps:
return self.end_percentile
progress = current_step / self.total_steps
threshold = (self.start_percentile +
progress * (self.end_percentile - self.start_percentile))
return threshold
class AdaptivePacingFunction:
"""
Self-paced learning that adjusts difficulty based on model performance.
"""
def __init__(self, initial_threshold: float = 0.2, min_threshold: float = 0.1,
max_threshold: float = 1.0, growth_rate: float = 0.005,
performance_window: int = 100):
"""
Initialize the pacing function.
Args:
initial_threshold: Starting difficulty threshold
min_threshold: Minimum allowed threshold
max_threshold: Maximum allowed threshold
growth_rate: How much to adjust threshold each step
performance_window: Number of recent losses to track
"""
self.current_threshold = initial_threshold
self.min_threshold = min_threshold
self.max_threshold = max_threshold
self.growth_rate = growth_rate
self.performance_window = performance_window
self.recent_losses = []
def update(self, current_loss: float):
"""
Update the pacing based on current performance.
Args:
current_loss: Current training loss
"""
self.recent_losses.append(current_loss)
if len(self.recent_losses) > self.performance_window:
self.recent_losses.pop(0)
# Calculate if model is improving
if len(self.recent_losses) >= 20:
recent_avg = np.mean(self.recent_losses[-10:])
older_avg = np.mean(self.recent_losses[:10])
if recent_avg < older_avg:
# Model improving, increase difficulty
self.current_threshold = min(
self.max_threshold,
self.current_threshold + self.growth_rate
)
else:
# Model struggling, decrease difficulty slightly
self.current_threshold = max(
self.min_threshold,
self.current_threshold - self.growth_rate * 0.3
)
def get_difficulty_threshold(self, current_step: int = None) -> float:
"""
Get the current difficulty threshold.
Args:
current_step: Not used in adaptive pacing, kept for interface compatibility
Returns:
Float threshold value (0-1)
"""
return self.current_threshold
class CurriculumScheduler:
"""
Schedules training batches according to curriculum learning strategy.
"""
def __init__(self, dataset: Dataset, difficulty_scorer,
pacing_function, mode: str = 'soft'):
"""
Initialize the scheduler.
Args:
dataset: Dataset to schedule
difficulty_scorer: Object with score(example) method
pacing_function: Object with get_difficulty_threshold(step) method
mode: 'hard' for filtering, 'soft' for importance sampling
"""
self.dataset = dataset
self.difficulty_scorer = difficulty_scorer
self.pacing_function = pacing_function
self.mode = mode
# Pre-compute initial difficulty scores
self.difficulty_scores = {}
self._update_all_difficulty_scores()
def _update_all_difficulty_scores(self):
"""
Recompute difficulty scores for all examples.
"""
for idx in range(len(self.dataset)):
example = self.dataset[idx]
self.difficulty_scores[idx] = self.difficulty_scorer.score(example)
def get_batch_indices(self, batch_size: int, current_step: int) -> List[int]:
"""
Get indices for the next batch according to curriculum.
Args:
batch_size: Number of examples to return
current_step: Current training step
Returns:
List of dataset indices
"""
threshold = self.pacing_function.get_difficulty_threshold(current_step)
if self.mode == 'hard':
# Hard filtering: only include examples below threshold
eligible_indices = [
idx for idx in range(len(self.dataset))
if self.difficulty_scores[idx] <= threshold
]
if len(eligible_indices) == 0:
# Fallback: use all examples if none are eligible
eligible_indices = list(range(len(self.dataset)))
if len(eligible_indices) <= batch_size:
return eligible_indices
else:
return random.sample(eligible_indices, batch_size)
else: # soft mode
# Importance sampling based on difficulty
weights = []
for idx in range(len(self.dataset)):
difficulty = self.difficulty_scores[idx]
if difficulty <= threshold:
weight = 1.0
else:
# Exponentially decay weight for harder examples
excess = difficulty - threshold
weight = np.exp(-5.0 * excess)
weights.append(weight)
# Normalize to probabilities
total_weight = sum(weights)
if total_weight == 0:
# Fallback to uniform sampling
probabilities = [1.0 / len(weights)] * len(weights)
else:
probabilities = [w / total_weight for w in weights]
# Sample indices
indices = np.random.choice(
len(self.dataset),
size=min(batch_size, len(self.dataset)),
replace=False,
p=probabilities
)
return indices.tolist()
class CurriculumLearningTrainer:
"""
Main trainer class that orchestrates curriculum learning.
"""
def __init__(self, model: nn.Module, dataset: Dataset,
difficulty_scorer, pacing_function,
device: str = 'cpu', scheduler_mode: str = 'soft'):
"""
Initialize the trainer.
Args:
model: PyTorch model to train
dataset: Training dataset
difficulty_scorer: Difficulty scoring object
pacing_function: Pacing function object
device: Device to train on ('cpu' or 'cuda')
scheduler_mode: 'hard' or 'soft' filtering
"""
self.model = model.to(device)
self.dataset = dataset
self.difficulty_scorer = difficulty_scorer
self.pacing_function = pacing_function
self.device = device
self.scheduler = CurriculumScheduler(
dataset, difficulty_scorer, pacing_function, scheduler_mode
)
self.criterion = nn.CrossEntropyLoss(reduction='none')
self.global_step = 0
def train_epoch(self, optimizer: optim.Optimizer, batch_size: int,
update_difficulty: bool = True) -> Tuple[float, float]:
"""
Train for one epoch.
Args:
optimizer: PyTorch optimizer
batch_size: Batch size
update_difficulty: Whether to update difficulty scores
Returns:
Tuple of (average_loss, accuracy)
"""
self.model.train()
total_loss = 0.0
total_correct = 0
total_examples = 0
num_batches = len(self.dataset) // batch_size
for batch_idx in range(num_batches):
# Get curriculum-based batch
indices = self.scheduler.get_batch_indices(batch_size, self.global_step)
# Gather batch data
batch_data = [self.dataset[idx] for idx in indices]
# Stack tensors
input_indices = torch.stack([item['indices'] for item in batch_data]).to(self.device)
labels = torch.stack([item['label'] for item in batch_data]).to(self.device)
# Forward pass
optimizer.zero_grad()
logits = self.model(input_indices)
# Compute loss
losses = self.criterion(logits, labels)
loss = losses.mean()
# Backward pass
loss.backward()
optimizer.step()
# Update difficulty scores if using adaptive scoring
if update_difficulty and hasattr(self.difficulty_scorer, 'update'):
for i, idx in enumerate(indices):
example_loss = losses[i].item()
self.difficulty_scorer.update(idx, example_loss)
# Update pacing function if adaptive
if hasattr(self.pacing_function, 'update'):
self.pacing_function.update(loss.item())
# Track metrics
total_loss += loss.item()
predictions = torch.argmax(logits, dim=1)
total_correct += (predictions == labels).sum().item()
total_examples += len(labels)
self.global_step += 1
avg_loss = total_loss / num_batches
accuracy = total_correct / total_examples
return avg_loss, accuracy
def evaluate(self, eval_dataset: Dataset, batch_size: int) -> Tuple[float, float]:
"""
Evaluate the model on a dataset.
Args:
eval_dataset: Dataset to evaluate on
batch_size: Batch size for evaluation
Returns:
Tuple of (average_loss, accuracy)
"""
self.model.eval()
total_loss = 0.0
total_correct = 0
total_examples = 0
with torch.no_grad():
for start_idx in range(0, len(eval_dataset), batch_size):
end_idx = min(start_idx + batch_size, len(eval_dataset))
batch_data = [eval_dataset[idx] for idx in range(start_idx, end_idx)]
input_indices = torch.stack([item['indices'] for item in batch_data]).to(self.device)
labels = torch.stack([item['label'] for item in batch_data]).to(self.device)
logits = self.model(input_indices)
losses = self.criterion(logits, labels)
total_loss += losses.sum().item()
predictions = torch.argmax(logits, dim=1)
total_correct += (predictions == labels).sum().item()
total_examples += len(labels)
avg_loss = total_loss / total_examples
accuracy = total_correct / total_examples
return avg_loss, accuracy
def train(self, num_epochs: int, batch_size: int, learning_rate: float,
eval_dataset: Optional[Dataset] = None,
eval_interval: int = 1) -> Dict[str, List[float]]:
"""
Full training loop.
Args:
num_epochs: Number of epochs to train
batch_size: Batch size
learning_rate: Learning rate
eval_dataset: Optional evaluation dataset
eval_interval: How often to evaluate (in epochs)
Returns:
Dictionary containing training history
"""
optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)
history = {
'train_loss': [],
'train_accuracy': [],
'eval_loss': [],
'eval_accuracy': [],
'difficulty_threshold': []
}
for epoch in range(num_epochs):
# Train for one epoch
train_loss, train_acc = self.train_epoch(optimizer, batch_size)
history['train_loss'].append(train_loss)
history['train_accuracy'].append(train_acc)
# Track current difficulty threshold
current_threshold = self.pacing_function.get_difficulty_threshold(self.global_step)
history['difficulty_threshold'].append(current_threshold)
print(f"Epoch {epoch + 1}/{num_epochs}")
print(f" Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}")
print(f" Difficulty Threshold: {current_threshold:.4f}")
# Evaluate if requested
if eval_dataset is not None and (epoch + 1) % eval_interval == 0:
eval_loss, eval_acc = self.evaluate(eval_dataset, batch_size)
history['eval_loss'].append(eval_loss)
history['eval_accuracy'].append(eval_acc)
print(f" Eval Loss: {eval_loss:.4f}, Eval Acc: {eval_acc:.4f}")
print()
return history
def create_synthetic_dataset(num_examples: int, num_classes: int,
vocab_size: int = 1000) -> Tuple[List[str], List[int], Dict[str, int], Dict[str, float]]:
"""
Create a synthetic text classification dataset for demonstration.
Args:
num_examples: Number of examples to generate
num_classes: Number of classes
vocab_size: Size of vocabulary
Returns:
Tuple of (texts, labels, vocab, vocab_frequency)
"""
# Create vocabulary
vocab = {'<PAD>': 0, '<UNK>': 1}
for i in range(vocab_size):
vocab[f'word_{i}'] = i + 2
# Create word frequency distribution (Zipf-like)
vocab_frequency = {}
for word, idx in vocab.items():
if word not in ['<PAD>', '<UNK>']:
# Zipf distribution: frequency inversely proportional to rank
rank = idx
frequency = 1.0 / (rank ** 0.8)
vocab_frequency[word] = frequency
# Generate texts and labels
texts = []
labels = []
for i in range(num_examples):
# Label determines text characteristics
label = i % num_classes
# Easy examples: short, common words
# Hard examples: long, rare words
if i < num_examples * 0.3:
# Easy examples
length = random.randint(5, 15)
word_indices = random.choices(range(2, 100), k=length)
elif i < num_examples * 0.6:
# Medium examples
length = random.randint(15, 30)
word_indices = random.choices(range(2, 500), k=length)
else:
# Hard examples
length = random.randint(30, 60)
word_indices = random.choices(range(2, vocab_size + 2), k=length)
# Create text from word indices
words = [f'word_{idx - 2}' for idx in word_indices]
text = ' '.join(words)
texts.append(text)
labels.append(label)
return texts, labels, vocab, vocab_frequency
def main():
"""
Main function demonstrating curriculum learning usage.
"""
print("=" * 80)
print("CURRICULUM LEARNING DEMONSTRATION")
print("=" * 80)
print()
# Set random seeds for reproducibility
random.seed(42)
np.random.seed(42)
torch.manual_seed(42)
# Create synthetic dataset
print("Creating synthetic dataset...")
num_train = 1000
num_eval = 200
num_classes = 5
train_texts, train_labels, vocab, vocab_freq = create_synthetic_dataset(
num_train, num_classes
)
eval_texts, eval_labels, _, _ = create_synthetic_dataset(
num_eval, num_classes
)
print(f" Training examples: {num_train}")
print(f" Evaluation examples: {num_eval}")
print(f" Vocabulary size: {len(vocab)}")
print(f" Number of classes: {num_classes}")
print()
# Create datasets
train_dataset = TextDataset(train_texts, train_labels, vocab)
eval_dataset = TextDataset(eval_texts, eval_labels, vocab)
# Initialize model
print("Initializing model...")
device = 'cuda' if torch.cuda.is_available() else 'cpu'
print(f" Using device: {device}")
model = TextClassifier(
vocab_size=len(vocab),
embedding_dim=128,
hidden_dim=256,
num_classes=num_classes,
dropout=0.3
)
print()
# Demonstrate different curriculum learning configurations
# Configuration 1: Heuristic difficulty with fixed pacing
print("-" * 80)
print("CONFIGURATION 1: Heuristic Difficulty + Fixed Pacing")
print("-" * 80)
difficulty_scorer_1 = HeuristicDifficultyScorer(vocab_freq)
pacing_function_1 = FixedPacingFunction(
total_steps=500,
start_percentile=0.2,
end_percentile=1.0
)
trainer_1 = CurriculumLearningTrainer(
model=model,
dataset=train_dataset,
difficulty_scorer=difficulty_scorer_1,
pacing_function=pacing_function_1,
device=device,
scheduler_mode='soft'
)
print("Training with heuristic difficulty and fixed pacing...")
history_1 = trainer_1.train(
num_epochs=5,
batch_size=32,
learning_rate=0.001,
eval_dataset=eval_dataset,
eval_interval=1
)
print()
# Configuration 2: Adaptive difficulty with self-paced learning
print("-" * 80)
print("CONFIGURATION 2: Adaptive Difficulty + Self-Paced Learning")
print("-" * 80)
# Reinitialize model for fair comparison
model_2 = TextClassifier(
vocab_size=len(vocab),
embedding_dim=128,
hidden_dim=256,
num_classes=num_classes,
dropout=0.3
)
difficulty_scorer_2 = AdaptiveDifficultyScorer(
smoothing_factor=0.9,
initial_difficulty=0.5
)
pacing_function_2 = AdaptivePacingFunction(
initial_threshold=0.2,
growth_rate=0.01,
performance_window=100
)
trainer_2 = CurriculumLearningTrainer(
model=model_2,
dataset=train_dataset,
difficulty_scorer=difficulty_scorer_2,
pacing_function=pacing_function_2,
device=device,
scheduler_mode='soft'
)
print("Training with adaptive difficulty and self-paced learning...")
history_2 = trainer_2.train(
num_epochs=5,
batch_size=32,
learning_rate=0.001,
eval_dataset=eval_dataset,
eval_interval=1
)
print()
# Configuration 3: Baseline (no curriculum, for comparison)
print("-" * 80)
print("CONFIGURATION 3: Baseline (No Curriculum)")
print("-" * 80)
# Reinitialize model
model_3 = TextClassifier(
vocab_size=len(vocab),
embedding_dim=128,
hidden_dim=256,
num_classes=num_classes,
dropout=0.3
)
# Use fixed pacing that immediately uses all data
difficulty_scorer_3 = HeuristicDifficultyScorer(vocab_freq)
pacing_function_3 = FixedPacingFunction(
total_steps=1, # Immediately use all data
start_percentile=1.0,
end_percentile=1.0
)
trainer_3 = CurriculumLearningTrainer(
model=model_3,
dataset=train_dataset,
difficulty_scorer=difficulty_scorer_3,
pacing_function=pacing_function_3,
device=device,
scheduler_mode='hard'
)
print("Training without curriculum (baseline)...")
history_3 = trainer_3.train(
num_epochs=5,
batch_size=32,
learning_rate=0.001,
eval_dataset=eval_dataset,
eval_interval=1
)
print()
# Compare results
print("=" * 80)
print("COMPARISON OF RESULTS")
print("=" * 80)
print()
print("Final Training Accuracy:")
print(f" Config 1 (Heuristic + Fixed): {history_1['train_accuracy'][-1]:.4f}")
print(f" Config 2 (Adaptive + Self-Paced): {history_2['train_accuracy'][-1]:.4f}")
print(f" Config 3 (Baseline): {history_3['train_accuracy'][-1]:.4f}")
print()
print("Final Evaluation Accuracy:")
print(f" Config 1 (Heuristic + Fixed): {history_1['eval_accuracy'][-1]:.4f}")
print(f" Config 2 (Adaptive + Self-Paced): {history_2['eval_accuracy'][-1]:.4f}")
print(f" Config 3 (Baseline): {history_3['eval_accuracy'][-1]:.4f}")
print()
print("Training completed successfully!")
print("=" * 80)
if __name__ == "__main__":
main()
This complete implementation provides a production-ready curriculum learning system for text classification. The code includes multiple difficulty scoring strategies, both fixed and adaptive pacing functions, flexible scheduling modes, and a comprehensive training framework. The main function demonstrates three different configurations, allowing you to compare curriculum learning approaches against a baseline without curriculum learning.
The implementation follows clean code principles with clear separation of concerns. Each class has a single well-defined responsibility. The difficulty scorers are interchangeable, as are the pacing functions, allowing easy experimentation with different curriculum strategies. The code includes extensive documentation and handles edge cases properly.
You can adapt this implementation to other domains by replacing the TextDataset and TextClassifier with appropriate classes for your task. The curriculum learning components remain the same regardless of the specific machine learning problem you are solving.
No comments:
Post a Comment