深入解析与应用：Delphi-2M 健康轨迹预测模型的开源实践与研究（下）

第三章：开发者实践指南：Delphi-2M 的编程应用

本章将深入代码层面，提供一个从零到一的实践路径。

3.1 环境准备与依赖安装

# 1. 克隆官方仓库
git clone https://github.com/gerstung-lab/Delphi.git
cd Delphi

# 2. 创建并激活独立的 Python 环境（强烈推荐）
# 使用 conda 可以更好地管理环境依赖
conda create -n delphi_env python=3.10 -y
conda activate delphi_env

# 3. 安装项目依赖
# requirements.txt 包含了 PyTorch, numpy, pandas, scikit-learn, tqdm, matplotlib 等
pip install -r requirements.txt

# 验证安装
python -c "import torch; print(f'PyTorch version: {torch.__version__}');"

注意事项：

CUDA 版本：
requirements.txt 通常不会指定 CUDA 版本的 PyTorch。请根据你的 GPU 和驱动版本，从 PyTorch 官网 获取合适的安装命令，例如
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118。硬件要求：虽然可以在 CPU 上运行，但训练一个完整规模的 Delphi-2M（尤其是
n_layer=12, n_embd=768）需要至少一张拥有 12GB 以上显存的 GPU。对于入门和调试，可以先用较小的模型配置。

3.2 数据准备：从原始记录到模型输入

这是最关键也最耗时的一步。模型要求一个高度结构化的二进制格式，我们必须将原始的、关系型的 EHR 数据转换为此格式。

3.2.1 核心数据格式：
[patient_id, age_days, token_id]

如前所述，所有数据需要最终聚合成一个
np.uint32 类型的二维数组，并保存为
.bin 文件。每一行代表一个健康事件，包含三列：

强制约束：

按
patient_id 排序：整个文件必须按
patient_id 分块。同一个患者的所有事件记录必须物理上连续存放。这是
train.py 中高效批处理的基础。患者内部按时间排序：在每个患者的事件块内部，事件必须按
age_days 严格递增排列。

3.2.2 事件映射与二进制文件生成：一个完整的 Python 脚本草稿

假设你的原始数据存储在两个 CSV 文件中：


patients.csv:
patient_id, date_of_birth, gender
events.csv:
patient_id, event_date, event_type, code


event_type 可以是
'ICD10',
'BMI',
'SMOKING' 等。
code 是对应的编码或值。

import pandas as pd
import numpy as np
from datetime import datetime
import os

# --- 1. 定义 Token 映射表 ---
# 这是一个简化的例子，你需要根据你的数据和 delphi_labels_chapters_colours_icd.csv
# 来构建一个完整的、一致的映射
# 注意：token_id 从 1 开始，0 保留给 padding
token_to_id = {
    'PAD': 0,
    'MALE': 3,
    'FEMALE': 4,
    'BMI_NORMAL': 5,
    'BMI_OVERWEIGHT': 6,
    'BMI_OBESE': 7,
    'SMOKING_NEVER': 8,
    'SMOKING_FORMER': 9,
    'SMOKING_CURRENT': 10,
    # 添加你需要的所有 ICD-10 编码
    'I10': 123, # 高血压
    'E11': 456, # 糖尿病
    'J45': 789, # 哮喘
    # ...
}
# 生成反向映射用于创建 labels.csv
id_to_token = {v: k for k, v in token_to_id.items()}
max_vocab_size = max(token_to_id.values()) + 1

# --- 2. 读取和预处理原始数据 ---
print("Reading raw data...")
patients_df = pd.read_csv('patients.csv', parse_dates=['date_of_birth'])
events_df = pd.read_csv('events.csv', parse_dates=['event_date'])

# 将患者信息合并到事件表
full_df = pd.merge(events_df, patients_df, on='patient_id')

# --- 3. 转换为 [patient_id, age_days, token_id] 格式 ---
print("Transforming events...")
transformed_records = []

