pyhealth

PyHealth

概述

PyHealth 为基于 EHR 数据的医疗机器学习提供端到端工作流：数据加载 → 医疗编码处理 → 患者级数据集构建 → 模型训练 → 评估。它原生支持 MIMIC-III、MIMIC-IV、eICU-CRD 和 OMOP-CDM 结构化数据库，并可处理各类数据库特有的数据格式。医疗编码（ICD-9、ICD-10、ATC、NDC、SNOMED）按层级编码系统组织，支持编码级嵌入和跨本体映射。只需几行代码即可实例化预构建任务，包括死亡率预测、药物推荐、再入院预测、住院时长预测和诊断编码预测。自定义任务遵循标准化接口。

适用场景

• 使用 MIMIC-III 或 MIMIC-IV 训练临床结局预测模型（死亡率、再入院、住院时长）
• 利用 ATC 编码层级构建药物推荐或药物相互作用预测模型
• 对医疗机构 EHR 系统中的 OMOP-CDM 格式数据进行机器学习处理
• 在医疗基准测试中使用预训练临床模型（RETAIN、GRASP、MedBERT）作为基线
• 构建带有时间结构的患者就诊序列，用于 RNN/Transformer 模型
• 使用合适的指标（AUROC、AUPRC、F1、Jaccard）评估临床预测模型
• 如需纯 EHR 预处理而不涉及机器学习，请使用 FIDDLE；如需临床笔记 NLP，请使用 clinical-longformer

前置条件

• Python 包：pyhealth、torch、pandas、scikit-learn
• 数据要求：MIMIC-III/IV CSV 文件（需要通过 PhysioNet 认证）、eICU 或 OMOP-CDM 数据库
• MIMIC 访问权限：在 physionet.org 申请（免费；需要完成 CITI 培训，约需 1 周）

pip install pyhealth torch pandas scikit-learn
# Download MIMIC-III: https://physionet.org/content/mimiciii/
# Download MIMIC-IV: https://physionet.org/content/mimiciv/

快速入门

from pyhealth.datasets import MIMIC3Dataset

# Load MIMIC-III (specify path to downloaded CSV files)
dataset = MIMIC3Dataset(
    root="path/to/mimic-iii/",
    tables=["DIAGNOSES_ICD", "PRESCRIPTIONS", "PROCEDURES_ICD"],
    code_mapping={"ICD9CM": "CCSCM"},  # map ICD-9 codes to CCS multi-level
    dev=True,  # dev=True uses 1% of data for fast testing
)

print(f"Patients: {dataset.stat()['num_patients']}")
print(f"Visits: {dataset.stat()['num_visits']}")

核心 API

模块 1：数据集加载

加载 MIMIC-III、MIMIC-IV、eICU 和 OMOP-CDM 数据集。

from pyhealth.datasets import MIMIC3Dataset, MIMIC4Dataset, eICUDataset

# MIMIC-III
mimic3 = MIMIC3Dataset(
    root="data/mimic-iii/",
    tables=["DIAGNOSES_ICD", "PRESCRIPTIONS", "PROCEDURES_ICD", "LABEVENTS"],
    code_mapping={"ICD9CM": "CCSCM", "NDC": "ATC3"},  # standardize codes
    dev=False,
)
stats = mimic3.stat()
print(f"MIMIC-III: {stats['num_patients']} patients, {stats['num_visits']} visits")

# MIMIC-IV
mimic4 = MIMIC4Dataset(
    root="data/mimic-iv/",
    tables=["diagnoses_icd", "prescriptions", "procedures_icd"],
    code_mapping={"ICD10CM": "CCSCM"},
    dev=True,
)

# eICU
eicu = eICUDataset(
    root="data/eicu/",
    tables=["diagnosis", "medication", "treatment"],
    dev=True,
)
print(f"eICU loaded: {eicu.stat()}")

# Explore dataset structure
patient_id = list(mimic3.patients.keys())[0]
patient = mimic3.patients[patient_id]
print(f"Patient {patient_id}: {len(patient.visits)} visits")

