深度学习知识体系总结与学习路径

# PyTorch 完整训练循环模板 import torch import torch.nn as nn from torch.utils.data import DataLoader from torch.utils.tensorboard import SummaryWriter from tqdm import tqdm class Trainer: def __init__(self, model, train_loader, val_loader, config): self.model = model self.train_loader = train_loader self.val_loader = val_loader self.config = config self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.model.to(self.device) self.criterion = nn.CrossEntropyLoss() self.optimizer = torch.optim.AdamW( model.parameters(), lr=config['lr'], weight_decay=config['weight_decay'] ) self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( self.optimizer, T_max=config['epochs'] ) self.writer = SummaryWriter(log_dir=config['log_dir']) self.best_acc = 0.0 def train_epoch(self): self.model.train() total_loss = 0.0 pbar = tqdm(self.train_loader, desc='Training') for batch_idx, (inputs, targets) in enumerate(pbar): inputs, targets = inputs.to(self.device), targets.to(self.device) self.optimizer.zero_grad() outputs = self.model(inputs) loss = self.criterion(outputs, targets) loss.backward() torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0) self.optimizer.step() total_loss += loss.item() pbar.set_postfix({'loss': loss.item()}) return total_loss / len(self.train_loader) @torch.no_grad() def validate(self): self.model.eval() total_loss = 0.0 correct = 0 total = 0 for inputs, targets in self.val_loader: inputs, targets = inputs.to(self.device), targets.to(self.device) outputs = self.model(inputs) loss = self.criterion(outputs, targets) total_loss += loss.item() _, predicted = outputs.max(1) total += targets.size(0) correct += predicted.eq(targets).sum().item() accuracy = 100.0 * correct / total avg_loss = total_loss / len(self.val_loader) return avg_loss, accuracy def fit(self): for epoch in range(self.config['epochs']): train_loss = self.train_epoch() val_loss, val_acc = self.validate() self.scheduler.step() self.writer.add_scalar('Loss/train', train_loss, epoch) self.writer.add_scalar('Loss/val', val_loss, epoch) self.writer.add_scalar('Acc/val', val_acc, epoch) if val_acc > self.best_acc: self.best_acc = val_acc torch.save(self.model.state_dict(), f"best_model_epoch{epoch}.pth") print(f"Epoch {epoch+1}/{self.config['epochs']} | " f"Train Loss: {train_loss:.4f} | Val Loss: {val_loss:.4f} | " f"Val Acc: {val_acc:.2f}%") self.writer.close() print(f"Training completed. Best accuracy: {self.best_acc:.2f}%") # 使用示例 if __name__ == '__main__': config = { 'lr': 1e-3, 'weight_decay': 1e-4, 'epochs': 50, 'log_dir': 'runs/experiment_001' } model = torchvision.models.resnet18(pretrained=True) model.fc = nn.Linear(512, 10) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False) trainer = Trainer(model, train_loader, val_loader, config) trainer.fit()

# 学习率范围测试 (LR Range Test / Cyclical LR Finder) import torch from torch_lr_finder import LRFinder import matplotlib.pyplot as plt def find_optimal_lr(model, train_loader, criterion, optimizer, device): lr_finder = LRFinder(model, optimizer, criterion, device=device) lr_finder.range_test(train_loader, end_lr=10, num_iter=100, step_mode="exp") # 绘制学习率 vs Loss 曲线 fig, ax = plt.subplots(1, 1, figsize=(10, 6)) ax.plot(lr_finder.history['lr'], lr_finder.history['loss']) ax.set_xscale('log') ax.set_xlabel('Learning Rate') ax.set_ylabel('Loss') ax.set_title('LR Range Test') ax.grid(True, alpha=0.3) plt.show() # 找到最优学习率（loss下降最快的区间） min_loss_idx = lr_finder.history['loss'].index(min(lr_finder.history['loss'])) optimal_lr = lr_finder.history['lr'][min_loss_idx] print(f"Optimal LR range: {optimal_lr/10:.2e} ~ {optimal_lr*10:.2e}") lr_finder.reset() return optimal_lr if __name__ == '__main__': model = torchvision.models.resnet18() optimizer = torch.optim.SGD(model.parameters(), lr=1e-7, weight_decay=1e-4) criterion = nn.CrossEntropyLoss() device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') optimal_lr = find_optimal_lr(model, train_loader, criterion, optimizer, device)

# PyTorch DDP (Distributed Data Parallel) 基础配置 import torch import torch.distributed as dist import torch.multiprocessing as mp from torch.nn.parallel import DistributedDataParallel as DDP from torch.utils.data.distributed import DistributedSampler def setup_ddp(rank, world_size): dist.init_process_group( backend='nccl' if torch.cuda.is_available() else 'gloo', init_method='tcp://localhost:12355', rank=rank, world_size=world_size ) torch.cuda.set_device(rank) def cleanup_ddp(): dist.destroy_process_group() def train_ddp(rank, world_size, model_fn, train_dataset, config): setup_ddp(rank, world_size) model = model_fn().to(rank) model = DDP(model, device_ids=[rank], find_unused_parameters=False) sampler = DistributedSampler(train_dataset, num_replicas=world_size, rank=rank) dataloader = DataLoader(train_dataset, batch_size=config['batch_size'], sampler=sampler, num_workers=4, pin_memory=True) optimizer = torch.optim.AdamW(model.parameters(), lr=config['lr']) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( optimizer, T_max=config['epochs'] ) for epoch in range(config['epochs']): sampler.set_epoch(epoch) # 确保每个epoch数据打乱不同 model.train() for inputs, targets in dataloader: inputs, targets = inputs.to(rank), targets.to(rank) optimizer.zero_grad() outputs = model(inputs) loss = nn.CrossEntropyLoss()(outputs, targets) loss.backward() optimizer.step() scheduler.step() if rank == 0: print(f"Epoch {epoch+1}, Loss: {loss.item():.4f}") cleanup_ddp() if __name__ == '__main__': world_size = torch.cuda.device_count() if torch.cuda.is_available() else 2 config = {'batch_size': 32, 'lr': 1e-3, 'epochs': 10} mp.spawn(train_ddp, args=(world_size, create_model, train_dataset, config), nprocs=world_size, join=True)