for _, row in full_df.iterrows():
    patient_id = row['patient_id']
    
    # 计算年龄（天）
    age_days = (row['event_date'] - row['date_of_birth']).days
    
    # 根据 event_type 映射到 token_id
    token_id = None
    event_type = row['event_type']
    code = row['code']

    if event_type == 'GENDER':
        token_id = token_to_id.get(code.upper())
    elif event_type == 'BMI':
        # 假设 code 是 BMI 数值，需要分档
        bmi_val = float(code)
        if bmi_val < 25: token_id = token_to_id.get('BMI_NORMAL')
        elif bmi_val < 30: token_id = token_to_id.get('BMI_OVERWEIGHT')
        else: token_id = token_to_id.get('BMI_OBESE')
    elif event_type == 'SMOKING':
        token_id = token_to_id.get(f'SMOKING_{code.upper()}')
    elif event_type == 'ICD10':
        token_id = token_to_id.get(code.upper())
    
    if token_id is None:
        # print(f"Warning: Unmapped event {event_type}:{code}")
        continue # 跳过无法映射的事件

    transformed_records.append([patient_id, age_days, token_id])

# --- 4. 创建 NumPy 数组并排序 ---
print("Creating and sorting array...")
if not transformed_records:
    raise ValueError("No valid events were transformed. Check your mapping.")

data_np = np.array(transformed_records, dtype=np.uint32)

# 排序：首先按 patient_id，然后按 age_days
# 这是模型训练数据处理的关键步骤
sorted_indices = np.lexsort((data_np[:, 1], data_np[:, 0]))
sorted_data_np = data_np[sorted_indices]

# --- 5. 分割训练集和验证集 ---
# 简单按患者分割，确保一个患者的记录不会同时出现在训练和验证集中
unique_pids = np.unique(sorted_data_np[:, 0])
np.random.shuffle(unique_pids)

split_ratio = 0.9
split_idx = int(len(unique_pids) * split_ratio)
train_pids = set(unique_pids[:split_idx])
val_pids = set(unique_pids[split_idx:])

train_mask = np.isin(sorted_data_np[:, 0], list(train_pids))
val_mask = np.isin(sorted_data_np[:, 0], list(val_pids))

train_data = sorted_data_np[train_mask]
val_data = sorted_data_np[val_mask]

# --- 6. 保存为二进制文件和 labels.csv ---
output_dir = 'data/my_custom_dataset'
os.makedirs(output_dir, exist_ok=True)

print(f"Saving training data to {output_dir}/train.bin")
train_data.tofile(f'{output_dir}/train.bin')

print(f"Saving validation data to {output_dir}/val.bin")
val_data.tofile(f'{output_dir}/val.bin')

# 保存 labels.csv
# 确保行号等于 token_id
labels_df = pd.DataFrame([id_to_token[i] for i in range(max_vocab_size)], columns=['label'])
labels_df.to_csv(f'{output_dir}/labels.csv', index=False, header=['label'])

print("Data preparation complete!")
print(f"Training set: {len(train_data)} events from {len(train_pids)} patients.")
print(f"Validation set: {len(val_data)} events from {len(val_pids)} patients.")

运行此脚本后，你将在
data/my_custom_dataset/ 目录下得到
train.bin,
val.bin, 和
labels.csv。你的数据现在“格式正确”，可以被 Delphi 的训练脚本读取了。

3.3 训练流程详解

3.3.1 运行官方 Demo

# 使用仓库自带的合成数据快速测试
python train.py config/train_delphi_demo.py --device=cuda --out_dir=run_demo


config/train_delphi_demo.py：这是一个配置文件，用 Python 定义了所有超参数。它使用的是一个小型模型配置，方便快速运行。
--device=cuda：指定使用 GPU。
--out_dir=run_demo：指定模型检查点、日志等输出的目录。