for visit in patient.visits[:2]:
    print(f"  Visit {visit.visit_id}:")
    print(f"    Diagnoses: {visit.get_code_list('CCSCM')[:5]}")
    print(f"    Medications: {visit.get_code_list('ATC3')[:3]}")

模块 2：任务构建

将原始数据集转换为可用于机器学习的任务数据集。

from pyhealth.tasks import mortality_prediction_mimic3_fn
from pyhealth.datasets import SampleDataset

# Mortality prediction task
# Each sample: patient's visit history → binary mortality label
mortality_dataset = SampleDataset(
    dataset=mimic3,
    task_fn=mortality_prediction_mimic3_fn,
)

print(f"Task: mortality prediction")
print(f"Samples: {len(mortality_dataset)}")

# Inspect a sample
sample = mortality_dataset[0]
print(f"Sample keys: {list(sample.keys())}")
print(f"Conditions (ICD codes): {sample['conditions'][:3]}")
print(f"Drugs (ATC codes): {sample['drugs'][:3]}")
print(f"Label (mortality): {sample['label']}")

# Custom task: 30-day readmission prediction
def readmission_30day_fn(patient):
    """Custom task function: predict 30-day readmission after discharge."""
    samples = []
    for i, visit in enumerate(patient.visits[:-1]):
        next_visit = patient.visits[i + 1]
        # Compute days between discharge and next admission
        days_gap = (next_visit.encounter_time - visit.discharge_time).days
        label = int(days_gap <= 30)

        samples.append({
            "visit_id": visit.visit_id,
            "patient_id": patient.patient_id,
            "conditions": visit.get_code_list("CCSCM"),
            "drugs": visit.get_code_list("ATC3"),
            "procedures": visit.get_code_list("ICD9PROC"),
            "label": label,
        })
    return samples

readmission_dataset = SampleDataset(dataset=mimic3, task_fn=readmission_30day_fn)
print(f"Readmission samples: {len(readmission_dataset)}")
pos_rate = sum(s["label"] for s in readmission_dataset) / len(readmission_dataset)
print(f"Positive rate (30-day readmission): {pos_rate:.2%}")

模块 3：医疗编码系统

使用 ICD、ATC、NDC 和其他层级医疗编码系统。

from pyhealth.medcode import InnerMap

# ICD-9-CM diagnosis codes
icd9 = InnerMap.load("ICD9CM")
code = "428.0"  # Heart failure, unspecified
print(f"Code: {code}")
print(f"Description: {icd9.lookup(code)}")
print(f"Ancestors: {icd9.get_ancestors(code)}")
print(f"Children: {icd9.get_children(code)[:5]}")

# ATC drug classification
atc = InnerMap.load("ATC")
drug_code = "A10BA02"  # Metformin
print(f"\nATC: {drug_code}")
print(f"Drug: {atc.lookup(drug_code)}")
print(f"L1 class: {atc.get_ancestors(drug_code)}")

# Cross-ontology code mapping
from pyhealth.medcode import CrossMap

# Map NDC (drug product codes) to ATC level 3
ndc_to_atc = CrossMap.load("NDC", "ATC3")
ndc_code = "0069-2587-30"  # example NDC
atc3_codes = ndc_to_atc.map(ndc_code)
print(f"NDC {ndc_code} → ATC3: {atc3_codes}")

# Map ICD-9 to ICD-10
icd9_to_icd10 = CrossMap.load("ICD9CM", "ICD10CM")
icd10 = icd9_to_icd10.map("428.0")
print(f"ICD-9 428.0 → ICD-10: {icd10}")

模块 4：模型训练

训练预实现的临床机器学习模型。

from pyhealth.models import Transformer, RETAIN
from pyhealth.datasets import split_by_patient, get_dataloader
import torch

# Train/val/test split (patient-level, no leakage)
train_ds, val_ds, test_ds = split_by_patient(mortality_dataset, [0.7, 0.1, 0.2])

