PyTorch深度学习框架60天进阶学习计划 - 第36天：医疗影像诊断（二）

PyTorch深度学习框架60天进阶学习计划 - 第36天：医疗影像诊断（二）

朋友们，我们继续上篇文章的内容。

六、实验案例：肺部结节检测

让我们设计一个具体的实验案例来应用我们学到的知识。这个案例将展示如何使用3D ResNet进行肺部结节检测，这是肺癌早期筛查的重要任务。

6.1 实验设置

6.2 使用LIDC-IDRI数据集

肺结像数据库联盟与图像数据库资源计划（LIDC-IDRI）是肺部CT分析的标准数据集之一。下面是如何设置实验的示例代码：

import os
import pandas as pd
import numpy as np
import torch
from torch.utils.data import DataLoader, random_split
import matplotlib.pyplot as plt
from tqdm import tqdm
import SimpleITK as sitk
import pylidc as pl

# 导入我们之前定义的模块
from preprocess import preprocess_ct_scan
from augmentation import CTAugmentation3D, LungCTDataset
from resnet3d import resnet18_3d, resnet50_3d
from loss_functions import FocalLoss, CombinedLoss, calculate_class_weights
from train_pipeline import LungCTTrainer, set_seed

# 设置随机种子
set_seed(42)

# 配置参数
CONFIG = {
    'data_dir': '/path/to/LIDC-IDRI',  # LIDC-IDRI数据集路径
    'output_dir': './output',          # 输出目录
    'input_size': (128, 128, 128),     # 输入大小
    'batch_size': 8,                  # 批处理大小
    'lr': 0.0002,                     # 学习率
    'epochs': 50,                     # 训练轮次
    'model_type': 'resnet18',         # 模型类型
    'focal_gamma': 2.0,               # Focal Loss的gamma参数
    'num_workers': 4,                 # 数据加载器的工作进程数
}

# 创建输出目录
os.makedirs(CONFIG['output_dir'], exist_ok=True)

def load_lidc_dataset():
    """
    加载LIDC-IDRI数据集并将其处理为适合深度学习的格式
    """
    print("Loading LIDC-IDRI dataset...")
    
    # 创建存储处理后数据的目录
    processed_dir = os.path.join(CONFIG['output_dir'], 'processed')
    os.makedirs(processed_dir, exist_ok=True)
    
    # 如果已经处理过数据，直接加载
    metadata_path = os.path.join(processed_dir, 'metadata.csv')
    if os.path.exists(metadata_path):
        print("Loading pre-processed data...")
        metadata = pd.read_csv(metadata_path)
        ct_paths = metadata['ct_path'].tolist()
        labels = metadata['has_nodule'].astype(int).tolist()
        return ct_paths, labels
    
    # 否则，从原始数据开始处理
    print("Processing raw LIDC-IDRI data...")
    
    # 获取所有LIDC-IDRI扫描
    scans = pl.query(pl.Scan)
    
    ct_paths = []
    labels = []
    
    for scan in tqdm(scans):
        try:
            # 获取这个扫描中的结节
            nodules = scan.cluster_annotations()
            
            # 如果结节超过3mm，标记为正样本(有肺结节)
            has_nodule = any(n.diameter() >= 3.0 for n in nodules)
            
            # 预处理CT扫描
            patient_id = scan.patient_id
            processed_path = os.path.join(processed_dir, f"{patient_id}.npy")
            
            # 如果还没处理过，则进行预处理
            if not os.path.exists(processed_path):
                # 获取DICOM路径
                dicom_path = os.path.join(CONFIG['data_dir'], 
                                         scan.get_path_to_dicom_files())
                
                # 预处理CT扫描
                processed_ct = preprocess_ct_scan(
                    dicom_path, output_size=CONFIG['input_size']
                )
                
                # 保存预处理后的数据
                np.save(processed_path, processed_ct)
            
            # 添加到数据列表
            ct_paths.append(processed_path)
            labels.append(int(has_nodule))
            
        except Exception as e:
            print(f"Error processing scan {scan.patient_id}: {e}")
            continue
    
    # 保存元数据
    metadata = pd.DataFrame({
        'ct_path': ct_paths,
        'has_nodule': labels
    })
    metadata.to_csv(metadata_path, index=False)
    
    print(f"Processed {len(ct_paths)} CT scans.")
    print(f"Positive samples (with nodules): {sum(labels)}")
    print(f"Negative samples (without nodules): {len(labels) - sum(labels)}")
    
    return ct_paths, labels

