当前位置：首页 > ops >正文

基于ResNet50的血细胞图像分类模型训练全记录

ops 2025/8/24 16:39:59

基于ResNet50的血细胞图像分类模型训练全记录

项目概述

本项目使用深度学习技术对血细胞图像进行自动分类，识别四种主要的血细胞类型：

嗜酸性粒细胞 (EOSINOPHIL)
淋巴细胞 (LYMPHOCYTE)
单核细胞 (MONOCYTE)
中性粒细胞 (NEUTROPHIL)

模型架构

采用预训练的ResNet50作为骨干网络，这是一个在ImageNet上预训练的深度残差网络，具有强大的特征提取能力。

模型参数统计：

总参数数量：23,516,228
网络深度：50层
预训练权重：ImageNet

数据集信息

训练集：

嗜酸性粒细胞：2,494张图像
淋巴细胞：2,480张图像
单核细胞：2,475张图像
中性粒细胞：2,496张图像

验证集：

嗜酸性粒细胞：374张图像
淋巴细胞：372张图像
单核细胞：371张图像
中性粒细胞：374张图像

训练策略

损失函数

使用加权交叉熵损失，针对数据不平衡问题：

weights = [0.9971, 1.0023, 1.0046, 0.9960]

训练参数

训练轮数： 50 epochs
早停机制： 验证准确率连续下降时自动停止
优化器： AdamW
学习率调度： 余弦退火重启调度器

核心代码实现

1. 数据集类定义

class BloodCellDataset(Dataset):"""血细胞图像数据集"""def __init__(self, data_root: str, split: str, transform=None):self.data_root = Path(data_root)self.split = splitself.transform = transform# 类别映射self.class_names = ['EOSINOPHIL', 'LYMPHOCYTE', 'MONOCYTE', 'NEUTROPHIL']self.class_to_idx = {name: i for i, name in enumerate(self.class_names)}# 加载图像路径和标签self.samples = self._load_samples()def _load_samples(self):"""加载样本路径和标签"""samples = []split_dir = self.data_root / self.splitfor class_name in self.class_names:class_dir = split_dir / class_nameif not class_dir.exists():continueclass_idx = self.class_to_idx[class_name]# 获取所有图片文件for img_path in class_dir.glob('*.jpeg'):samples.append((str(img_path), class_idx))return samples

2. 模型架构定义

class BloodCellClassifier(nn.Module):"""血细胞分类器"""def __init__(self, model_name: str = 'resnet50', num_classes: int = 4, pretrained: bool = True):super(BloodCellClassifier, self).__init__()self.model_name = model_nameself.num_classes = num_classes# 加载预训练模型if model_name == 'resnet50':self.backbone = models.resnet50(pretrained=pretrained)in_features = self.backbone.fc.in_featuresself.backbone.fc = nn.Linear(in_features, num_classes)def forward(self, x):return self.backbone(x)

3. 数据增强策略

def _setup_transforms(self):"""设置数据变换"""# 训练时的数据增强self.train_transform = transforms.Compose([transforms.Resize((256, 256)),transforms.RandomResizedCrop(224, scale=(0.8, 1.0)),transforms.RandomHorizontalFlip(p=0.5),transforms.RandomRotation(degrees=10),transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.1, hue=0.1),transforms.ToTensor(),transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])])# 验证和测试时的变换self.val_transform = transforms.Compose([transforms.Resize((224, 224)),transforms.ToTensor(),transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])])

4. 训练循环实现

def train_epoch(self) -> Dict[str, float]:"""训练一个epoch"""self.model.train()running_loss = 0.0correct = 0total = 0pbar = tqdm(self.train_loader, desc='训练中')for batch_idx, (images, labels) in enumerate(pbar):images, labels = images.to(self.device), labels.to(self.device)# 前向传播self.optimizer.zero_grad()outputs = self.model(images)loss = self.criterion(outputs, labels)# 反向传播loss.backward()self.optimizer.step()# 统计running_loss += loss.item()_, predicted = outputs.max(1)total += labels.size(0)correct += predicted.eq(labels).sum().item()# 更新进度条pbar.set_postfix({'Loss': f'{loss.item():.4f}','Acc': f'{100.*correct/total:.2f}%'})epoch_loss = running_loss / len(self.train_loader)epoch_acc = 100. * correct / totalreturn {'loss': epoch_loss, 'accuracy': epoch_acc}

5. 早停机制

def _setup_training(self):"""设置训练相关组件"""# 计算类别权重来处理不平衡数据集class_counts = self._calculate_class_weights()if class_counts:weights = torch.tensor([1.0/count for count in class_counts.values()], dtype=torch.float32)weights = weights / weights.sum() * len(weights)  # 归一化self.criterion = nn.CrossEntropyLoss(weight=weights.to(self.device))# 优化器 - 使用AdamWself.optimizer = optim.AdamW(self.model.parameters(),lr=0.001,weight_decay=0.01,betas=(0.9, 0.999),eps=1e-8)# 余弦退火重启调度器self.scheduler = optim.lr_scheduler.CosineAnnealingWarmRestarts(self.optimizer,T_0=10,  # 初始重启周期T_mult=2,  # 周期倍增因子eta_min=1e-6  # 最小学习率)# 早停机制self.early_stopping_patience = 15self.early_stopping_counter = 0self.best_val_loss = float('inf')

6. 主训练函数

def train(self, epochs: int = 50):"""训练模型"""print(f"🚀 开始训练 {epochs} 个epochs")best_val_acc = 0.0train_losses = []val_losses = []train_accs = []val_accs = []for epoch in range(epochs):# 训练train_metrics = self.train_epoch()# 验证val_metrics = self.validate()# 记录指标train_losses.append(train_metrics['loss'])val_losses.append(val_metrics['loss'])train_accs.append(train_metrics['accuracy'])val_accs.append(val_metrics['accuracy'])# 学习率调度self.scheduler.step()# 早停检查if val_metrics['loss'] < self.best_val_loss:self.best_val_loss = val_metrics['loss']self.early_stopping_counter = 0else:self.early_stopping_counter += 1if self.early_stopping_counter >= self.early_stopping_patience:print(f"\n⏱️ Early stopping triggered after {epoch+1} epochs")break# 保存最佳模型if val_metrics['accuracy'] > best_val_acc:best_val_acc = val_metrics['accuracy']self.save_model('best.pt')print(f"✅ 保存最佳模型 (验证准确率: {best_val_acc:.2f}%)")