用PaddlePaddle动态图复现ResNet50，从零搭建一个眼底病变分类器（附完整代码）-开发者社区

基于PaddlePaddle动态图的ResNet50眼底病变分类实战指南

在医疗影像分析领域，自动化的疾病筛查系统正逐渐成为临床医生的得力助手。眼底病变的早期发现对预防视力损伤至关重要，而深度学习技术为这一任务提供了新的可能性。本文将带领读者使用PaddlePaddle的动态图模式，从零开始构建一个ResNet50模型，实现对病理性近视的自动分类。不同于静态图的编程方式，动态图模式允许我们以更直观的Python编程思维来构建和调试模型，特别适合初学者快速上手深度学习项目实践。

1. 项目环境与数据准备

1.1 环境配置与PaddlePaddle安装

在开始项目前，我们需要确保环境配置正确。PaddlePaddle支持多种安装方式，推荐使用conda创建独立的Python环境：

conda create -n paddle python=3.7 conda activate paddle pip install paddlepaddle-gpu==2.3.0 -i https://mirror.baidu.com/pypi/simple

对于没有GPU设备的用户，可以安装CPU版本：

pip install paddlepaddle==2.3.0 -i https://mirror.baidu.com/pypi/simple

提示：安装完成后，可以通过python -c "import paddle; paddle.utils.run_check()"验证安装是否成功。

1.2 眼底数据集解析与预处理

我们使用的PALM眼底筛查数据集包含三类图像：

病理性近视(PM)：文件名以P开头
高度近视(HM)：文件名以H开头
正常视力(Normal)：文件名以N开头

为简化问题，我们将病理性近视作为正类(标签1)，其余作为负类(标签0)。首先让我们查看数据分布：

import os from collections import Counter datadir = 'PALM-Training400' filenames = os.listdir(datadir) label_counts = Counter(['PM' if f.startswith('P') else 'Non-PM' for f in filenames]) print(label_counts) # 输出：Counter({'Non-PM': 300, 'PM': 100})

可以看到数据存在类别不平衡问题，这在后续训练中需要特别注意。接下来我们定义数据预处理流程：

import cv2 import numpy as np def transform_img(img): # 统一缩放到224x224尺寸，适应ResNet输入 img = cv2.resize(img, (224, 224)) # 转换通道顺序为[C,H,W] img = np.transpose(img, (2,0,1)).astype('float32') # 归一化到[-1,1]范围 img = img / 255. * 2 - 1.0 return img

2. ResNet50模型架构深度解析

2.1 残差连接的核心思想

ResNet的核心创新在于引入了残差学习机制，解决了深层网络训练中的梯度消失问题。传统卷积网络的堆叠会遇到性能退化问题，即随着深度增加，准确率不升反降。ResNet通过跨层连接(shortcut)实现了恒等映射，让网络可以专注于学习残差部分。

残差块的基本数学表达为：

y = F(x) + x

其中：

x是输入特征
F(x)是残差函数
+表示逐元素相加

2.2 Bottleneck结构实现

ResNet50使用的是Bottleneck结构的残差块，包含三个卷积层：

1x1卷积：降维，减少计算量
3x3卷积：空间特征提取
1x1卷积：升维，恢复通道数

在PaddlePaddle中实现如下：

import paddle from paddle.nn import Conv2D, BatchNorm2D, ReLU class BottleneckBlock(paddle.nn.Layer): def __init__(self, in_channels, out_channels, stride=1, shortcut=True): super().__init__() self.conv1 = Conv2D(in_channels, out_channels, 1, stride=1, bias_attr=False) self.bn1 = BatchNorm2D(out_channels) self.conv2 = Conv2D(out_channels, out_channels, 3, stride=stride, padding=1, bias_attr=False) self.bn2 = BatchNorm2D(out_channels) self.conv3 = Conv2D(out_channels, out_channels*4, 1, bias_attr=False) self.bn3 = BatchNorm2D(out_channels*4) if not shortcut: self.shortcut = Conv2D(in_channels, out_channels*4, 1, stride=stride) self.relu = ReLU() self.shortcut_flag = shortcut def forward(self, x): identity = x out = self.conv1(x) out = self.bn1(out) out = self.relu(out) out = self.conv2(out) out = self.bn2(out) out = self.relu(out) out = self.conv3(out) out = self.bn3(out) if not self.shortcut_flag: identity = self.shortcut(x) out += identity return self.relu(out)

2.3 完整的ResNet50架构

基于Bottleneck块，我们可以构建完整的ResNet50模型。模型包含5个阶段：

初始卷积和池化层
4个由Bottleneck块组成的阶段
全局平均池化和全连接层