训练过程中，你会在终端看到 loss 的下降曲线。完成后，在
run_demo/ 目录下会生成一个
ckpt.pt 文件。

3.3.2 自定义训练配置

创建一个新文件
config/train_my_delphi.py，可以复制
train_delphi_demo.py 的内容，然后修改关键参数：

# config/train_my_delphi.py

# --- 输出与数据配置 ---
out_dir = 'runs/my_delphi_full'  # 你的模型输出目录
dataset = 'my_custom_dataset'    # 必须与 data/ 下的文件夹名一致
eval_interval = 2000
log_interval = 100
wandb_log = False # 如果你用 wandb，可以设为 True
wandb_project = 'delphi-experiments'

# --- 模型架构参数 ---
# 根据你的数据集大小和 GPU 显存调整
# 以下是接近原始 Delphi-2M 的配置
vocab_size = 2000 # 确保大于你 labels.csv 中的最大 token_id
n_layer = 12
n_head = 12
n_embd = 768
block_size = 1024 # 序列最大长度，越长能捕捉的病史越远，但越耗内存
dropout = 0.1
bias = True

# --- 训练超参数 ---
batch_size = 32  # 根据你的显存调整
learning_rate = 6e-4
max_iters = 200000
weight_decay = 1e-1
beta1 = 0.9
beta2 = 0.95
grad_clip = 1.0
decay_lr = True

# --- 模型特定的数据处理参数 ---
t_min = 1.0
mask_ties = True
# ignore_tokens 是模型在计算损失时会忽略的 token
# 通常是 padding, 和那些我们不希望模型预测的静态信息（如性别）
ignore_tokens = [0, 3, 4, 5, 6, 7, 8, 9, 10] # 根据你的 token 映射表修改

# --- 设备 ---
device = 'cuda' # or 'cpu'
compile = True # PyTorch 2.0+ 的编译加速，推荐

3.3.3 启动训练

python train.py config/train_my_delphi.py


train.py 脚本会执行以下核心逻辑：

加载配置，初始化模型和优化器。使用
np.memmap 以内存映射的方式加载你生成的
.bin 文件，这对于处理超大文件非常高效，因为它不会一次性将整个文件读入内存。
get_p2i 函数会扫描一次数据，建立一个从
patient_id 到其在
.bin 文件中
[起始索引, 长度] 的映射，从而能快速定位每个患者的数据。在训练循环中，
get_batch 函数负责采样一个批次的数据：
它随机选择一批患者。对于每个患者，从其完整轨迹中随机截取一个长度为
block_size 的片段作为输入
X。
Y 就是
X 向左平移一个时间步的结果（下一个事件）。为了模拟真实世界中“没有事件发生”的年份，它会以一定概率在序列中插入
no_event token。 将
X, Y 送入模型，计算
loss，反向传播，更新权重。

3.4 模型推理与风险预测

训练完成后，你得到了
runs/my_delphi_full/ckpt.pt。现在，我们可以用它来做预测。

3.4.1 加载模型

import torch
from model import Delphi, DelphiConfig
import numpy as np

# 1. 加载 checkpoint
ckpt_path = 'runs/my_delphi_full/ckpt.pt'
checkpoint = torch.load(ckpt_path, map_location='cpu')

# 2. 恢复模型配置
model_args = checkpoint['model_args']
config = DelphiConfig(**model_args)

# 3. 初始化模型并加载权重
model = Delphi(config)
state_dict = checkpoint['model']
model.load_state_dict(state_dict)

# 4. 切换到评估模式并移动到设备
model.eval()
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model.to(device)
print(f"Model loaded on {device}")

3.4.2 单个病例的预测

假设我们有一个新患者，我们想预测他未来的风险。首先需要将他/她的病史整理成 Token 序列。

