Synthetic Data: When Artificial Intelligence Builds Its Own Training Data
Imagine trying to train an artificial intelligence model to detect a rare disease early, but there are only a few hundred real cases inβ¦
| Catastrophe Type | Main Symptom | Severity | Vulnerable Architectures |
|---|---|---|---|
| Mode Collapse | Identical repetitive outputs | π΄ Critical | GANs |
| Catastrophic Forgetting | Forgetting previous knowledge | π΄ Critical | All (especially Continual Learning) |
| Gradient Explosion | Loss β NaN or Inf | π΄ Critical | RNNs, Deep networks |
| Gradient Vanishing | Early layers don't learn | π‘ Medium | Very deep networks, RNNs |
| Training Instability | Severe loss oscillations | π High | GANs, Large Transformers |
| Dead Neurons | Part of network inactive | π‘ Medium | Networks with ReLU |
| Oscillation/Divergence | Non-convergence | π High | High LR, inappropriate architecture |
min_G max_D V(D, G) = E[log D(x)] + E[log(1 - D(G(z)))]from torch.nn.utils import spectral_normclass Generator(nn.Module):def __init__(self):# Instead of regular Conv2dself.conv1 = spectral_norm(nn.Conv2d(128, 256, 3))self.conv2 = spectral_norm(nn.Conv2d(256, 512, 3))
class MinibatchDiscrimination(nn.Module):def forward(self, x):# Calculate similarity between samples in batchdistances = compute_pairwise_distances(x)# If all very similar β probably fakediversity_score = distances.mean()return torch.cat([x, diversity_score], dim=1)
# Epoch 1-10: 4x4 images# Epoch 11-20: 8x8 images# Epoch 21-30: 16x16 images# ...# Epoch 61-70: 1024x1024 images
Network weights: WTask 1: W moves toward Task 1 optimum β WβTask 2: Wβ moves toward Task 2 optimum β WβBut: Wβ might be very bad for Task 1!
Loss = Loss_task_new + Ξ» Ξ£ F_i (ΞΈ_i - ΞΈ*_i)Β²F_i = Fisher Information Matrix (importance of weight i for previous task)ΞΈ*_i = optimal weight for previous taskΞ» = protection amount (usually 1000-10000)
class EWC:def __init__(self, model, dataloader, lambda_=1000):self.model = modelself.lambda_ = lambda_self.fisher = {}self.optimal_params = {}# Compute Fisher Informationself._compute_fisher(dataloader)def _compute_fisher(self, dataloader):"""Calculate importance of each weight for current task"""self.model.eval()for name, param in self.model.named_parameters():self.fisher[name] = torch.zeros_like(param)self.optimal_params[name] = param.data.clone()for data, target in dataloader:self.model.zero_grad()output = self.model(data)loss = F.cross_entropy(output, target)loss.backward()# Fisher = gradientΒ²for name, param in self.model.named_parameters():self.fisher[name] += param.grad.data ** 2# Normalizationfor name in self.fisher:self.fisher[name] /= len(dataloader)def penalty(self):"""Penalty for changing important weights"""loss = 0for name, param in self.model.named_parameters():fisher = self.fisher[name]optimal = self.optimal_params[name]loss += (fisher * (param - optimal) ** 2).sum()return self.lambda_ * loss
class ProgressiveNN(nn.Module):def __init__(self):super().__init__()self.columns = nn.ModuleList() # Each task one columnself.lateral_connections = nn.ModuleList()def add_task(self, input_size, hidden_size, output_size):"""Add new column for new task"""new_column = nn.Sequential(nn.Linear(input_size, hidden_size),nn.ReLU(),nn.Linear(hidden_size, output_size))# Lateral connections from previous columnsif len(self.columns) > 0:lateral = nn.ModuleList([nn.Linear(hidden_size, hidden_size)for _ in range(len(self.columns))])self.lateral_connections.append(lateral)self.columns.append(new_column)# Freeze previous columnsfor i in range(len(self.columns) - 1):for param in self.columns[i].parameters():param.requires_grad = False
class GEM:def __init__(self, model, memory_size_per_task=100):self.model = modelself.memory = {} # {task_id: (data, labels)}self.memory_size = memory_size_per_taskdef store_samples(self, task_id, dataloader):"""Store representative samples from task"""data_list, label_list = [], []for data, labels in dataloader:data_list.append(data)label_list.append(labels)if len(data_list) * data.size(0) >= self.memory_size:breakself.memory[task_id] = (torch.cat(data_list)[:self.memory_size],torch.cat(label_list)[:self.memory_size])def project_gradient(self, current_grad):"""If gradient on previous tasks is negative, project it"""for task_id in self.memory.keys():mem_grad = self.compute_gradient(task_id)# Calculate dot productdot = sum((g1 * g2).sum() for g1, g2 in zip(current_grad, mem_grad))# If negative (damages previous task)if dot < 0:# Projectmem_norm = sum((g ** 2).sum() for g in mem_grad)for i, (g, m) in enumerate(zip(current_grad, mem_grad)):current_grad[i] = g - (dot / mem_norm) * mreturn current_grad
gradient_layer_1 = gradient_output Γ W_n Γ W_(n-1) Γ ... Γ W_2 Γ W_1If each W > 1:gradient becomes very large (e.g., 1.1^100 = 13780)If each W < 1:gradient becomes very small (e.g., 0.9^100 = 0.0000266)
# Method 1: Clip by norm (recommended)torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)# Method 2: Clip by valuetorch.nn.utils.clip_grad_value_(model.parameters(), clip_value=0.5)# Complete usagefor epoch in range(num_epochs):for data, target in dataloader:optimizer.zero_grad()output = model(data)loss = criterion(output, target)loss.backward()# π§ This line prevents catastrophe!torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)optimizer.step()
# Suppose gradients are:gradients = [10.0, 50.0, 100.0, 5.0]norm = sqrt(10Β² + 50Β² + 100Β² + 5Β²) = 112.36# max_norm = 1.0# If norm > max_norm:scale = max_norm / norm = 1.0 / 112.36 = 0.0089# New gradients:clipped_gradients = [g * scale for g in gradients]# = [0.089, 0.445, 0.89, 0.0445]
def init_weights(m):if isinstance(m, nn.Linear):# Xavier initialization to prevent explosionnn.init.xavier_uniform_(m.weight)if m.bias is not None:nn.init.zeros_(m.bias)elif isinstance(m, nn.LSTM):# Orthogonal initialization for RNNfor name, param in m.named_parameters():if 'weight_hh' in name:nn.init.orthogonal_(param)elif 'weight_ih' in name:nn.init.xavier_uniform_(param)elif 'bias' in name:nn.init.zeros_(param)model.apply(init_weights)
class GradientMonitor:def __init__(self, alert_threshold=10.0):self.alert_threshold = alert_thresholdself.history = []def check_gradients(self, model):total_norm = 0for p in model.parameters():if p.grad is not None:param_norm = p.grad.data.norm(2)total_norm += param_norm.item() ** 2total_norm = total_norm ** 0.5self.history.append(total_norm)if total_norm > self.alert_threshold:print(f"β οΈ WARNING: Gradient norm = {total_norm:.2f}")return Truereturn False
Suppose network has 100 layersEach layer: activation = sigmoid(Wx + b)gradient at layer 100 = 1.0gradient at layer 50 = 0.01gradient at layer 10 = 0.0000001 β almost zero!gradient at layer 1 = 10^-20 β completely zero!
# β Bad: Sigmoid (small derivative)activation = nn.Sigmoid()# β Good: ReLU (derivative 1 for x > 0)activation = nn.ReLU()# β Better: Leaky ReLU (non-zero derivative for all x)activation = nn.LeakyReLU(negative_slope=0.01)# β Best for Transformers: GELUactivation = nn.GELU()
class ResidualBlock(nn.Module):def __init__(self, dim):super().__init__()self.layer1 = nn.Linear(dim, dim)self.layer2 = nn.Linear(dim, dim)self.activation = nn.ReLU()def forward(self, x):residual = x # Save inputout = self.layer1(x)out = self.activation(out)out = self.layer2(out)# Add residualout = out + residual # π§ This line saves gradient flow!out = self.activation(out)return out
Without skip connection:gradient_layer1 = gradient Γ W_100 Γ W_99 Γ ... Γ W_2 β becomes zeroWith skip connection:gradient_layer1 = gradient Γ (1 + W_100 Γ W_99 Γ ... Γ W_2) β at least original gradient remains!
# Example oscillating loss:Epoch 1: Loss = 2.5Epoch 2: Loss = 2.1Epoch 3: Loss = 1.8Epoch 4: Loss = 3.2 β π₯ Why worse?Epoch 5: Loss = 1.5Epoch 6: Loss = 2.9 β π₯ Again!Epoch 7: Loss = 1.2
class WarmupScheduler:def __init__(self, optimizer, warmup_steps, total_steps):self.optimizer = optimizerself.warmup_steps = warmup_stepsself.total_steps = total_stepsself.step_count = 0self.base_lr = optimizer.param_groups[0]['lr']def step(self):self.step_count += 1if self.step_count < self.warmup_steps:# Warmup: gradual increaselr = self.base_lr * (self.step_count / self.warmup_steps)else:# Decay: gradual decreaseprogress = (self.step_count - self.warmup_steps) / (self.total_steps - self.warmup_steps)lr = self.base_lr * (1 - progress)for param_group in self.optimizer.param_groups:param_group['lr'] = lr
class EMA:def __init__(self, model, decay=0.999):self.model = modelself.decay = decayself.shadow = {}# Store copy of weightsfor name, param in model.named_parameters():if param.requires_grad:self.shadow[name] = param.data.clone()def update(self):"""Update EMA weights"""for name, param in self.model.named_parameters():if param.requires_grad:new_average = (1.0 - self.decay) * param.data + self.decay * self.shadow[name]self.shadow[name] = new_average.clone()