train_loader = get_dataloader(train_ds, batch_size=32, shuffle=True)
val_loader   = get_dataloader(val_ds,   batch_size=64, shuffle=False)
test_loader  = get_dataloader(test_ds,  batch_size=64, shuffle=False)

print(f"Train: {len(train_ds)} | Val: {len(val_ds)} | Test: {len(test_ds)}")

# Transformer model for EHR sequence modeling
model = Transformer(
    dataset=mortality_dataset,
    feature_keys=["conditions", "drugs", "procedures"],
    label_key="label",
    mode="binary",        # binary classification
    embedding_dim=128,
    num_heads=4,
    num_layers=2,
    dropout=0.1,
)

print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

# RETAIN: Reverse Time Attention model (interpretable clinical ML)
retain_model = RETAIN(
    dataset=mortality_dataset,
    feature_keys=["conditions", "drugs"],
    label_key="label",
    mode="binary",
    embedding_dim=64,
)

模块 5：训练与评估

from pyhealth.trainer import Trainer

trainer = Trainer(
    model=model,
    metrics=["pr_auc", "roc_auc", "f1"],  # PR-AUC, ROC-AUC, F1
)

trainer.train(
    train_dataloader=train_loader,
    val_dataloader=val_loader,
    epochs=20,
    optimizer_params={"lr": 1e-3},
    weight_decay=1e-5,
    monitor="pr_auc",           # early stopping metric
    monitor_criterion="max",
    load_best_model_after_train=True,
)

# Evaluate on test set
results = trainer.evaluate(test_loader)
print("Test results:")
for metric, value in results.items():
    print(f"  {metric}: {value:.4f}")

模块 6：药物推荐

根据患者的就诊历史预测其应接受哪些药物。

from pyhealth.tasks import drug_recommendation_mimic3_fn
from pyhealth.models import GAMENet
from pyhealth.datasets import SampleDataset, split_by_patient, get_dataloader

# Drug recommendation task
drug_dataset = SampleDataset(
    dataset=mimic3,
    task_fn=drug_recommendation_mimic3_fn,
)

train_ds, val_ds, test_ds = split_by_patient(drug_dataset, [0.7, 0.1, 0.2])

# GAMENet: graph-augmented memory network for drug recommendation
gamenet = GAMENet(
    dataset=drug_dataset,
    feature_keys=["conditions", "procedures"],
    label_key="drugs",
    mode="multilabel",     # recommend multiple drugs per visit
    embedding_dim=64,
)

train_loader = get_dataloader(train_ds, batch_size=16, shuffle=True)

trainer = Trainer(model=gamenet, metrics=["jaccard", "f1", "prauc"])
trainer.train(train_dataloader=train_loader,
              val_dataloader=get_dataloader(val_ds, 32),
              epochs=30, monitor="jaccard")
print("Drug recommendation model trained")

核心概念

患者-就诊-事件层级

PyHealth 将 EHR 数据组织为 Patient → Visit → 医疗编码。每个 Visit 包含多个表中的带时间戳事件（诊断、药物、操作、检验）。机器学习模型将每位患者视为一系列就诊，并将每次就诊视为一组医疗编码，从而捕捉疾病的时间演变。

编码映射与标准化

原始 EHR 编码（ICD-9、NDC）非常具体且数量众多。PyHealth 的 code_mapping 参数可自动将其转换为更粗粒度的本体（CCSCM 约有 260 个类别，而 ICD-9 编码约有 15,000 个），从而缩小词汇表并支持在不同数据集之间迁移。

常见工作流

工作流 1：完整的死亡率预测流程

from pyhealth.datasets import MIMIC3Dataset, SampleDataset, split_by_patient, get_dataloader
from pyhealth.tasks import mortality_prediction_mimic3_fn
from pyhealth.models import Transformer
from pyhealth.trainer import Trainer

# 1. Load data
dataset = MIMIC3Dataset(
    root="data/mimic-iii/",
    tables=["DIAGNOSES_ICD", "PRESCRIPTIONS", "PROCEDURES_ICD"],
    code_mapping={"ICD9CM": "CCSCM", "NDC": "ATC3"},
    dev=True,
)

