正文
获取原文件references/examples.md
The starter's invented-data CPU path and the public synthetic MIMIC-III
set_task -> split -> Transformer -> Trainer path were executed. Restricted MIMIC-IV,
full SleepEDF and specialized molecular-model experiments below are illustrative,
with constructor/schema contracts checked against released source.
1. A runnable plumbing check
From the skill root:
uv run --no-project --isolated --python 3.12 --with pyhealth==2.0.2 python assets/starter_pipeline.py --demo --epochs 1
The asset requires an explicit --demo or --root choice. Demo records are invented.
For real MIMIC-III, use an authorized local root, e.g. --root ./data/mimic-iii;
optional --dev limits patient processing. It checks for both binary classes in
train/validation/test before AUC evaluation. The public synthetic bucket with dev=True (up to 1000 patients) yielded
20 eligible next-visit samples (18 negative, 2 positive), too few to support three
partitions with both classes. Use the accuracy-only smoke example in SKILL.md to
check that download; do not interpret its scores clinically.
2. Fit processors only on training patients
For a modest dataset, gather raw task dictionaries, partition patient IDs, then
create processed datasets. This example assumes an already configured base and
task and was exercised with small synthetic Patient fixtures. For large cohorts,
implement the same policy with streaming/cache-backed preparation.
import random
from pyhealth.datasets import create_sample_dataset
raw = [sample for patient in base.iter_patients() for sample in task(patient)]
ids = sorted({sample["patient_id"] for sample in raw})
random.Random(42).shuffle(ids)
n_train, n_val = int(0.6 * len(ids)), int(0.2 * len(ids))
groups = [set(ids[:n_train]), set(ids[n_train:n_train + n_val]),
set(ids[n_train + n_val:])]
raw_parts = [[s for s in raw if s["patient_id"] in group] for group in groups]
if any(not part for part in raw_parts):
raise ValueError("Need a nonempty cohort in each partition")
train = create_sample_dataset(raw_parts[0], input_schema=task.input_schema,
output_schema=task.output_schema)
val, test = [create_sample_dataset(
part, input_schema=task.input_schema, output_schema=task.output_schema,
input_processors=train.input_processors, output_processors=train.output_processors,
) for part in raw_parts[1:]]
Check class counts and unknown-code rates; random patient splitting does not stratify outcomes or enforce calendar time. Define the target label vocabulary in advance, especially for multilabel tasks: silently dropping unseen test labels would overstate performance. Keep fitted processors with the checkpoint.
3. MIMIC-IV readmission adaptation
from datetime import timedelta
from pyhealth.datasets import MIMIC4EHRDataset
from pyhealth.tasks import ReadmissionPredictionMIMIC4
from pyhealth.models import RETAIN
base = MIMIC4EHRDataset(
root="./data/mimic-iv", # parent of hosp/ and icu/
tables=["diagnoses_icd", "procedures_icd", "prescriptions"],
cache_dir="./cache/mimic4",
)
task = ReadmissionPredictionMIMIC4(window=timedelta(days=30))
# Apply the patient/processor partition recipe above, then:
model = RETAIN(dataset=train, embedding_dim=32)
Use binary metrics with Trainer(metrics=["pr_auc", "roc_auc"]). The upstream task
uses the next recorded admission, excludes the final admission and checks a strict
less-than threshold. Resolve follow-up completeness and transfers before calling
this a 30-day post-discharge readmission study. Attention is an inspection aid.
4. Medication recommendation
from pyhealth.tasks import DrugRecommendationMIMIC3
from pyhealth.models import SafeDrug
task = DrugRecommendationMIMIC3()
# Use a MIMIC3 base containing diagnoses_icd/procedures_icd/prescriptions;
# apply the raw-patient partition and training-processor recipe above.
model = SafeDrug(dataset=train, embedding_dim=32, hidden_dim=32)
Built-in MIMIC drug tasks map NDC to ATC level 3 and create nested history features;
current-visit medications are excluded from drugs_hist. Verify this on actual
samples, check mapped label coverage and resource availability. Use explicit
multilabel metrics such as pr_auc_samples and jaccard_samples, and report DDI
behavior separately from predictive accuracy. A safe-sounding model name is not a
prescribing guarantee. See models for specialized-model caveats.
5. Length of stay
from pyhealth.tasks import LengthOfStayPredictionMIMIC3
from pyhealth.models import RNN
task = LengthOfStayPredictionMIMIC3()
# The base needs prescriptions as well as diagnoses_icd/procedures_icd.
# Apply the partition/processor recipe, then:
model = RNN(dataset=train, rnn_type="GRU", embedding_dim=32, hidden_dim=64)
Use Trainer(metrics=["accuracy", "f1_macro", "cohen_kappa"]) and a matching
monitor. The built-in task is whole-visit code-based, ten-class LOS classification;
filtering features to a declared prediction time is necessary for prospective claims.
6. Sleep staging
from pyhealth.datasets import SleepEDFDataset
from pyhealth.tasks import SleepStagingSleepEDF
from pyhealth.models import SparcNet
base = SleepEDFDataset(root="./data/sleep-edf", subset="cassette")
task = SleepStagingSleepEDF(chunk_duration=30.0)
# Group recordings from one subject together before creating train/val/test.
# After preparing compatible signal tensors and processors:
model = SparcNet(dataset=train)
The root contains SC-subjects.xls and sleep-cassette/ (or ST-subjects.xls
and sleep-telemetry/). Initial metadata preparation uses pandas Excel loading
and can require xlrd for the .xls workbook. Alternatively provide the correctly
prepared sleepedf-cassette-pyhealth.csv/sleepedf-telemetry-pyhealth.csv.
No cache_dir= argument on this dataset wrapper in 2.0.2. Verify person IDs across
nights; an epoch/recording split is not necessarily a subject split. The built-in
task retains W/N1/N2/N3/N4/REM as six classes; decide explicitly whether to merge N3/N4.
Check MNE annotation availability and channel layout before training. Suitable
multiclass metric names include accuracy, f1_macro, cohen_kappa.
7. Baselines and checkpoints
LogisticRegression(dataset=train) is a PyHealth embedding-plus-linear-head model.
Also consider a separately specified fixed-feature baseline when that is the intended
comparison. Low logistic-model PR-AUC does not prove deeper models cannot help;
compare with prevalence and inspect data quality/label definitions.
After training, reuse the same architecture and processor vocabularies:
trainer.load_ckpt("./output/experiment/best.ckpt")
y_true, y_prob, mean_loss = trainer.inference(test_loader)
The checkpoint path is output_path/exp_name/best.ckpt when logging is enabled.
It contains weights, not a complete preprocessing pipeline. Medical-code-only
workflows and tested tokenizer examples are in medical codes.