def run_experiment():
    """
    运行肺部结节检测实验
    """
    # 加载数据集
    ct_paths, labels = load_lidc_dataset()
    
    # 计算类别权重
    class_weights = calculate_class_weights(labels, method='effective')
    print(f"Class weights: {class_weights}")
    
    # 创建数据增强器
    augmentation = CTAugmentation3D(
        rotation_range=(-15, 15),
        scale_range=(0.85, 1.15),
        shift_range=(-10, 10),
        noise_factor=0.02,
        brightness_range=(0.9, 1.1),
        contrast_range=(0.9, 1.1),
        p_rotation=0.7,
        p_scale=0.5,
        p_shift=0.5,
        p_noise=0.3,
        p_brightness=0.4,
        p_contrast=0.4,
        p_elastic=0.2
    )
    
    # 创建数据集
    full_dataset = LungCTDataset(ct_paths, labels, transform=augmentation)
    
    # 数据集划分
    train_size = int(0.7 * len(full_dataset))
    val_size = int(0.15 * len(full_dataset))
    test_size = len(full_dataset) - train_size - val_size
    
    train_dataset, val_dataset, test_dataset = random_split(
        full_dataset, [train_size, val_size, test_size]
    )
    
    print(f"Dataset sizes - Train: {train_size}, Val: {val_size}, Test: {test_size}")
    
    # 创建模型
    if CONFIG['model_type'] == 'resnet18':
        model = resnet18_3d(num_classes=2)
    elif CONFIG['model_type'] == 'resnet50':
        model = resnet50_3d(num_classes=2)
    else:
        raise ValueError(f"Unknown model type: {CONFIG['model_type']}")
    
    # 创建损失函数
    loss_fn = FocalLoss(alpha=class_weights, gamma=CONFIG['focal_gamma'])
    
    # 创建训练器
    trainer = LungCTTrainer(
        model=model,
        train_dataset=train_dataset,
        val_dataset=val_dataset,
        test_dataset=test_dataset,
        batch_size=CONFIG['batch_size'],
        lr=CONFIG['lr'],
        loss_fn=loss_fn,
        class_weights=class_weights,
        experiment_name=f"lung_nodule_{CONFIG['model_type']}"
    )
    
    # 训练模型
    history = trainer.train(epochs=CONFIG['epochs'])
    
    # 测试模型
    test_results = trainer.test()
    
    # 保存测试结果
    results_path = os.path.join(CONFIG['output_dir'], 'test_results.csv')
    pd.DataFrame([test_results]).to_csv(results_path, index=False)
    
    # 打印测试结果汇总
    print("\nTest Results Summary:")
    print(f"Accuracy: {test_results['accuracy']:.4f}")
    print(f"Precision: {test_results['precision']:.4f}")
    print(f"Recall: {test_results['recall']:.4f}")
    print(f"F1 Score: {test_results['f1']:.4f}")
    if test_results['auc']:
        print(f"AUC: {test_results['auc']:.4f}")
    
    # 绘制ROC曲线(二分类问题)
    if test_results['auc']:
        from sklearn.metrics import roc_curve, auc
        fpr, tpr, _ = roc_curve(
            test_results['targets'], 
            np.array(test_results['probabilities'])[:, 1]
        )
        roc_auc = auc(fpr, tpr)
        
        plt.figure(figsize=(10, 8))
        plt.plot(fpr, tpr, color='darkorange', lw=2, 
                label=f'ROC curve (area = {roc_auc:.3f})')
        plt.plot([0, 1], [0, 1], color='navy', lw=2, linestyle='--')
        plt.xlim([0.0, 1.0])
        plt.ylim([0.0, 1.05])
        plt.xlabel('False Positive Rate')
        plt.ylabel('True Positive Rate')
        plt.title('Receiver Operating Characteristic')
        plt.legend(loc="lower right")
        plt.savefig(os.path.join(CONFIG['output_dir'], 'roc_curve.png'))
        plt.show()
    
    # 混淆矩阵
    from sklearn.metrics import confusion_matrix
    import seaborn as sns
    
    cm = confusion_matrix(test_results['targets'], test_results['predictions'])
    plt.figure(figsize=(8, 6))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
               xticklabels=['No Nodule', 'Nodule'],
               yticklabels=['No Nodule', 'Nodule'])
    plt.xlabel('Predicted')
    plt.ylabel('True')
    plt.title('Confusion Matrix')
    plt.savefig(os.path.join(CONFIG['output_dir'], 'confusion_matrix.png'))
    plt.show()
    
    return test_results, history