# 2. Build task dataset
task_ds = SampleDataset(dataset, task_fn=mortality_prediction_mimic3_fn)
print(f"Samples: {len(task_ds)}, Positive rate: {sum(s['label'] for s in task_ds)/len(task_ds):.2%}")

# 3. Split
train_ds, val_ds, test_ds = split_by_patient(task_ds, [0.7, 0.1, 0.2])

# 4. Model
model = Transformer(
    dataset=task_ds,
    feature_keys=["conditions", "drugs"],
    label_key="label",
    mode="binary",
    embedding_dim=128,
)

# 5. Train
trainer = Trainer(model=model, metrics=["pr_auc", "roc_auc"])
trainer.train(
    train_dataloader=get_dataloader(train_ds, 32, shuffle=True),
    val_dataloader=get_dataloader(val_ds, 64),
    epochs=15,
    monitor="pr_auc",
)

# 6. Evaluate
results = trainer.evaluate(get_dataloader(test_ds, 64))
print(f"Test PR-AUC: {results['pr_auc']:.4f}")
print(f"Test ROC-AUC: {results['roc_auc']:.4f}")

工作流 2：模型对比基准测试

from pyhealth.models import Transformer, RETAIN, MedBERT
from pyhealth.trainer import Trainer
from pyhealth.datasets import get_dataloader
import pandas as pd

models = {
    "Transformer": Transformer(task_ds, ["conditions", "drugs"], "label", "binary"),
    "RETAIN":      RETAIN(task_ds, ["conditions", "drugs"], "label", "binary"),
}

results_list = []
for name, model in models.items():
    trainer = Trainer(model=model, metrics=["pr_auc", "roc_auc", "f1"])
    trainer.train(
        train_dataloader=get_dataloader(train_ds, 32, shuffle=True),
        val_dataloader=get_dataloader(val_ds, 64),
        epochs=10,
        monitor="pr_auc",
    )
    test_results = trainer.evaluate(get_dataloader(test_ds, 64))
    test_results["model"] = name
    results_list.append(test_results)
    print(f"{name}: PR-AUC={test_results['pr_auc']:.4f}, ROC-AUC={test_results['roc_auc']:.4f}")

results_df = pd.DataFrame(results_list).set_index("model")
results_df.to_csv("model_comparison.csv")
print(results_df)

关键参数

参数	模块/类	默认值	范围/选项	作用
tables	MIMIC3Dataset	—	MIMIC 表名列表	要加载的 EHR 表；表越多，特征越丰富，加载越慢
code_mapping	MIMIC3Dataset	{}	{"ICD9CM": "CCSCM"}	将原始代码映射到标准化本体
dev	MIMIC3Dataset	False	True/False	True 使用 1% 的数据进行快速开发
feature_keys	所有模型	—	代码类型字符串列表	选择作为模型输入的医疗代码类型
embedding_dim	Transformer、RETAIN	128	64–512	代码和患者嵌入的隐藏维度
num_heads	Transformer	4	1–16	多头注意力头数（必须能整除 embedding_dim）
num_layers	Transformer	2	1–6	Transformer 编码器层数
dropout	所有模型	0.1	0–0.5	正则化的 dropout 比率
mode	所有模型	—	"binary"、"multiclass"、"multilabel"	预测任务类型
monitor	Trainer.train	"loss"	"pr_auc"、"roc_auc"、"f1"	用于提前停止和模型选择的指标

最佳实践

1. 始终使用 split_by_patient，切勿随机拆分：按就诊记录随机拆分会导致数据泄漏——同一患者可能因不同就诊记录同时出现在训练集和测试集中。PyHealth 的 split_by_patient 可确保严格按患者划分数据。

2. 开发期间使用 dev=True：MIMIC-III 包含约 46,000 名患者。加载完整数据集需要几分钟；开发模式只需几秒即可加载约 460 名患者。仅在最终训练时切换为 dev=False。

