alterlab-pyhealth

PyHealth：医疗 AI 工具包

概述

PyHealth 是一个面向医疗 AI 的 Python 库，提供用于临床机器学习的数据集、任务定义、模型、训练器和医学编码工具。开发医疗预测模型、处理临床数据、使用医学编码系统，或在任何临床应用之前验证模型时，请使用此技能。

版本注意事项（请先阅读）。 此技能面向 PyHealth 2.x（当前版本为 2.0.2，2026 年 9 月）。2.0 版本的重写带来了 API 变更，许多教程和 2025 年之前的代码片段都会在这些地方出错：

• 任务是需要实例化的类，例如 MortalityPredictionMIMIC4()、DrugRecommendationMIMIC3()——而不是旧的蛇形命名法 mortality_prediction_mimic4_fn 函数。将实例传给 dataset.set_task(task)。
• 数据集需要显式指定表列表。 单数据源加载器使用 root= + tables=[...]（MIMIC3Dataset、MIMIC4EHRDataset、eICUDataset、OMOPDataset）；多模态 MIMIC4Dataset 使用 ehr_root= + ehr_tables=[...]（还可选用 note_root/cxr_root）。
• 模型只接受 SampleDataset 和超参数，例如 Transformer(dataset=samples, embedding_dim=128)。特征键、标签键和模式均从任务的 input_schema / output_schema 读取；1.x 中的 feature_keys= / label_key= / mode= 参数会引发 TypeError。
• 指标名称不带 _score 后缀：pr_auc、roc_auc、f1；多标签／药物推荐使用 *_samples 系列（jaccard_samples、f1_samples、pr_auc_samples）。将 metrics=[...] 传给 Trainer 构造函数，并将 monitor= 设为其中一个名称。
• 检查点使用 trainer.save_ckpt(path) / trainer.load_ckpt(path)（没有 trainer.save）。
• 2.0.2 要求 Python 3.12 或 3.13，并固定使用自身的依赖栈（numpy 2.2、pandas 2.3、torch 2.7、transformers 4.53），因此请将其安装在专用环境中，不要与 pandas 3 / transformers 5 安装在一起。

不确定类名或参数名时，请查看已安装的源码，不要轻信较旧的代码片段。

何时使用此技能

在以下情形中调用此技能：

• 使用医疗数据集：MIMIC-III、MIMIC-IV、eICU、OMOP、睡眠 EEG 数据、医学图像
• 临床预测任务：死亡率预测、再入院预测、住院时长预测、药物推荐
• 医学编码：在 ICD-9/10、NDC、RxNorm、ATC、CCS 编码系统之间转换
• 处理临床数据：序列事件、生理信号、临床文本、医学图像
• 实现医疗模型：RETAIN、SafeDrug、GAMENet、StageNet、用于 EHR 的 Transformer
• 评估临床模型：公平性指标、校准、可解释性、不确定性量化

不适用情形

情形	改用
清理原始 ECG/EEG/EDA 信号并计算 HRV 或 SCR 特征（不训练模型）	alterlab-neurokit2
读取、匿名化或转换 DICOM 图像文件	alterlab-pydicom
在表格型临床数据集上进行 Kaplan-Meier / Cox 生存分析建模	alterlab-scikit-survival
按生物标志物分层的队列报告，并提供按 GRADE 分级的治疗建议	alterlab-clinical-decision
对非 EHR 数据进行通用表格型机器学习（scikit-learn 管道）	alterlab-scikit-learn

核心功能

PyHealth 通过模块化的 5 阶段管道运行：

1. 数据加载：用于 EHR、信号、影像和文本数据集的标准化加载器
2. 任务定义：预定义的临床预测任务（任务类）或自定义 BaseTask 子类
3. 模型选择：基线模型、通用深度学习模型和医疗领域专用模型
4. 训练：Trainer 支持最佳检查点选择、监控和评估
5. 验证：校准、保形预测、公平性指标和可解释性方法

PyHealth 2.x 使用基于 polars 的数据层，并缓存任务样本，从而提高大型 EHR 表的内存使用效率。

快速入门工作流

from pyhealth.datasets import MIMIC4EHRDataset, split_by_patient, get_dataloader
from pyhealth.tasks import MortalityPredictionMIMIC4
from pyhealth.models import Transformer
from pyhealth.trainer import Trainer

# 1. Load dataset (declare the tables the task needs) and set the task (a class instance)
dataset = MIMIC4EHRDataset(
    root="/path/to/mimic-iv/2.2",
    tables=["diagnoses_icd", "procedures_icd", "prescriptions"],
)
sample_dataset = dataset.set_task(MortalityPredictionMIMIC4())

# 2. Split data by patient (no leakage across splits)
train, val, test = split_by_patient(sample_dataset, [0.7, 0.1, 0.2], seed=42)