class ResNet50(paddle.nn.Layer): def __init__(self, num_classes=1): super().__init__() self.conv1 = Conv2D(3, 64, 7, stride=2, padding=3, bias_attr=False) self.bn1 = BatchNorm2D(64) self.relu = ReLU() self.maxpool = paddle.nn.MaxPool2D(3, stride=2, padding=1) # 四个残差阶段 self.layer1 = self._make_layer(64, 64, 3, stride=1) self.layer2 = self._make_layer(256, 128, 4, stride=2) self.layer3 = self._make_layer(512, 256, 6, stride=2) self.layer4 = self._make_layer(1024, 512, 3, stride=2) self.avgpool = paddle.nn.AdaptiveAvgPool2D(1) self.fc = paddle.nn.Linear(2048, num_classes) def _make_layer(self, in_channels, out_channels, blocks, stride): layers = [] # 第一个块可能需要下采样 layers.append(BottleneckBlock(in_channels, out_channels, stride, shortcut=(stride==1))) for _ in range(1, blocks): layers.append(BottleneckBlock(out_channels*4, out_channels)) return paddle.nn.Sequential(*layers) def forward(self, x): x = self.conv1(x) x = self.bn1(x) x = self.relu(x) x = self.maxpool(x) x = self.layer1(x) x = self.layer2(x) x = self.layer3(x) x = self.layer4(x) x = self.avgpool(x) x = paddle.flatten(x, 1) x = self.fc(x) return x

3. 模型训练与优化策略

3.1 数据加载器实现

高效的数据加载对训练过程至关重要。我们实现一个自定义DataLoader：

from paddle.io import Dataset import random class EyeDataset(Dataset): def __init__(self, datadir, mode='train'): self.datadir = datadir self.mode = mode self.file_list = self._load_data() if mode == 'train': random.shuffle(self.file_list) def _load_data(self): file_list = [] for filename in os.listdir(self.datadir): if filename.startswith('P'): label = 1 else: label = 0 file_list.append((os.path.join(self.datadir, filename), label)) return file_list def __getitem__(self, idx): img_path, label = self.file_list[idx] img = cv2.imread(img_path) img = transform_img(img) return img.astype('float32'), np.array([label]).astype('float32') def __len__(self): return len(self.file_list)

3.2 训练流程实现

考虑到医疗数据的不平衡性，我们采用加权损失函数和动态学习率：

def train_model(): # 初始化模型 model = ResNet50() model.train() # 定义优化器和学习率调度 scheduler = paddle.optimizer.lr.PolynomialDecay( learning_rate=0.01, decay_steps=1000, end_lr=0.0001) optimizer = paddle.optimizer.Momentum( learning_rate=scheduler, momentum=0.9, parameters=model.parameters(), weight_decay=0.001) # 定义加权损失函数 pos_weight = paddle.to_tensor([3.0]) # 正样本权重 criterion = paddle.nn.BCEWithLogitsLoss(weight=pos_weight) # 数据加载 train_dataset = EyeDataset('PALM-Training400', 'train') train_loader = paddle.io.DataLoader( train_dataset, batch_size=16, shuffle=True) # 训练循环 for epoch in range(10): for batch_idx, (data, target) in enumerate(train_loader): output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() optimizer.clear_grad() if batch_idx % 10 == 0: print(f'Epoch: {epoch}, Batch: {batch_idx}, Loss: {loss.numpy()[0]}') # 每个epoch后验证 validate(model) return model

3.3 模型验证与指标分析

医疗领域的模型评估需要更全面的指标：

def validate(model): model.eval() dataset = EyeDataset('PALM-Validation400', 'valid') loader = paddle.io.DataLoader(dataset, batch_size=8) all_preds = [] all_labels = [] with paddle.no_grad(): for data, target in loader: output = model(data) pred = paddle.nn.functional.sigmoid(output) all_preds.append(pred.numpy()) all_labels.append(target.numpy()) preds = np.concatenate(all_preds) labels = np.concatenate(all_labels) # 计算各项指标 from sklearn.metrics import accuracy_score, roc_auc_score, f1_score acc = accuracy_score(labels, preds > 0.5) auc = roc_auc_score(labels, preds) f1 = f1_score(labels, preds > 0.5) print(f'Validation - Accuracy: {acc:.4f}, AUC: {auc:.4f}, F1: {f1:.4f}') model.train()

4. 模型部署与性能优化

4.1 模型保存与加载

训练完成后，我们需要保存模型以备后续使用：

def save_model(model, path): paddle.save(model.state_dict(), path + '.pdparams') paddle.save(optimizer.state_dict(), path + '.pdopt') def load_model(path, model): state_dict = paddle.load(path + '.pdparams') model.set_state_dict(state_dict) return model

4.2 模型量化与加速

为提升推理速度，我们可以应用动态图量化：

quant_model = paddle.quantization.quantize_dynamic( model, {paddle.nn.Linear, paddle.nn.Conv2D}, dtype='int8')

4.3 实际应用示例

最后，我们实现一个简单的预测函数：

def predict_image(model, img_path): img = cv2.imread(img_path) img = transform_img(img) img = paddle.to_tensor(img[np.newaxis, ...]) model.eval() with paddle.no_grad(): logit = model(img) prob = paddle.nn.functional.sigmoid(logit).numpy()[0][0] print(f'Pathological Myopia Probability: {prob:.4f}') return prob

在实际医疗场景中部署时，还需要考虑以下优化点：