3. 应用代码映射，使不同数据集保持标准一致：原始 ICD-9 代码约有 15,000 个唯一值；CCSCM 会将其缩减为约 260 个具有临床意义的类别。这能大幅缩减模型词汇表，并支持有意义的代码嵌入；对小型数据集尤其重要。

4. 除 ROC-AUC 外，还要使用 PR-AUC 进行评估：临床数据集通常存在严重类别不平衡（例如，死亡率为 10%）。对于不平衡任务，PR-AUC 比 ROC-AUC 更有参考价值——预测结果全部为负类的模型，其 ROC-AUC 约为 0.5，而 PR-AUC 会接近正类比例（约 0.1）。

常用示例

示例：导出患者嵌入

import torch
from pyhealth.datasets import get_dataloader
import numpy as np

model.eval()
embeddings = []
labels = []

with torch.no_grad():
    for batch in get_dataloader(test_ds, batch_size=64):
        # Get patient-level representation (before classification head)
        hidden = model.get_patient_representation(batch)
        embeddings.append(hidden.cpu().numpy())
        labels.append(batch["label"].numpy())

embeddings = np.concatenate(embeddings, axis=0)
labels = np.concatenate(labels, axis=0)
np.save("patient_embeddings.npy", embeddings)
np.save("patient_labels.npy", labels)
print(f"Embeddings: {embeddings.shape}")

示例：使用加权采样器处理类别不平衡

import torch
from torch.utils.data import WeightedRandomSampler
from pyhealth.datasets import get_dataloader

# Compute class weights for imbalanced mortality prediction
labels = [sample["label"] for sample in train_ds]
pos = sum(labels)
neg = len(labels) - pos
weights = [1.0/neg if l == 0 else 1.0/pos for l in labels]
sampler = WeightedRandomSampler(weights, num_samples=len(weights), replacement=True)

# Use sampler in dataloader
balanced_loader = torch.utils.data.DataLoader(
    train_ds, batch_size=32, sampler=sampler,
    collate_fn=train_ds.collate_fn
)
print(f"Balanced loader: {len(balanced_loader)} batches")

故障排除

问题	原因	解决方案
加载数据集时出现 FileNotFoundError	root 路径错误或缺少 MIMIC CSV 文件	确认 root 中包含正确的 CSV 文件；使用 ls data/mimic-iii/*.csv 查看文件列表
feature_keys 中的代码类型出现 KeyError	未加载该代码类型或映射失败	确保 tables 包含所需的表，并且 code_mapping 映射到预期的本体
任务数据集中的所有样本都具有 label=0	任务函数逻辑错误或数据问题	对单个患者打印 task_fn(patient)；检查标签推导逻辑
加载完整 MIMIC 时出现内存错误	数据集过大；RAM 不足	开发期间使用 dev=True；分块加载或缩减 tables 列表
测试集上的 ROC-AUC 约为 0.5	严重的类别不平衡导致预测器退化	使用 WeightedRandomSampler；改为报告 PR-AUC；降低分类阈值
训练期间出现 CUDA 显存不足	批次大小过大	将 batch_size 从 32 降至 8 或 16；使用梯度累积
代码映射返回空列表	在交叉映射中找不到代码	检查代码格式（例如，ICD-9 代码是否带小数点）；尝试 InnerMap.load("ICD9CM").lookup(code)

相关技能

• statsmodels-statistical-modeling — 用于临床结局预测的逻辑回归基线
• scikit-learn-machine-learning — 基于 PyHealth 特征的传统机器学习基线（随机森林、梯度提升）
• clinical-decision-support-documents — 将临床机器学习模型输出转化为决策支持工具

参考资料

• PyHealth 文档 — API 参考、教程和任务说明
• PyHealth 论文：Zhao 等人（2021），arXiv — 库的设计和基准任务
• MIMIC-III 论文：Johnson 等人（2016），Nature Scientific Data — 数据集介绍
• MIMIC-IV 论文：Johnson 等人（2023），Nature Scientific Data — 更新后的数据集
• RETAIN 论文：Choi 等人（2016），NeurIPS — 可解释的临床预测模型
• PyHealth GitHub — 源代码和示例