# --- 1. 准备一个病人的历史数据 ---
# 使用与训练时完全相同的 token_to_id 映射
patient_history = [
    (365 * 30, 'MALE'),          # 30岁, 男性
    (365 * 40, 'SMOKING_CURRENT'),# 40岁, 当前吸烟
    (365 * 42, 'I10'),           # 42岁, 诊断高血压
    (365 * 48, 'E11'),           # 48岁, 诊断糖尿病
]

# 转换为 token_id 和年龄(天)序列
ages, tokens = zip(*patient_history)
token_ids = [token_to_id[event] for event in tokens]

# --- 2. 转换为 PyTorch 张量 ---
# 注意需要增加 batch 维度
age_tensor = torch.tensor([ages], dtype=torch.float32, device=device)
token_tensor = torch.tensor([token_ids], dtype=torch.long, device=device)

# --- 3. 处理序列长度 ---
seq_len = token_tensor.size(1)
# 如果序列短于 block_size，则进行 padding
if seq_len < config.block_size:
    pad_len = config.block_size - seq_len
    # 0 是 padding token
    padding = (0, 0, 0, pad_len) # (pad_left, pad_right, pad_top, pad_bottom)
    token_tensor = torch.nn.functional.pad(token_tensor, padding[:2], 'constant', 0)
    age_tensor = torch.nn.functional.pad(age_tensor, padding[2:], 'constant', 0)
else:
    # 如果序列长于 block_size，只取最近的 block_size 个事件
    token_tensor = token_tensor[:, -config.block_size:]
    age_tensor = age_tensor[:, -config.block_size:]

# --- 4. 前向推理 ---
with torch.no_grad():
    logits, _, _ = model(token_tensor, age_tensor) # targets=None 表示是推理模式

# --- 5. 解析预测结果 ---
# logits 的形状是 (batch_size, seq_len, vocab_size)
# 我们通常关心最后一个时间步的预测，因为它包含了所有历史信息对未来最直接的预测
next_event_logits = logits[0, -1, :] # 取最后一个时间步的输出

# 使用 softmax 将 logits 转换为概率
probabilities = torch.softmax(next_event_logits, dim=0)

# --- 6. 输出特定疾病的风险 ---
# 假设我们关心心梗(I21)和肺癌(C34)的风险
risk myocardial_infarction = probabilities[token_to_id['I21']].item()
risk lung_cancer = probabilities[token_to_id['C34']].item()

print(f"预测的下一事件为心肌梗死的概率: {risk_myocardial_infarction:.4f}")
print(f"预测的下一事件为肺癌的概率: {risk_lung_cancer:.4f}")

重要说明：

上述代码输出的
probabilities 是**“下一事件”**的概率分布。它回答的是“在紧邻的下一个时间点，最可能发生什么事？”如果你想得到“未来 5 年内患肺癌的概率”，这是一个更复杂的计算。你需要基于模型的时间预测功能，对时间轴进行积分。
evaluate_delphi.ipynb 中包含了计算**累积风险函数（Cumulative Incidence Function）**的完整实现，这才是临床研究中更常用的风险指标。其核心思想是：将每个离散时间点的新发风险累加起来，得到一个随时间增长的累积风险曲线。

3.5 高级应用：轨迹采样与可解释性

3.5.1 生成健康轨迹（数字孪生）

这个功能最能体现模型的“生成式”本质。

import torch.nn.functional as F

# 假设你已经加载了模型，并准备好初始的 age_tensor, token_tensor
# 我们将从历史轨迹的末尾开始生成

# 设置为生成模式
temperature = 0.8 # 控制生成随机性的参数，越低越确定性
top_k = 50        # 每次只从概率最高的 50 个 token 中采样

generated_tokens = token_tensor.clone().squeeze(0)
generated_ages = age_tensor.clone().squeeze(0)

# 让我们生成未来 20 年（约 7300 天）的轨迹
max_future_age = generated_ages[-1].item() + 7300 