if __name__ == "__main__":
    test_results, history = run_experiment()

6.3 实验结果分析

一个典型的肺部结节检测实验可能会产生以下结果：

性能指标

常见错误类型分析

1. 假阳性（误报）：

   • 血管交叉误判为结节
   • 炎症区域误判为结节
   • 肺门区域组织密度变化误判

2. 假阴性（漏报）：

   • 与血管贴近的结节被忽略
   • 毛玻璃结节难以检测
   • 小结节（<5mm）检测率低

解决改进方案

针对上述问题，可以采取以下措施改进模型：

1. 数据增强优化：

   • 针对小结节样本进行重点增强
   • 添加更多模拟毛玻璃结节的样本

2. 损失函数调整：

   • 增加漏检惩罚权重（减少假阴性）
   • 考虑使用非对称Focal Loss

3. 模型优化：

   • 使用多尺度特征融合
   • 尝试注意力机制，突出潜在结节区域

七、医疗AI应用的伦理和局限性

在医疗AI发展中，伦理问题和技术局限性同样重要：

7.1 伦理问题

1. 数据隐私：患者数据需严格保护
2. 责任归属：AI辅助诊断的最终责任归属问题
3. 透明性：AI决策过程的可解释性
4. 公平性：不同群体间的公平诊断问题

7.2 技术局限性

1. 数据集代表性：医学数据集通常规模有限，可能不足以代表全人群
2. 域泛化能力：模型在不同扫描设备或医院间的泛化性
3. 罕见病例：对罕见病例或变异的识别能力有限
4. 假阳性率：控制假阳性率与保证高敏感性间的平衡

八、总结与展望

8.1 课程总结

今天我们学习了：

• 3D ResNet网络的设计与实现
• 医学影像预处理的关键步骤
• 针对医学影像的数据增强技术
• 处理类别不平衡的损失函数设计
• 肺部结节检测的完整工作流程

8.2 未来发展方向

医学影像AI正朝以下方向发展：

1. 多模态融合：结合CT、MRI、病理等多种数据源
2. 自监督学习：减少对标注数据的依赖
3. 因果推理：引入因果关系，提高决策可解释性
4. 联邦学习：保护患者隐私的分布式学习
5. 临床工作流集成：AI技术向临床工作流的无缝整合

希望今天的学习让你对医疗影像AI有了更深入的理解！在深度学习中，技术能力固然重要，但对应用领域（如医学）的理解同样关键。记住，在医疗AI中，我们不仅追求高精度，更要保证安全性和可靠性。

清华大学全五版的《DeepSeek教程》完整的文档需要的朋友，关注我私信：deepseek 即可获得。

怎么样今天的内容还满意吗？再次感谢朋友们的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财务自由，还请给个赞，谢谢！