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

我的第一个医学AI项目：用PyTorch训练一个肺炎X光片分类器（附完整代码与数据集）

news 2026/7/3 13:50:54

从零构建医学影像AI：PyTorch实战肺炎X光片分类系统

深夜的医院影像科，实习医生小李正对着电脑屏幕上一张张肺部X光片皱眉——这些胸片中有正常肺部、细菌性肺炎和病毒性肺炎三种类型，细微的纹理差异让刚入行的他难以把握。这让我想起三年前在MIT实验室遇到的类似场景，当时我们团队用深度学习技术开发了一个辅助诊断系统。今天，我将带您完整复现这个项目的开发过程，从环境搭建到模型部署，手把手教您构建一个能自动识别肺炎类型的智能系统。

1. 医学AI项目的特殊性与准备

医学影像分析不同于普通计算机视觉任务，一张胸片背后可能关系着患者的生命健康。在开始编码前，我们需要特别关注三个核心问题：数据合规性、模型可解释性和临床实用性。我强烈建议在个人电脑上使用conda创建独立环境，避免依赖冲突：

conda create -n medai python=3.8 conda activate medai pip install torch==1.12.0+cu113 torchvision==0.13.0+cu113 -f https://download.pytorch.org/whl/torch_stable.html pip install pandas matplotlib opencv-python

提示：医学图像通常采用DICOM格式，但公开数据集多已转换为PNG/JPG。建议安装pydicom库以备不时之需

我们将使用COVID-19 Radiography Dataset，这个数据集包含：

正常肺部图像：1,345张
新冠肺炎图像：3,616张
病毒性肺炎图像：1,345张
细菌性肺炎图像：2,538张

医学图像预处理要点：

窗宽窗位调整（对比度优化）
非均匀光照校正
肺野区域分割（ROI提取）
标准化到统一尺寸（通常256×256或512×512）

2. 构建医学专用数据管道

医学影像的Dataset类需要特殊设计，考虑以下关键点：

import torch from torch.utils.data import Dataset import cv2 import numpy as np class ChestXrayDataset(Dataset): def __init__(self, dataframe, transform=None): self.df = dataframe self.transform = transform # 医学图像特有的预处理流程 self.med_transform = Compose([ RandomGammaCorrection(gamma_range=(0.8, 1.2)), # 模拟不同曝光条件 LungFieldSegmentation(), # 肺野分割 CLAHE(clip_limit=2.0) # 对比度受限自适应直方图均衡 ]) def __getitem__(self, idx): img_path = self.df.iloc[idx]['path'] image = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE) image = self.med_transform(image) if self.transform: image = self.transform(image) label = self.df.iloc[idx]['label'] return image, torch.tensor(label, dtype=torch.long) def __len__(self): return len(self.df)

注意：医学数据必须严格划分训练/验证/测试集，建议采用分层抽样保证各类别比例一致

数据增强策略对比：

常规增强	医学专用增强	适用场景
随机旋转	小角度旋转(±15°)	保持解剖结构
颜色抖动	灰度值扰动	模拟不同设备
随机裁剪	固定中心裁剪	保留关键区域
水平翻转	禁止翻转	维持左右解剖对称性

3. 设计医学影像专用网络架构

基于ResNet50改进的MedResNet模型：

import torch.nn as nn import torchvision.models as models class MedResNet(nn.Module): def __init__(self, num_classes=3): super().__init__() base_model = models.resnet50(pretrained=True) # 修改第一层卷积：输入通道改为1，核大小适应医学图像 base_model.conv1 = nn.Conv2d(1, 64, kernel_size=7, stride=2, padding=3, bias=False) # 添加注意力模块 self.attention = nn.Sequential( nn.AdaptiveAvgPool2d(1), nn.Conv2d(2048, 512, 1), nn.ReLU(), nn.Conv2d(512, 2048, 1), nn.Sigmoid() ) self.backbone = nn.Sequential(*list(base_model.children())[:-2]) self.pool = nn.AdaptiveAvgPool2d(1) self.classifier = nn.Linear(2048, num_classes) def forward(self, x): features = self.backbone(x) att = self.attention(features) features = features * att pooled = self.pool(features).view(features.size(0), -1) return self.classifier(pooled)

模型设计考量：

输入层调整：医学影像多为单通道，需修改首层卷积
注意力机制：引导模型关注肺野关键区域
预训练策略：采用自然图像预训练+医学图像微调
特征保留：避免过度下采样，保留细小病灶特征

4. 训练策略与医学评估指标

医学模型需要特殊的训练循环设计：

def train_epoch(model, loader, criterion, optimizer, device): model.train() total_loss = 0 correct = 0 for images, labels in loader: images, labels = images.to(device), labels.to(device) optimizer.zero_grad() outputs = model(images) loss = criterion(outputs, labels) # 添加L2正则化和梯度裁剪 l2_reg = torch.tensor(0.).to(device) for param in model.parameters(): l2_reg += torch.norm(param) loss += 0.001 * l2_reg loss.backward() nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() _, predicted = torch.max(outputs.data, 1) correct += (predicted == labels).sum().item() return total_loss / len(loader), correct / len(loader.dataset)

医学评估指标矩阵：

指标	公式	医学意义
敏感度	TP/(TP+FN)	避免漏诊危重病例
特异度	TN/(TN+FP)	减少假阳性带来的过度医疗
AUC-ROC	-	综合评估模型判别能力
F1-Score	2(PrecisionRecall)/(Precision+Recall)	平衡精确率和召回率

5. 模型解释与临床应用

可视化模型决策过程对医学AI至关重要：

import torch.nn.functional as F def generate_cam(model, image): model.eval() features = model.backbone(image.unsqueeze(0)) att = model.attention(features) weighted_features = features * att # 获取类别权重 params = list(model.classifier.parameters()) weight_softmax = params[0].data # 生成类激活图 bz, nc, h, w = weighted_features.shape cams = torch.zeros((bz, h, w)) for idx, cls_weight in enumerate(weight_softmax): cam = (cls_weight * weighted_features).sum(dim=1) cam = F.relu(cam) cam = cam - cam.min() cam = cam / cam.max() cams += cam.detach().cpu() return cams.squeeze()

临床部署建议：