while generated_ages[-1].item() < max_future_age:
    # 获取当前序列
    input_tokens = generated_tokens.unsqueeze(0)
    input_ages = generated_ages.unsqueeze(0)
    
    # 模型预测
    with torch.no_grad():
        logits, _, _ = model(input_tokens, input_ages)
    
    # 获取下一个事件的 logits
    next_event_logits = logits[0, -1, :]
    
    # 应用 temperature
    next_event_logits = next_event_logits / temperature
    
    # Top-k 过滤
    if top_k is not None:
        top_k_logits, top_k_indices = torch.topk(next_event_logits, top_k)
        # 将不在 top_k 中的 logits 设为 -inf
        next_event_logits[next_event_logits < top_k_logits[:, -1:]] = float('-inf')
    
    # 采样下一个 token
    probs = F.softmax(next_event_logits, dim=-1)
    next_token = torch.multinomial(probs, num_samples=1)
    
    # --- (简化) 假设我们预测下一个事件的时间是固定的2年后 ---
    next_age_increment = 365 * 2 
    next_age = generated_ages[-1] + next_age_increment
    
    # 追加到序列中
    generated_tokens = torch.cat((generated_tokens, next_token), dim=0)
    generated_ages = torch.cat((generated_ages, next_age.unsqueeze(0)), dim=0)
    
    # 如果预测了"死亡"或"结束"的 token，就停止
    if next_token.item() == 1: # 假设 1 是结束 token
        break

print("生成的未来轨迹 (Token IDs):", generated_tokens.tolist())
print("对应的年龄 (天):", generated_ages.tolist())

# 你可以将 token_id 翻译回疾病标签进行解读
generated_labels = [id_to_token.get(tid, 'UNK') for tid in generated_tokens.tolist()]
print("生成的未来轨迹 (标签):", generated_labels)

注意：上述时间采样是高度简化的。实际的 Delphi 模型在训练时会学习一个
time-to-event 的分布。在采样时，应该从这个预测出的分布中随机采样一个时间间隔，而不是固定为 2 年。这部分的完整实现需要深入
model.py 中关于时间损失函数的部分。

3.5.2 可视化与可解释性

官方提供的 Jupyter Notebooks 是进行可解释性分析的最好工具。


evaluate_delphi.ipynb：该文件包含了评估模型性能的完整流程，包括：

计算 C-index/AUC。绘制校准曲线，检查模型预测的概率是否与真实发生率一致。注意力可视化：通过热力图展示模型在做特定预测时关注了哪些历史事件。


shap_analysis.ipynb：该文件展示了如何使用 SHAP 库：

解释单个预测：为什么模型认为这位患者未来的肺癌风险高？SHAP 会给出每个历史事件（如“吸烟”、“慢性阻塞性肺病”）的贡献度。解释群体特征：对整个验证集进行分析，找出对预测某类疾病（如心衰）贡献最大的前几个共性特征。

通过这些工具，开发者不仅可以验证模型的“智商”（预测准不准），还能窥探其“内心世界”（为什么这么预测），这对于建立信任和发现新的医学知识至关重要。

第四章：伦理、合规与未来展望

4.1 数据隐私与安全：不可逾越的红线

处理个体健康数据，尤其是在可识别或准可识别的层面上，隐私是最高优先级。

静态与动态风险：即使数据经过“去标识化”（移除姓名、地址），攻击者仍可能通过结合其他公开数据集进行“再识别”。模型权重本身是否会泄露训练数据中的个体信息，这是一个正在被积极研究的领域（模型逆向攻击）。合成数据的双刃剑：虽然合成轨迹是隐私保护的强大工具，但必须确保合成数据的质量和多样性，避免继承原始数据中的偏见，同时要保证其无法被反推出任何真实个体的信息。合规框架：在全球范围内，任何涉及健康数据的处理都必须遵守当地法规，如欧盟的 GDPR、美国的 HIPAA、以及中国的《个人信息保护法》和《人类遗传资源管理条例》。这意味着数据跨境传输、使用目的限定、患者知情同意等都必须有完备的法律文件和流程支撑。