# 3. Create data loaders
train_loader = get_dataloader(train, batch_size=64, shuffle=True)
val_loader = get_dataloader(val, batch_size=64, shuffle=False)
test_loader = get_dataloader(test, batch_size=64, shuffle=False)

# 4. Initialize the model: inputs, label ("mortality"), and mode ("binary")
#    all come from the task schema, so only hyperparameters are passed
model = Transformer(dataset=sample_dataset, embedding_dim=128)

trainer = Trainer(model=model, metrics=["pr_auc", "roc_auc", "f1"])  # device auto-detected
trainer.train(
    train_dataloader=train_loader,
    val_dataloader=val_loader,
    epochs=50,
    monitor="pr_auc",            # AUPRC — robust for the rare-mortality class
    monitor_criterion="max",
)

# 5. Evaluate (uses the metrics passed to the Trainer)
results = trainer.evaluate(test_loader)

详细文档

请阅读与你当前步骤对应的参考文件：

文件	阅读时机	主要主题
references/datasets.md	加载 MIMIC/eICU/OMOP/信号数据集、拆分数据时	Patient/Event 结构、加载器、split_by_patient / split_by_visit / split_by_sample
references/medical_coding.md	转换或分组 ICD、NDC、RxNorm、ATC、CCS 代码时	InnerMap 查找与层级结构、CrossMap 转换
references/tasks.md	选择预定义任务或编写自定义任务时	2.x 任务类、input_schema / output_schema、自定义 BaseTask
references/models.md	选择和配置模型时	基线模型、RNN/CNN/Transformer、RETAIN、SafeDrug、GAMENet、StageNet、GAT/GCN
references/preprocessing.md	了解原始事件如何转换为张量时	Schema 字符串处理器（"sequence"、"timeseries"、"binary"、...）
references/training_evaluation.md	训练、指标、校准、不确定性、可解释性相关内容	Trainer、指标字符串、保形预测、Chefer/IG 归因

安装

uv venv --python 3.13 .venv-pyhealth      # PyHealth 2.0.2 supports Python 3.12-3.13
source .venv-pyhealth/bin/activate
uv pip install "pyhealth>=2.0.2"

要求（PyHealth 2.0.2）：

• Python 3.12 或 3.13（>=3.12,<3.14）——如果默认解释器为 3.14，请使用 --python 3.13 创建环境。
• PyTorch、polars、pandas、scikit-learn 和 transformers 已作为固定版本依赖安装——请在独立环境中使用 PyHealth，以免这些固定版本与其他项目发生冲突。
• PyPI 上有 2.1 alpha 版本系列；除非需要仅在 alpha 版本中提供的功能，否则请使用 2.0.x 版本。

常见用例

用例 1：ICU 死亡率预测

目标：预测患者在重症监护病房中的死亡率

方法：

1. 加载 MIMIC-IV 数据集 → 阅读 references/datasets.md
2. 应用死亡率预测任务 → 阅读 references/tasks.md
3. 选择可解释模型（RETAIN）或便于归因的模型（Transformer）→ 阅读 references/models.md
4. 训练并评估 → 阅读 references/training_evaluation.md
5. 解读预测结果，供临床审核 → 阅读 references/training_evaluation.md

用例 2：安全用药推荐

目标：推荐药物，同时避免药物相互作用

方法：

1. 加载 EHR 数据集（MIMIC-III/IV、eICU 或 OMOP）→ 阅读 references/datasets.md
2. 应用 DrugRecommendation* 任务 → 阅读 references/tasks.md
3. 使用 SafeDrug 或 GAMENet，它们会根据数据集构建 DDI 图 → 阅读 references/models.md
4. 预处理药物代码 → 阅读 references/medical_coding.md
5. 使用多标签指标（jaccard_samples、f1_samples、pr_auc_samples）进行评估 → 阅读 references/training_evaluation.md

用例 3：医院再入院预测

目标：识别有再入院风险的患者

方法：

1. 加载多站点 EHR 数据（eICU 或 OMOP）→ 阅读 references/datasets.md
2. 应用 ReadmissionPrediction* 任务 → 阅读 references/tasks.md
3. 处理类别不平衡问题（报告 AUPRC，而不只是 AUROC）→ 阅读 references/training_evaluation.md
4. 训练 Transformer 或 RNN 模型 → 阅读 references/models.md
5. 校准预测结果并评估公平性 → 阅读 references/training_evaluation.md

用例 4：睡眠分期

目标：根据 EEG 信号对睡眠阶段进行分类

方法：

