Python 医疗 AI 常用库及实战案例解析

自动化 & Agent

实战案例解析

医学图像处理与分析：MONAI

MONAI 是专为医学影像（CT/MRI）打造的 PyTorch 库。下面展示一个端到端的医学图像分割任务示例，包含数据加载、预处理、3D UNet 模型构建和训练流程。

首先安装依赖：

pip install monai nibabel

代码实现如下：

from monai.transforms import (
    LoadImaged, AddChanneld, Spacingd, Orientationd,
    ScaleIntensityRanged, RandCropByPosNegLabeld, Compose, EnsureTyped
)
from monai.networks.nets import UNet
from monai.losses import DiceLoss
from monai.metrics import DiceMetric
from monai.data import Dataset, DataLoader, decollate_batch
from monai.inferers import sliding_window_inference
import torch
import numpy as np
import matplotlib.pyplot as plt

# 1. 数据准备与预处理
data_dir = "./medical_images/"
train_files = [
    {"image": data_dir + "patient1_CT.nii.gz", "label": data_dir + "patient1_mask.nii.gz"},
    {"image": data_dir + "patient2_CT.nii.gz", "label": data_dir + "patient2_mask.nii.gz"}
]

# 医学图像预处理流程
train_transforms = Compose([
    LoadImaged(keys=["image", "label"]),
    AddChanneld(keys=["image", "label"]),
    Spacingd(
        keys=["image", "label"],
        pixdim=(1.0, 1.0, 2.0),
        mode=("bilinear", "nearest")
    ),
    Orientationd(keys=["image", "label"], axcodes="RAS"),
    ScaleIntensityRanged(
        keys=["image"], a_min=-1000, a_max=1000,
        b_min=0.0, b_max=1.0, clip=True
    ),
    RandCropByPosNegLabeld(
        keys=["image", "label"], label_key="label",
        spatial_size=(96, 96, 64), pos=1, neg=1, num_samples=4
    ),
    EnsureTyped(keys=["image", "label"], dtype=torch.float32)
])

train_ds = Dataset(data=train_files, transform=train_transforms)
train_loader = DataLoader(train_ds, batch_size=2, shuffle=True)

# 2. 构建 3D 分割模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = UNet(
    spatial_dims=3, in_channels=1, out_channels=2,
    channels=(16, 32, 64, 128, 256), strides=(2, 2, 2, 2), num_res_units=2
).to(device)

# 3. 设置训练参数
loss_function = DiceLoss(to_onehot_y=True, softmax=True)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
dice_metric = DiceMetric(include_background=False, reduction="mean")

# 4. 训练循环
max_epochs = 50
for epoch in range(max_epochs):
    model.train()
    epoch_loss = 0
    for batch_data in train_loader:
        inputs, labels = batch_data["image"].to(device), batch_data["label"].to(device)
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = loss_function(outputs, labels)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    epoch_loss /= len(train_loader)

    # 验证步骤
    model.eval()
    with torch.no_grad():
        metric_sum = 0.0
        for val_data in train_loader:
            val_images, val_labels = val_data["image"].to(device), val_data["label"].to(device)
            val_outputs = sliding_window_inference(val_images, (96, 96, 64), 4, model)
            val_outputs = [torch.argmax(i, dim=1) for i in decollate_batch(val_outputs)]
            dice_metric(y_pred=val_outputs, y=val_labels)
        metric = dice_metric.aggregate().item()
        dice_metric.reset()
    print(f"Epoch {epoch+1}/{max_epochs}, Loss: {epoch_loss:.4f}, Dice: {metric:.4f}")

# 5. 可视化结果
def visualize_slice(image, label, prediction, slice_index=25):
    fig, axes = plt.subplots(1, 3, figsize=(15, 5))
    axes[0].imshow(image[0, 0, :, :, slice_index], cmap="gray")
    axes[0].set_title("Input Image")
    axes[0].axis("off")
    axes[1].imshow(label[0, 0, :, :, slice_index], cmap="jet")
    axes[1].set_title("Ground Truth")
    axes[1].axis("off")
    axes[2].imshow(prediction[0, 0, :, :, slice_index], cmap="jet")
    axes[2].set_title("Prediction")
    axes[2].axis("off")
    plt.show()

test_data = train_ds[0]
image = test_data["image"].unsqueeze(0).to(device)
with torch.no_grad():
    prediction = sliding_window_inference(image, (96, 96, 64), 4, model)
    prediction = torch.argmax(prediction, dim=1, keepdim=True)
    visualize_slice(image.cpu().numpy(), test_data["label"].unsqueeze(0).numpy(), prediction.cpu().numpy())

关键组件说明：

• 数据预处理：Spacingd 标准化分辨率差异，Orientationd 统一方向（RAS），ScaleIntensityRanged 处理 CT 值范围。
• 3D UNet 架构：专门处理体积数据，残差单元提升梯度流动。
• 特定技术：Dice 损失函数处理类别不平衡，Sliding Window 推理处理大尺寸体积数据。

进阶优化建议：

# 添加数据增强
from monai.transforms import RandRotated, RandFlipd, RandZoomd
train_transforms.insert(6, Compose([
    RandRotated(keys=["image", "label"], range_x=0.3, prob=0.5),
    RandFlipd(keys=["image", "label"], spatial_axis=0, prob=0.5),
    RandZoomd(keys=["image", "label"], min_zoom=0.9, max_zoom=1.1, prob=0.5)
]))

