Master transfer learning techniques using pretrained models, from feature extraction to fine-tuning strategies and domain adaptation.
Transfer learning is arguably the most impactful technique in modern computer vision. Instead of training models from scratch, we leverage knowledge learned from large datasets like ImageNet to solve new problems with limited data.
This lesson covers the spectrum of transfer learning approaches: using pretrained models as fixed feature extractors, fine-tuning strategies, domain adaptation for distribution shift, and knowledge distillation for model compression.
Using pretrained models as fixed feature extractors is the simplest form of transfer learning:
1import torch
2import torch.nn as nn
3import torchvision.models as models
4from torchvision import transforms
5from torch.utils.data import DataLoader
6import numpy as np
7
8class FeatureExtractor:
9 """
10 Extract features from pretrained models for downstream tasks.
11 """
12
13 def __init__(self, model_name: str = 'resnet50', device: str = 'cuda'):
14 self.device = device if torch.cuda.is_available() else 'cpu'
15
16 # Load pretrained model
17 self.model = self._load_model(model_name)
18 self.model.to(self.device)
19 self.model.eval()
20
21 # Freeze all parameters
22 for param in self.model.parameters():
23 param.requires_grad = False
24
25 # Standard ImageNet preprocessing
26 self.preprocess = transforms.Compose([
27 transforms.Resize(256),
28 transforms.CenterCrop(224),
29 transforms.ToTensor(),
30 transforms.Normalize(
31 mean=[0.485, 0.456, 0.406],
32 std=[0.229, 0.224, 0.225]
33 )
34 ])
35
36 def _load_model(self, model_name: str) -> nn.Module:
37 """Load pretrained model and remove classification head."""
38 if model_name == 'resnet50':
39 model = models.resnet50(pretrained=True)
40 # Remove the final FC layer
41 model = nn.Sequential(*list(model.children())[:-1])
42 self.feature_dim = 2048
43
44 elif model_name == 'efficientnet_b0':
45 model = models.efficientnet_b0(pretrained=True)
46 model = nn.Sequential(*list(model.children())[:-1])
47 self.feature_dim = 1280
48
49 elif model_name == 'vit_b_16':
50 model = models.vit_b_16(pretrained=True)
51 # ViT returns class token, we need to extract it differently
52 self.feature_dim = 768
53 return model
54
55 else:
56 raise ValueError(f"Unknown model: {model_name}")
57
58 return model
59
60 @torch.no_grad()
61 def extract_features(self, images: torch.Tensor) -> np.ndarray:
62 """
63 Extract features from a batch of images.
64
65 Args:
66 images: Preprocessed images [B, C, H, W]
67
68 Returns:
69 Feature vectors [B, feature_dim]
70 """
71 images = images.to(self.device)
72 features = self.model(images)
73
74 # Flatten if needed (e.g., from [B, 2048, 1, 1] to [B, 2048])
75 features = features.squeeze()
76
77 return features.cpu().numpy()
78
79 def extract_from_dataloader(self, dataloader: DataLoader) -> tuple:
80 """
81 Extract features from entire dataset.
82
83 Returns:
84 features: [N, feature_dim]
85 labels: [N]
86 """
87 all_features = []
88 all_labels = []
89
90 for images, labels in dataloader:
91 features = self.extract_features(images)
92 all_features.append(features)
93 all_labels.extend(labels.numpy())
94
95 return np.vstack(all_features), np.array(all_labels)
96
97
98class LinearClassifier(nn.Module):
99 """
100 Simple linear classifier on top of extracted features.
101 """
102
103 def __init__(self, feature_dim: int, num_classes: int):
104 super().__init__()
105 self.classifier = nn.Sequential(
106 nn.Linear(feature_dim, 512),
107 nn.ReLU(),
108 nn.Dropout(0.5),
109 nn.Linear(512, num_classes)
110 )
111
112 def forward(self, x: torch.Tensor) -> torch.Tensor:
113 return self.classifier(x)
114
115
116# Example: Using features with sklearn
117def train_with_sklearn(features: np.ndarray, labels: np.ndarray):
118 """Train simple classifiers on extracted features."""
119 from sklearn.linear_model import LogisticRegression
120 from sklearn.svm import SVC
121 from sklearn.model_selection import train_test_split
122 from sklearn.metrics import accuracy_score
123
124 X_train, X_test, y_train, y_test = train_test_split(
125 features, labels, test_size=0.2, random_state=42
126 )
127
128 # Logistic Regression
129 lr = LogisticRegression(max_iter=1000)
130 lr.fit(X_train, y_train)
131 lr_acc = accuracy_score(y_test, lr.predict(X_test))
132
133 # SVM
134 svm = SVC(kernel='rbf')
135 svm.fit(X_train, y_train)
136 svm_acc = accuracy_score(y_test, svm.predict(X_test))
137
138 return {'logistic_regression': lr_acc, 'svm': svm_acc}
139
140
141# Usage example
142extractor = FeatureExtractor(model_name='resnet50')
143print(f"Feature dimension: {extractor.feature_dim}")
144print(f"Model on device: {extractor.device}")
145
146# Demo with random data
147demo_images = torch.randn(4, 3, 224, 224)
148features = extractor.extract_features(demo_images)
149print(f"Extracted features shape: {features.shape}")Fine-tuning adapts pretrained weights to your specific task:
1import torch
2import torch.nn as nn
3import torchvision.models as models
4from typing import List, Optional
5import math
6
7class FineTuner:
8 """
9 Comprehensive fine-tuning strategies for pretrained models.
10 """
11
12 def __init__(self, model_name: str = 'resnet50', num_classes: int = 10,
13 device: str = 'cuda'):
14 self.device = device if torch.cuda.is_available() else 'cpu'
15 self.num_classes = num_classes
16
17 # Load base model
18 self.model = self._create_model(model_name)
19 self.model.to(self.device)
20
21 def _create_model(self, model_name: str) -> nn.Module:
22 """Create model with custom classification head."""
23 if model_name == 'resnet50':
24 model = models.resnet50(pretrained=True)
25 num_features = model.fc.in_features
26 model.fc = nn.Linear(num_features, self.num_classes)
27
28 elif model_name == 'efficientnet_b0':
29 model = models.efficientnet_b0(pretrained=True)
30 num_features = model.classifier[1].in_features
31 model.classifier = nn.Sequential(
32 nn.Dropout(0.2),
33 nn.Linear(num_features, self.num_classes)
34 )
35
36 return model
37
38 def freeze_layers(self, num_layers_to_freeze: int = None,
39 freeze_bn: bool = True):
40 """
41 Freeze early layers of the network.
42
43 Args:
44 num_layers_to_freeze: Number of layers to freeze (from start)
45 freeze_bn: Whether to freeze batch normalization layers
46 """
47 if num_layers_to_freeze is None:
48 # Default: freeze all except final layer
49 for name, param in self.model.named_parameters():
50 if 'fc' not in name and 'classifier' not in name:
51 param.requires_grad = False
52 else:
53 # Freeze specific number of layers
54 children = list(self.model.children())
55 for i, child in enumerate(children[:num_layers_to_freeze]):
56 for param in child.parameters():
57 param.requires_grad = False
58
59 # Optionally freeze batch norm statistics
60 if freeze_bn:
61 for module in self.model.modules():
62 if isinstance(module, nn.BatchNorm2d):
63 module.eval()
64
65 def gradual_unfreeze(self, epoch: int, unfreeze_schedule: List[int]):
66 """
67 Gradually unfreeze layers during training.
68
69 Args:
70 epoch: Current training epoch
71 unfreeze_schedule: List of epochs at which to unfreeze next block
72 """
73 children = list(self.model.children())
74 num_blocks = len(children)
75
76 # Determine how many blocks to unfreeze
77 blocks_to_unfreeze = sum(1 for e in unfreeze_schedule if epoch >= e)
78
79 # Unfreeze from the end (closest to output)
80 for i, child in enumerate(reversed(children)):
81 if i < blocks_to_unfreeze:
82 for param in child.parameters():
83 param.requires_grad = True
84
85 def get_layer_groups(self) -> List[List[nn.Parameter]]:
86 """
87 Get parameter groups for discriminative learning rates.
88
89 Returns groups from earliest to latest layers.
90 """
91 # For ResNet: group by residual blocks
92 groups = []
93
94 # Early layers (stem)
95 stem_params = []
96 for name, param in self.model.named_parameters():
97 if 'layer1' not in name and 'layer2' not in name and 'layer3' not in name and 'layer4' not in name and 'fc' not in name:
98 stem_params.append(param)
99 groups.append(stem_params)
100
101 # Residual blocks
102 for layer_name in ['layer1', 'layer2', 'layer3', 'layer4']:
103 layer_params = []
104 for name, param in self.model.named_parameters():
105 if layer_name in name:
106 layer_params.append(param)
107 if layer_params:
108 groups.append(layer_params)
109
110 # Classification head
111 head_params = []
112 for name, param in self.model.named_parameters():
113 if 'fc' in name or 'classifier' in name:
114 head_params.append(param)
115 groups.append(head_params)
116
117 return groups
118
119 def get_discriminative_lr_optimizer(self, base_lr: float = 1e-3,
120 lr_decay: float = 0.1):
121 """
122 Create optimizer with discriminative learning rates.
123
124 Earlier layers get smaller learning rates.
125 """
126 groups = self.get_layer_groups()
127
128 param_groups = []
129 for i, group_params in enumerate(groups):
130 # Earlier groups get smaller LR
131 lr = base_lr * (lr_decay ** (len(groups) - 1 - i))
132 param_groups.append({
133 'params': group_params,
134 'lr': lr
135 })
136
137 return torch.optim.Adam(param_groups)
138
139
140class WarmupCosineScheduler:
141 """
142 Learning rate scheduler with warmup and cosine annealing.
143 Recommended for fine-tuning.
144 """
145
146 def __init__(self, optimizer, warmup_epochs: int, total_epochs: int,
147 warmup_lr: float = 1e-6, min_lr: float = 1e-6):
148 self.optimizer = optimizer
149 self.warmup_epochs = warmup_epochs
150 self.total_epochs = total_epochs
151 self.warmup_lr = warmup_lr
152 self.min_lr = min_lr
153
154 # Store base learning rates
155 self.base_lrs = [group['lr'] for group in optimizer.param_groups]
156
157 def step(self, epoch: int):
158 """Update learning rate based on epoch."""
159 if epoch < self.warmup_epochs:
160 # Linear warmup
161 alpha = epoch / self.warmup_epochs
162 for i, group in enumerate(self.optimizer.param_groups):
163 group['lr'] = self.warmup_lr + alpha * (self.base_lrs[i] - self.warmup_lr)
164 else:
165 # Cosine annealing
166 progress = (epoch - self.warmup_epochs) / (self.total_epochs - self.warmup_epochs)
167 for i, group in enumerate(self.optimizer.param_groups):
168 group['lr'] = self.min_lr + 0.5 * (self.base_lrs[i] - self.min_lr) * (1 + math.cos(math.pi * progress))
169
170
171# Example usage
172finetuner = FineTuner(model_name='resnet50', num_classes=10)
173print(f"Total parameters: {sum(p.numel() for p in finetuner.model.parameters()):,}")
174
175# Freeze all but last layer
176finetuner.freeze_layers()
177trainable = sum(p.numel() for p in finetuner.model.parameters() if p.requires_grad)
178print(f"Trainable parameters after freezing: {trainable:,}")
179
180# Get discriminative LR optimizer
181optimizer = finetuner.get_discriminative_lr_optimizer(base_lr=1e-3, lr_decay=0.1)
182print(f"\nLearning rates per group:")
183for i, group in enumerate(optimizer.param_groups):
184 print(f" Group {i}: {group['lr']:.6f}")Domain adaptation addresses distribution shift between training and target domains:
1import torch
2import torch.nn as nn
3import torch.nn.functional as F
4from typing import Tuple
5
6class DomainAdversarialNetwork(nn.Module):
7 """
8 Domain-Adversarial Neural Network (DANN).
9
10 Uses gradient reversal to learn domain-invariant features.
11 """
12
13 def __init__(self, feature_extractor: nn.Module, feature_dim: int,
14 num_classes: int):
15 super().__init__()
16
17 self.feature_extractor = feature_extractor
18
19 # Task classifier (label predictor)
20 self.classifier = nn.Sequential(
21 nn.Linear(feature_dim, 256),
22 nn.ReLU(),
23 nn.Dropout(0.5),
24 nn.Linear(256, num_classes)
25 )
26
27 # Domain discriminator
28 self.domain_discriminator = nn.Sequential(
29 nn.Linear(feature_dim, 256),
30 nn.ReLU(),
31 nn.Dropout(0.5),
32 nn.Linear(256, 2) # source vs target
33 )
34
35 def forward(self, x: torch.Tensor, alpha: float = 1.0
36 ) -> Tuple[torch.Tensor, torch.Tensor]:
37 """
38 Forward pass with gradient reversal for domain discriminator.
39
40 Args:
41 x: Input images
42 alpha: Gradient reversal strength
43
44 Returns:
45 class_output: Class predictions
46 domain_output: Domain predictions
47 """
48 features = self.feature_extractor(x)
49 features = features.view(features.size(0), -1)
50
51 # Class prediction (normal gradient flow)
52 class_output = self.classifier(features)
53
54 # Domain prediction (reversed gradient)
55 reversed_features = GradientReversal.apply(features, alpha)
56 domain_output = self.domain_discriminator(reversed_features)
57
58 return class_output, domain_output
59
60
61class GradientReversal(torch.autograd.Function):
62 """
63 Gradient Reversal Layer for domain adversarial training.
64 """
65
66 @staticmethod
67 def forward(ctx, x: torch.Tensor, alpha: float) -> torch.Tensor:
68 ctx.alpha = alpha
69 return x.view_as(x)
70
71 @staticmethod
72 def backward(ctx, grad_output: torch.Tensor) -> Tuple[torch.Tensor, None]:
73 return -ctx.alpha * grad_output, None
74
75
76class MMDLoss(nn.Module):
77 """
78 Maximum Mean Discrepancy loss for domain adaptation.
79
80 Measures distance between source and target distributions.
81 """
82
83 def __init__(self, kernel_type: str = 'rbf', kernel_mul: float = 2.0,
84 kernel_num: int = 5):
85 super().__init__()
86 self.kernel_type = kernel_type
87 self.kernel_mul = kernel_mul
88 self.kernel_num = kernel_num
89
90 def gaussian_kernel(self, source: torch.Tensor, target: torch.Tensor,
91 kernel_mul: float, kernel_num: int,
92 fix_sigma: float = None) -> torch.Tensor:
93 """Compute Gaussian kernel matrix."""
94 n_samples = source.size(0) + target.size(0)
95 total = torch.cat([source, target], dim=0)
96
97 # Compute pairwise distances
98 total0 = total.unsqueeze(0).expand(
99 total.size(0), total.size(0), total.size(1)
100 )
101 total1 = total.unsqueeze(1).expand(
102 total.size(0), total.size(0), total.size(1)
103 )
104 L2_distance = ((total0 - total1) ** 2).sum(2)
105
106 # Bandwidth selection
107 if fix_sigma:
108 bandwidth = fix_sigma
109 else:
110 bandwidth = torch.sum(L2_distance) / (n_samples ** 2 - n_samples)
111
112 bandwidth /= kernel_mul ** (kernel_num // 2)
113 bandwidth_list = [bandwidth * (kernel_mul ** i) for i in range(kernel_num)]
114
115 # Sum of Gaussian kernels
116 kernel_val = [torch.exp(-L2_distance / bw) for bw in bandwidth_list]
117 return sum(kernel_val)
118
119 def forward(self, source: torch.Tensor,
120 target: torch.Tensor) -> torch.Tensor:
121 """
122 Compute MMD loss between source and target features.
123 """
124 batch_size = source.size(0)
125 kernels = self.gaussian_kernel(
126 source, target, self.kernel_mul, self.kernel_num
127 )
128
129 # Split kernel matrix
130 XX = kernels[:batch_size, :batch_size]
131 YY = kernels[batch_size:, batch_size:]
132 XY = kernels[:batch_size, batch_size:]
133
134 # MMD^2 = E[k(X,X')] + E[k(Y,Y')] - 2*E[k(X,Y)]
135 mmd = XX.mean() + YY.mean() - 2 * XY.mean()
136 return mmd
137
138
139class DomainAdaptationTrainer:
140 """Training pipeline for domain adaptation."""
141
142 def __init__(self, model: DomainAdversarialNetwork, device: str = 'cuda'):
143 self.model = model.to(device)
144 self.device = device
145
146 self.optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
147 self.class_criterion = nn.CrossEntropyLoss()
148 self.domain_criterion = nn.CrossEntropyLoss()
149
150 def train_step(self, source_data: Tuple[torch.Tensor, torch.Tensor],
151 target_data: torch.Tensor, epoch: int, max_epochs: int):
152 """
153 Single training step with domain adversarial loss.
154 """
155 source_imgs, source_labels = source_data
156 target_imgs = target_data
157
158 source_imgs = source_imgs.to(self.device)
159 source_labels = source_labels.to(self.device)
160 target_imgs = target_imgs.to(self.device)
161
162 # Adaptation parameter (schedule from 0 to 1)
163 p = epoch / max_epochs
164 alpha = 2.0 / (1.0 + math.exp(-10 * p)) - 1
165
166 self.optimizer.zero_grad()
167
168 # Source forward pass
169 class_output, domain_output_source = self.model(source_imgs, alpha)
170
171 # Target forward pass
172 _, domain_output_target = self.model(target_imgs, alpha)
173
174 # Classification loss (only on source)
175 class_loss = self.class_criterion(class_output, source_labels)
176
177 # Domain loss
178 source_domain_labels = torch.zeros(source_imgs.size(0),
179 dtype=torch.long, device=self.device)
180 target_domain_labels = torch.ones(target_imgs.size(0),
181 dtype=torch.long, device=self.device)
182
183 domain_loss = self.domain_criterion(domain_output_source, source_domain_labels) + self.domain_criterion(domain_output_target, target_domain_labels)
184
185 # Total loss
186 loss = class_loss + domain_loss
187 loss.backward()
188 self.optimizer.step()
189
190 return {
191 'total_loss': loss.item(),
192 'class_loss': class_loss.item(),
193 'domain_loss': domain_loss.item(),
194 'alpha': alpha
195 }
196
197
198# Example
199import torchvision.models as models
200
201# Create feature extractor from pretrained ResNet
202backbone = models.resnet18(pretrained=True)
203feature_extractor = nn.Sequential(*list(backbone.children())[:-1])
204
205dann = DomainAdversarialNetwork(
206 feature_extractor=feature_extractor,
207 feature_dim=512,
208 num_classes=10
209)
210
211print(f"DANN parameters: {sum(p.numel() for p in dann.parameters()):,}")Knowledge distillation transfers knowledge from large models to smaller ones:
1import torch
2import torch.nn as nn
3import torch.nn.functional as F
4from typing import Optional
5
6class KnowledgeDistillation:
7 """
8 Knowledge Distillation framework.
9
10 Transfer knowledge from a large teacher model to a smaller student.
11 """
12
13 def __init__(self, teacher: nn.Module, student: nn.Module,
14 temperature: float = 4.0, alpha: float = 0.7,
15 device: str = 'cuda'):
16 """
17 Args:
18 teacher: Pretrained teacher model
19 student: Student model to train
20 temperature: Softmax temperature (higher = softer distributions)
21 alpha: Weight for distillation loss (1-alpha for hard labels)
22 """
23 self.device = device if torch.cuda.is_available() else 'cpu'
24
25 self.teacher = teacher.to(self.device)
26 self.student = student.to(self.device)
27
28 # Freeze teacher
29 self.teacher.eval()
30 for param in self.teacher.parameters():
31 param.requires_grad = False
32
33 self.temperature = temperature
34 self.alpha = alpha
35
36 self.optimizer = torch.optim.Adam(self.student.parameters(), lr=1e-3)
37 self.hard_loss_fn = nn.CrossEntropyLoss()
38
39 def distillation_loss(self, student_logits: torch.Tensor,
40 teacher_logits: torch.Tensor,
41 labels: torch.Tensor) -> torch.Tensor:
42 """
43 Compute combined distillation and hard label loss.
44
45 Distillation loss: KL divergence between soft teacher and student outputs
46 Hard loss: Cross entropy with ground truth labels
47 """
48 T = self.temperature
49
50 # Soft targets from teacher
51 soft_teacher = F.softmax(teacher_logits / T, dim=1)
52
53 # Soft predictions from student
54 soft_student = F.log_softmax(student_logits / T, dim=1)
55
56 # KL divergence loss (scaled by T^2 as per original paper)
57 distill_loss = F.kl_div(soft_student, soft_teacher,
58 reduction='batchmean') * (T ** 2)
59
60 # Hard label loss
61 hard_loss = self.hard_loss_fn(student_logits, labels)
62
63 # Combined loss
64 loss = self.alpha * distill_loss + (1 - self.alpha) * hard_loss
65
66 return loss, distill_loss, hard_loss
67
68 def train_step(self, images: torch.Tensor,
69 labels: torch.Tensor) -> dict:
70 """Single training step."""
71 images = images.to(self.device)
72 labels = labels.to(self.device)
73
74 self.optimizer.zero_grad()
75
76 # Get teacher predictions (no gradient)
77 with torch.no_grad():
78 teacher_logits = self.teacher(images)
79
80 # Get student predictions
81 student_logits = self.student(images)
82
83 # Compute loss
84 loss, distill_loss, hard_loss = self.distillation_loss(
85 student_logits, teacher_logits, labels
86 )
87
88 loss.backward()
89 self.optimizer.step()
90
91 # Calculate accuracy
92 _, predicted = student_logits.max(1)
93 accuracy = (predicted == labels).float().mean()
94
95 return {
96 'total_loss': loss.item(),
97 'distill_loss': distill_loss.item(),
98 'hard_loss': hard_loss.item(),
99 'accuracy': accuracy.item()
100 }
101
102
103class FeatureDistillation(nn.Module):
104 """
105 Feature-level knowledge distillation.
106
107 Matches intermediate feature representations, not just outputs.
108 """
109
110 def __init__(self, teacher: nn.Module, student: nn.Module,
111 hint_layers: list):
112 """
113 Args:
114 teacher: Teacher model
115 student: Student model
116 hint_layers: Names of layers to match features
117 """
118 super().__init__()
119 self.teacher = teacher
120 self.student = student
121 self.hint_layers = hint_layers
122
123 # Hooks to capture intermediate features
124 self.teacher_features = {}
125 self.student_features = {}
126
127 self._register_hooks()
128
129 # Adaptation layers to match dimensions
130 self.adaptation = nn.ModuleDict()
131
132 def _register_hooks(self):
133 """Register forward hooks to capture features."""
134 def get_teacher_hook(name):
135 def hook(module, input, output):
136 self.teacher_features[name] = output
137 return hook
138
139 def get_student_hook(name):
140 def hook(module, input, output):
141 self.student_features[name] = output
142 return hook
143
144 for name in self.hint_layers:
145 # Register hooks for both models
146 for n, m in self.teacher.named_modules():
147 if n == name:
148 m.register_forward_hook(get_teacher_hook(name))
149
150 for n, m in self.student.named_modules():
151 if n == name:
152 m.register_forward_hook(get_student_hook(name))
153
154 def feature_loss(self) -> torch.Tensor:
155 """Compute feature matching loss."""
156 total_loss = 0
157
158 for name in self.hint_layers:
159 teacher_feat = self.teacher_features.get(name)
160 student_feat = self.student_features.get(name)
161
162 if teacher_feat is not None and student_feat is not None:
163 # Adapt student features if dimensions don't match
164 if teacher_feat.shape != student_feat.shape:
165 if name not in self.adaptation:
166 # Create adaptation layer
167 self.adaptation[name] = nn.Conv2d(
168 student_feat.size(1),
169 teacher_feat.size(1),
170 kernel_size=1
171 ).to(student_feat.device)
172 student_feat = self.adaptation[name](student_feat)
173
174 # L2 loss between features
175 total_loss += F.mse_loss(student_feat, teacher_feat)
176
177 return total_loss
178
179
180class AttentionTransfer(nn.Module):
181 """
182 Attention Transfer for knowledge distillation.
183
184 Transfers attention maps (spatial importance) from teacher to student.
185 """
186
187 @staticmethod
188 def attention_map(features: torch.Tensor) -> torch.Tensor:
189 """
190 Compute attention map from feature maps.
191
192 Attention = sum over channels of squared activations.
193 """
194 # features: [B, C, H, W]
195 return torch.sum(features ** 2, dim=1) # [B, H, W]
196
197 @staticmethod
198 def attention_loss(student_features: torch.Tensor,
199 teacher_features: torch.Tensor) -> torch.Tensor:
200 """Compute attention transfer loss."""
201 # Get attention maps
202 student_att = AttentionTransfer.attention_map(student_features)
203 teacher_att = AttentionTransfer.attention_map(teacher_features)
204
205 # Normalize attention maps
206 student_att = student_att / (student_att.sum(dim=(1, 2), keepdim=True) + 1e-8)
207 teacher_att = teacher_att / (teacher_att.sum(dim=(1, 2), keepdim=True) + 1e-8)
208
209 # Resize if needed
210 if student_att.shape != teacher_att.shape:
211 student_att = F.interpolate(
212 student_att.unsqueeze(1),
213 size=teacher_att.shape[1:],
214 mode='bilinear',
215 align_corners=False
216 ).squeeze(1)
217
218 # L2 loss between attention maps
219 return F.mse_loss(student_att, teacher_att)
220
221
222# Example: Create teacher and student models
223import torchvision.models as models
224
225# Large teacher
226teacher = models.resnet50(pretrained=True)
227
228# Smaller student
229student = models.resnet18(pretrained=False)
230
231# Initialize distillation
232distiller = KnowledgeDistillation(
233 teacher=teacher,
234 student=student,
235 temperature=4.0,
236 alpha=0.7
237)
238
239# Demo training step
240images = torch.randn(8, 3, 224, 224)
241labels = torch.randint(0, 1000, (8,))
242
243metrics = distiller.train_step(images, labels)
244print("Distillation metrics:")
245for k, v in metrics.items():
246 print(f" {k}: {v:.4f}")Build a production-ready transfer learning system with progressive unfreezing and monitoring.
1import torch
2import torch.nn as nn
3import torchvision.models as models
4from torchvision import transforms, datasets
5from torch.utils.data import DataLoader, random_split
6import numpy as np
7from typing import Dict, List, Tuple
8import copy
9
10class TransferLearningPipeline:
11 """
12 Production-ready transfer learning pipeline.
13
14 Features:
15 - Multiple pretrained backbone options
16 - Progressive unfreezing strategy
17 - Discriminative learning rates
18 - Early stopping
19 - Model checkpointing
20 """
21
22 def __init__(self, backbone: str = 'resnet50', num_classes: int = 10,
23 device: str = None):
24 self.device = device or ('cuda' if torch.cuda.is_available() else 'cpu')
25 self.num_classes = num_classes
26 self.backbone_name = backbone
27
28 # Create model
29 self.model = self._create_model(backbone)
30 self.model.to(self.device)
31
32 # Training state
33 self.best_model_state = None
34 self.best_accuracy = 0
35 self.history = {'train_loss': [], 'val_loss': [],
36 'train_acc': [], 'val_acc': []}
37
38 def _create_model(self, backbone: str) -> nn.Module:
39 """Create model with pretrained backbone."""
40 if backbone == 'resnet50':
41 model = models.resnet50(weights=models.ResNet50_Weights.DEFAULT)
42 num_features = model.fc.in_features
43 model.fc = nn.Sequential(
44 nn.Dropout(0.5),
45 nn.Linear(num_features, self.num_classes)
46 )
47
48 elif backbone == 'efficientnet_b0':
49 model = models.efficientnet_b0(weights=models.EfficientNet_B0_Weights.DEFAULT)
50 num_features = model.classifier[1].in_features
51 model.classifier = nn.Sequential(
52 nn.Dropout(0.2),
53 nn.Linear(num_features, self.num_classes)
54 )
55
56 elif backbone == 'vit_b_16':
57 model = models.vit_b_16(weights=models.ViT_B_16_Weights.DEFAULT)
58 num_features = model.heads.head.in_features
59 model.heads.head = nn.Linear(num_features, self.num_classes)
60
61 else:
62 raise ValueError(f"Unknown backbone: {backbone}")
63
64 return model
65
66 def freeze_backbone(self):
67 """Freeze all backbone layers, keep head trainable."""
68 for name, param in self.model.named_parameters():
69 if 'fc' not in name and 'classifier' not in name and 'heads' not in name:
70 param.requires_grad = False
71
72 def unfreeze_all(self):
73 """Unfreeze all layers."""
74 for param in self.model.parameters():
75 param.requires_grad = True
76
77 def get_param_groups(self, base_lr: float) -> List[Dict]:
78 """
79 Get parameter groups with discriminative learning rates.
80
81 Head gets base_lr, backbone gets base_lr / 10.
82 """
83 head_params = []
84 backbone_params = []
85
86 for name, param in self.model.named_parameters():
87 if 'fc' in name or 'classifier' in name or 'heads' in name:
88 head_params.append(param)
89 else:
90 backbone_params.append(param)
91
92 return [
93 {'params': backbone_params, 'lr': base_lr / 10},
94 {'params': head_params, 'lr': base_lr}
95 ]
96
97 def train_epoch(self, dataloader: DataLoader, optimizer: torch.optim.Optimizer,
98 criterion: nn.Module) -> Tuple[float, float]:
99 """Train for one epoch."""
100 self.model.train()
101 total_loss = 0
102 correct = 0
103 total = 0
104
105 for images, labels in dataloader:
106 images = images.to(self.device)
107 labels = labels.to(self.device)
108
109 optimizer.zero_grad()
110 outputs = self.model(images)
111 loss = criterion(outputs, labels)
112 loss.backward()
113 optimizer.step()
114
115 total_loss += loss.item() * images.size(0)
116 _, predicted = outputs.max(1)
117 correct += (predicted == labels).sum().item()
118 total += labels.size(0)
119
120 return total_loss / total, correct / total
121
122 @torch.no_grad()
123 def validate(self, dataloader: DataLoader,
124 criterion: nn.Module) -> Tuple[float, float]:
125 """Validate model."""
126 self.model.eval()
127 total_loss = 0
128 correct = 0
129 total = 0
130
131 for images, labels in dataloader:
132 images = images.to(self.device)
133 labels = labels.to(self.device)
134
135 outputs = self.model(images)
136 loss = criterion(outputs, labels)
137
138 total_loss += loss.item() * images.size(0)
139 _, predicted = outputs.max(1)
140 correct += (predicted == labels).sum().item()
141 total += labels.size(0)
142
143 return total_loss / total, correct / total
144
145 def train(self, train_loader: DataLoader, val_loader: DataLoader,
146 epochs: int = 20, lr: float = 1e-3,
147 patience: int = 5, progressive_unfreeze: bool = True):
148 """
149 Full training loop with optional progressive unfreezing.
150
151 Phase 1: Train only the head (frozen backbone)
152 Phase 2: Fine-tune entire network with discriminative LR
153 """
154 criterion = nn.CrossEntropyLoss()
155 early_stop_counter = 0
156
157 print(f"Training on {self.device}")
158 print(f"Backbone: {self.backbone_name}")
159 print(f"Parameters: {sum(p.numel() for p in self.model.parameters()):,}")
160
161 if progressive_unfreeze:
162 # Phase 1: Frozen backbone
163 print("\n=== Phase 1: Training head only ===")
164 self.freeze_backbone()
165
166 trainable = sum(p.numel() for p in self.model.parameters() if p.requires_grad)
167 print(f"Trainable parameters: {trainable:,}")
168
169 optimizer = torch.optim.Adam(
170 filter(lambda p: p.requires_grad, self.model.parameters()),
171 lr=lr
172 )
173
174 phase1_epochs = min(epochs // 3, 5)
175 for epoch in range(phase1_epochs):
176 train_loss, train_acc = self.train_epoch(train_loader, optimizer, criterion)
177 val_loss, val_acc = self.validate(val_loader, criterion)
178
179 self._update_history(train_loss, val_loss, train_acc, val_acc)
180 self._check_best(val_acc)
181
182 print(f"Epoch {epoch+1}/{phase1_epochs} | "
183 f"Train: {train_loss:.4f}/{train_acc:.3f} | "
184 f"Val: {val_loss:.4f}/{val_acc:.3f}")
185
186 # Phase 2: Full fine-tuning
187 print("\n=== Phase 2: Fine-tuning all layers ===")
188 self.unfreeze_all()
189 remaining_epochs = epochs - phase1_epochs
190
191 else:
192 self.unfreeze_all()
193 remaining_epochs = epochs
194
195 # Get discriminative LR optimizer
196 param_groups = self.get_param_groups(lr if progressive_unfreeze else lr / 10)
197 optimizer = torch.optim.Adam(param_groups)
198
199 scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
200 optimizer, mode='max', factor=0.5, patience=2, verbose=True
201 )
202
203 for epoch in range(remaining_epochs):
204 train_loss, train_acc = self.train_epoch(train_loader, optimizer, criterion)
205 val_loss, val_acc = self.validate(val_loader, criterion)
206
207 self._update_history(train_loss, val_loss, train_acc, val_acc)
208
209 # Check for improvement
210 if val_acc > self.best_accuracy:
211 self.best_accuracy = val_acc
212 self.best_model_state = copy.deepcopy(self.model.state_dict())
213 early_stop_counter = 0
214 else:
215 early_stop_counter += 1
216
217 scheduler.step(val_acc)
218
219 current_epoch = epoch + (epochs - remaining_epochs) + 1
220 print(f"Epoch {current_epoch}/{epochs} | "
221 f"Train: {train_loss:.4f}/{train_acc:.3f} | "
222 f"Val: {val_loss:.4f}/{val_acc:.3f} | "
223 f"Best: {self.best_accuracy:.3f}")
224
225 # Early stopping
226 if early_stop_counter >= patience:
227 print(f"Early stopping after {current_epoch} epochs")
228 break
229
230 # Load best model
231 if self.best_model_state:
232 self.model.load_state_dict(self.best_model_state)
233 print(f"\nLoaded best model with accuracy: {self.best_accuracy:.3f}")
234
235 def _update_history(self, train_loss, val_loss, train_acc, val_acc):
236 self.history['train_loss'].append(train_loss)
237 self.history['val_loss'].append(val_loss)
238 self.history['train_acc'].append(train_acc)
239 self.history['val_acc'].append(val_acc)
240
241 def _check_best(self, val_acc):
242 if val_acc > self.best_accuracy:
243 self.best_accuracy = val_acc
244 self.best_model_state = copy.deepcopy(self.model.state_dict())
245
246 @torch.no_grad()
247 def predict(self, images: torch.Tensor) -> torch.Tensor:
248 """Make predictions."""
249 self.model.eval()
250 images = images.to(self.device)
251 outputs = self.model(images)
252 _, predicted = outputs.max(1)
253 return predicted.cpu()
254
255 def save(self, path: str):
256 """Save model checkpoint."""
257 torch.save({
258 'model_state': self.model.state_dict(),
259 'backbone': self.backbone_name,
260 'num_classes': self.num_classes,
261 'best_accuracy': self.best_accuracy,
262 'history': self.history
263 }, path)
264 print(f"Model saved to {path}")
265
266 def load(self, path: str):
267 """Load model checkpoint."""
268 checkpoint = torch.load(path, map_location=self.device)
269 self.model.load_state_dict(checkpoint['model_state'])
270 self.best_accuracy = checkpoint['best_accuracy']
271 self.history = checkpoint['history']
272 print(f"Model loaded from {path}")
273
274
275def main():
276 # Initialize pipeline
277 pipeline = TransferLearningPipeline(
278 backbone='resnet50',
279 num_classes=10
280 )
281
282 print(f"Model created with {pipeline.backbone_name} backbone")
283 print(f"Total parameters: {sum(p.numel() for p in pipeline.model.parameters()):,}")
284
285 # Create dummy data for demonstration
286 transform = transforms.Compose([
287 transforms.Resize(256),
288 transforms.CenterCrop(224),
289 transforms.ToTensor(),
290 transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
291 ])
292
293 # In practice, use real dataset:
294 # dataset = datasets.ImageFolder('path/to/data', transform=transform)
295
296 print("\nPipeline ready for training!")
297 print("Call pipeline.train(train_loader, val_loader, epochs=20)")
298
299if __name__ == "__main__":
300 main()This production-ready pipeline demonstrates transfer learning best practices including progressive unfreezing (training head first, then full model), discriminative learning rates (lower LR for pretrained layers), early stopping, and model checkpointing.
Next lesson covers Video Analysis including optical flow, object tracking with SORT and DeepSORT, and action recognition techniques.
Your progress is saved in this browser, on this device only.