1. 加载睡眠 EEG 数据集（SleepEDF、SHHS、ISRUC）→ 阅读 references/datasets.md
2. 应用睡眠分期任务（SleepStagingSleepEDF 或旧版 sleep_staging_*_fn 函数）→ 阅读 references/tasks.md
3. 预处理 EEG 信号（滤波、分段）→ 阅读 references/preprocessing.md
4. 训练 CNN、SparcNet 或 ContraWR 模型 → 阅读 references/models.md
5. 评估各阶段的性能（f1_macro、cohen_kappa）→ 阅读 references/training_evaluation.md

用例 5：医学编码转换

目标：统一不同编码系统中的诊断

方法：

1. 阅读 references/medical_coding.md，了解全面的指导
2. 使用 CrossMap 在 ICD-9、ICD-10 和 CCS 之间转换
3. 将编码归入具有临床意义的类别
4. 集成到数据集处理中

用例 6：临床文本转 ICD 编码

目标：从临床笔记中自动分配 ICD 编码

方法：

1. 加载包含临床笔记的 MIMIC-III → 阅读 references/datasets.md
2. 应用 MIMIC3ICD9Coding 任务 → 阅读 references/tasks.md
3. 预处理临床文本 → 阅读 references/preprocessing.md
4. 使用 TransformersModel(dataset=..., model_name="emilyalsentzer/Bio_ClinicalBERT") → 阅读 references/models.md
5. 使用多标签指标进行评估 → 阅读 references/training_evaluation.md

最佳实践

数据处理

1. 始终按患者划分：确保每位患者只出现在一个数据划分中，以防止数据泄漏

   from pyhealth.datasets import split_by_patient
   train, val, test = split_by_patient(sample_dataset, [0.7, 0.1, 0.2], seed=42)
   

2. 检查数据集统计信息：在建模前了解数据

   dataset.stats()  # prints patient and event counts (returns None)
   

3. 使用适当的预处理方法：根据数据类型选择处理器（参见 references/preprocessing.md）

模型开发

1. 从基线开始：使用简单模型建立基线性能

   • 二分类／多分类任务使用 LogisticRegression
   • 使用 MLP 建立初始深度学习基线

2. 选择适合任务的模型：

   • 需要可解释性 → RETAIN、AdaCare（设计上具备）；Transformer（事后归因）
   • 药物推荐 → SafeDrug、GAMENet
   • 长序列 → Transformer
   • 图关系 → GAT / GCN

3. 监控验证指标：根据任务使用适当的指标，并处理类别不平衡。PyHealth 指标字符串（传入 Trainer(metrics=[...]) / monitor=）：

   • 二分类：roc_auc、pr_auc（对于罕见事件，优先使用 pr_auc）、f1、accuracy
   • 多分类：f1_macro、f1_weighted、accuracy、cohen_kappa
   • 多标签／药物推荐：jaccard_samples、f1_samples、pr_auc_samples、ddi（报告为 ddi_score）
   • 回归：mae、mse、kl_divergence

临床验证

1. 校准预测结果：确保概率可靠（参见 references/training_evaluation.md）
2. 评估公平性：对不同人口统计群体进行评估，以检测偏差
3. 量化不确定性：为预测提供置信度估计（共形预测集）
4. 解释预测结果：使用注意力／相关性图、SHAP 或积分梯度供临床医生审查
5. 全面验证：使用来自不同时期或不同机构的留出测试集
6. 透明报告：发布临床预测模型时，遵循 TRIPOD+AI（BMJ 2024;385:e078378）

局限性与注意事项

数据要求

• 大型数据集：深度学习模型需要充足的数据（数千名患者）
• 数据质量：缺失数据和编码错误会影响性能
• 时间一致性：必要时，确保训练集／测试集划分遵循时间顺序
• 访问权限：MIMIC 和 eICU 需要通过 PhysioNet 资质认证并签署数据使用协议；切勿将受限记录复制到协议未涵盖的提示词、笔记本或代码库中

临床验证

• 外部验证：在来自不同医院／系统的数据上进行测试
• 前瞻性评估：在部署前于真实临床环境中进行验证
• 临床审查：请临床医生审查预测结果及其解释
• 用于决策支持，而非诊断：将模型输出作为供具备资质的临床医生参考的研究级风险估计；作为医疗器械部署时须遵守器械法规（例如 FDA SaMD、EU MDR）
• 伦理考量：处理隐私（HIPAA/GDPR）、公平性和安全性问题

计算资源

• 推荐使用 GPU：高效训练深度学习模型
• 内存要求：大型数据集可能需要 16GB 以上 RAM
• 存储空间：医疗保健数据集可能占用几十到几百 GB，另需预留任务样本缓存空间

故障排查

常见问题

TypeError: ... unexpected keyword argument 'feature_keys'（或 'root'）：

• 你使用的是 1.x 风格的调用。向模型仅传入 dataset= 和超参数；使用 MIMIC4EHRDataset(root=..., tables=...) 或 MIMIC4Dataset(ehr_root=..., ehr_tables=...)