# 使用更先进的模型
from monai.networks.nets import SwinUNETR
model = SwinUNETR(img_size=(96, 96, 64), in_channels=1, out_channels=2, feature_size=48).to(device)

# 混合精度训练
scaler = torch.cuda.amp.GradScaler()

此示例展示了 MONAI 在医学图像分析中的核心优势：领域特定的预处理、优化的 3D 网络架构以及高效的大体积数据处理能力。

电子病历数据分析：Pandas + Polars + PyGWalker

对于表格型 EHR 数据，结合 Polars 的高效处理和 PyGWalker 的交互可视化能大幅提升探索效率。

安装命令：

pip install pandas polars pygwalker

示例代码：

import pandas as pd
import polars as pl
import pygwalker as pyg
from datetime import datetime

# 示例 EHR 数据结构
data = {
    "patient_id": [1001, 1002, 1003, 1001, 1004],
    "visit_date": ["2023-01-15", "2023-02-20", "2023-01-05", "2023-03-10", "2023-02-28"],
    "diagnosis": ["Hypertension", "Diabetes", "Hypertension", "Asthma", "Diabetes"],
    "medication": ["Lisinopril", "Metformin", "Amlodipine", "Albuterol", "Insulin"],
    "age": [45, 62, 58, 36, 70],
    "blood_pressure": ["140/90", "130/85", "150/95", "120/80", "145/88"],
    "lab_result": [None, 6.5, 7.1, None, 8.0]
}

# 方案 1: 使用 Pandas 加载数据
df_pd = pd.DataFrame(data)
df_pd["visit_date"] = pd.to_datetime(df_pd["visit_date"])

# 方案 2: 使用 Polars 加载大数据（更高效）
df_pl = pl.DataFrame(data).with_columns(pl.col("visit_date").str.to_date("%Y-%m-%d"))

# 转换为 Pandas 供 PyGWalker 使用
df_processed = df_pl.to_pandas()

# 使用 PyGWalker 进行交互式分析
walker = pyg.walk(
    df_processed,
    spec="./ehr_analysis.json",
    dark="light",
    show_cloud_tool=False,
    field_specs={
        "age": {"analyticType": "dimension"},
        "lab_result": {"analyticType": "measure"}
    }
)

关键技巧：

• 小数据：直接使用 pd.read_csv。
• 大数据：先用 Polars 加速处理，再转 Pandas 用于可视化。
• 交互分析：拖拽字段到 X/Y 轴自动生成图表，如患者年龄分布直方图或疾病类型环形图。
• 高级查询：可在界面中使用 Pandas 语法进行复杂过滤，例如提取收缩压并计算高血压标志位。

典型工作流：

1. 数据加载：Polars 处理百万级记录（比 Pandas 快 5-10 倍）。
2. 预处理：在 Polars 中完成清洗转换。
3. 交互探索：拖拽字段自动切换箱线图或多维度对比。

药物发现与分子模拟：RDKit

RDKit 用于化合物表示、分子指纹计算及 QSAR 模型构建。

安装命令：

conda install -c conda-forge rdkit matplotlib

示例代码：

from rdkit import Chem
from rdkit.Chem import AllChem, DataStructs
from rdkit.Chem.Draw import MolsToGridImage
import matplotlib.pyplot as plt

# 1. 分子创建与可视化
ethanol = Chem.MolFromSmiles("CCO")  # 乙醇
ethylamine = Chem.MolFromSmiles("CCN")  # 乙胺

img = MolsToGridImage([ethanol, ethylamine], legends=["Ethanol (CCO)", "Ethylamine (CCN)"], subImgSize=(300, 300))
plt.imshow(img)
plt.axis('off')
plt.show()

# 2. 分子指纹生成
fp_ethanol = AllChem.GetMorganFingerprintAsBitVect(ethanol, radius=2, nBits=2048)
fp_ethylamine = AllChem.GetMorganFingerprintAsBitVect(ethylamine, radius=2, nBits=2048)

# 3. 相似度计算
similarity = DataStructs.TanimotoSimilarity(fp_ethanol, fp_ethylamine)

# 4. 结果输出
print("=" * 50)
print(f"Tanimoto Similarity: {similarity:.4f}")
print("=" * 50)
print("Interpretation:")
print(f"- Ethanol and Ethylamine are {similarity*100:.1f}% similar")
print("- Similarity range: 0.0 (different) to 1.0 (identical)")
print("- Value >0.5 suggests significant structural similarity")

代码解析：

• 分子创建：MolFromSmiles() 将 SMILES 字符串转换为分子对象。
• 分子指纹：使用 Morgan 指纹算法，考虑原子周围两键范围内的结构特征。
• 相似度计算：Tanimoto 系数衡量共有特征比例，取值 0.0 到 1.0。
• 结果解释：相似度约 30% 表示有相同骨架但末端基团不同。在药物发现中，0.3-0.7 通常视为中等相似。

应用场景：

• 虚拟筛选：快速筛选大型化合物数据库。
• 先导化合物优化：评估结构修饰后的相似性变化。
• 聚类分析：对化合物库进行结构分组。

注意事项：相似度值高度依赖指纹参数，结构相似不等于活性相似，需结合生物实验验证。该工具对小分子效果最佳（分子量 < 800 Da）。