4.2 模型偏见与公平性：技术向善的责任

模型是数据的镜子，数据中的社会偏见会不可避免地被模型学习并放大。

已知偏倚：UK Biobank 的偏倚（种族、社会经济地位）是明确的。如果直接将未经校准的模型应用于一个多元化的城市（如纽约或伦敦），它可能会系统性地低估少数族裔的健康风险，导致医疗资源分配不公，加剧健康鸿沟。检测与缓解：
检测：必须像第三章所述，进行详尽的分层性能评估。缓解：方法包括：在训练时对不同亚组进行重采样或加权；使用对抗性训练来强制模型学习与种族等敏感属性无关的特征表示；在模型输出后进行后处理校准。 透明度：模型的开发者应公开其训练数据的构成、已知的偏倚以及所做的缓解尝试。使用者也有责任理解这些限制，并对模型的输出保持批判性思维。

4.3 临床应用的门槛：从研究到实践的鸿沟

Delphi-2M 要从《Nature》的封面走进医院的诊室，还有很长的路要走。

严谨的临床验证：需要设计并执行大规模、多中心、前瞻性的临床试验，证明在实际临床工作流中，使用 Delphi-2M 辅助决策能改善患者的健康结局（如降低发病率、死亡率、提高生活质量）。监管审批：在美国，这需要向 FDA 申请作为医疗器械（SaMD，Software as a Medical Device）的批准。在欧洲，需要获得 CE 标志。审批过程极其严格，需要提供详尽的安全性、有效性和鲁棒性证据。人机交互与工作流整合：模型的风险预测结果如何以一种医生能快速理解、信任且不增加信息过载的方式呈现出来？是集成到电子病历系统中的一个弹出警告，还是一个详细的风险报告？这需要医学专家、UI/UX 设计师和软件工程师的紧密合作。责任归属：如果模型的错误建议导致了不良后果，责任谁来承担？是医生、医院，还是软件开发者？这需要完善的法律和伦理框架来界定。

4.4 未来展望：更广阔的图景

Delphi-2M 仅仅是一个开始。未来的发展方向令人兴奋：

多模态数据融合：将基因组学、蛋白质组学、医学影像（如 CT、MRI）、可穿戴设备数据（如心率、步态）与 EHR 数据融合。一个更全面的“数字孪生”将能提供更精准的预测和干预建议。更精细的时间动态建模：从预测“首次诊断”发展到建模疾病的严重程度、复发、对治疗的响应以及多器官功能的协同衰退。因果推断：当前的模型主要学习相关性。未来的模型将尝试学习因果关系。例如，不仅能预测服用某药物后患者的风险变化，还能推断出这种变化是药物的直接效果还是其他混杂因素导致。个性化干预模拟：模型的终极目标是成为“个体化医疗的模拟器”。医生可以输入患者的数字孪生，然后模拟“如果患者开始服用A药”或“如果患者改变了饮食结构”，其未来 10 年的健康轨迹会如何改变，从而为每位患者选择最优的干预方案。

结语

Delphi-2M 的开源，是人工智能与精准医学交汇处的一座重要里程碑。它向我们展示了生成式模型在理解和预测复杂生命系统方面的巨大潜力。然而，它也像一面棱镜，折射出我们在数据、算法、伦理和应用方面面临的挑战。作为研究者和开发者，我们手中的代码不仅是创造未来的工具，更承载着一份沉甸甸的社会责任。唯有在严谨的科学精神、深厚的人文关怀和严格的伦理准则指引下，我们才能确保这项强大的技术真正走向“科技向善”，最终造福全人类的健康。

附录

A… 关键配置参数详解

（此处可创建一个表格，详细解释
DelphiConfig 中每个参数的含义、取值范围和对模型的影响。）