ImportError 或缺少表：

• 确保已下载数据集文件，且根路径指向带版本号的文件夹
• 确认表名存在于数据集的 YAML 配置中

内存不足：

• 减小批次大小
• 缩短序列长度（Transformer 的 max_seq_len）
• 向数据集加载器传入 dev=True（例如 MIMIC4EHRDataset(..., dev=True)），以便使用前 1,000 名患者的数据进行原型验证
• 分块处理数据

性能不佳：

• 检查类别不平衡情况，并使用合适的指标（pr_auc 与 roc_auc）
• 检查预处理（归一化、缺失数据处理）
• 提高模型容量或增加训练轮数
• 检查训练集/测试集划分中是否存在数据泄漏

训练速度慢：

• 使用 GPU（Trainer(..., device="cuda")）
• 增大批次大小（如果内存允许）
• 缩短序列长度
• 使用更轻量的模型（用 CNN 或 RNN 替代 Transformer）

获取帮助

• 文档：https://pyhealth.readthedocs.io/
• GitHub 问题：https://github.com/sunlabuiuc/PyHealth/issues
• 示例/笔记本：https://github.com/sunlabuiuc/PyHealth/tree/master/examples

示例：完整工作流

# Complete mortality prediction pipeline (PyHealth 2.0.x)
import torch
from pyhealth.datasets import MIMIC4EHRDataset, split_by_patient, get_dataloader
from pyhealth.tasks import MortalityPredictionMIMIC4
from pyhealth.models import Transformer
from pyhealth.trainer import Trainer
from pyhealth.interpret.methods import CheferRelevance

# 1. Load dataset (declare the tables the task needs)
dataset = MIMIC4EHRDataset(
    root="/data/mimic-iv/2.2",
    tables=["diagnoses_icd", "procedures_icd", "prescriptions"],
)
dataset.stats()

# 2. Define task (instantiate the task class)
sample_dataset = dataset.set_task(MortalityPredictionMIMIC4())
print(f"Generated {len(sample_dataset)} samples")

# 3. Split data (by patient to prevent leakage)
train_ds, val_ds, test_ds = split_by_patient(sample_dataset, [0.7, 0.1, 0.2], seed=42)

# 4. Create data loaders
train_loader = get_dataloader(train_ds, batch_size=64, shuffle=True)
val_loader = get_dataloader(val_ds, batch_size=64)
test_loader = get_dataloader(test_ds, batch_size=64)

# 5. Initialize the model (schema-driven; swap in RETAIN(dataset=sample_dataset,
#    embedding_dim=128) for a model that is interpretable by design)
model = Transformer(dataset=sample_dataset, embedding_dim=128, heads=2, num_layers=2)

# 6. Train, keeping the best checkpoint by validation AUPRC
trainer = Trainer(model=model, metrics=["accuracy", "pr_auc", "roc_auc", "f1"])
trainer.train(
    train_dataloader=train_loader,
    val_dataloader=val_loader,
    epochs=50,
    optimizer_class=torch.optim.Adam,
    optimizer_params={"lr": 1e-3},
    weight_decay=1e-5,
    monitor="pr_auc",          # AUPRC for the imbalanced (rare-mortality) outcome
    monitor_criterion="max",
    patience=5,                # early stopping
)

# 7. Evaluate on the test set (uses the metrics passed to the Trainer)
for metric, value in trainer.evaluate(test_loader).items():
    print(f"  {metric}: {value:.4f}")

# 8. Predictions with patient IDs: inference() returns (y_true, y_prob, loss),
#    extended with patient_ids when return_patient_ids=True
y_true, y_prob, loss, patient_ids = trainer.inference(test_loader, return_patient_ids=True)
positive_prob = y_prob if y_prob.ndim == 1 else y_prob[..., -1]
high_risk_idx = int(positive_prob.argmax())
print(f"Highest-risk patient: {patient_ids[high_risk_idx]} ({float(positive_prob[high_risk_idx]):.3f})")

# 9. Token-level relevance (Chefer; supported by Transformer and StageAttentionNet)
relevance = CheferRelevance(model)
batch = next(iter(get_dataloader(test_ds, batch_size=1, shuffle=False)))
for feature_key, rel in relevance.attribute(**batch).items():
    print(f"{feature_key}: top tokens -> {rel[0].topk(min(5, rel.shape[-1])).indices.tolist()}")

# 10. Save the trained weights
trainer.save_ckpt("./models/mortality_transformer.pt")

资源

如需了解各组件的详细信息，请参阅 references/ 中的参考文件：datasets.md、medical_coding.md、tasks.md、models.md、preprocessing.md 和 training_evaluation.md（关于何时阅读各文件，请参阅 详细文档 下方的表格）。

属于 AlterLab Academic Skills 套件。