第四章：技术展望：医疗AI的\”第一性原理\”时代
4.1 从\”相关性\”到\”因果性\”的范式跃迁
4.1.1 因果革命：超越统计相关性的医学认知
传统医疗AI主要关注\”是什么\”的关联性问题，而第一性原理范式致力于回答\”为什么\”的因果性问题。这种转变要求我们从观察性数据分析转向干预性推理和反事实预测。
因果推理的三个层次：
# 医疗因果发现与推理系统
import numpy as np
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
from scipy import stats
import dowhy
from dowhy import CausalModel
import warnings
warnings.filterwarnings(\’ignore\’)
class CausalMedicalAI:
def __init__(self):
\”\”\”初始化医疗因果AI系统\”\”\”
self.causal_models = {
}
self.causal_discovery_results = {
}
def simulate_medical_causal_data(self, n_samples=1000):
\”\”\”模拟医疗因果数据\”\”\”
print(\”=== 生成模拟医疗因果数据 ===\”)
np.random.seed(42)
# 定义因果结构
# U1: 遗传因素 (未观测)
# U2: 环境因素 (未观测)
# X1: 吸烟 (0=不吸, 1=吸烟)
# X2: 饮酒 (0=不饮, 1=饮酒)
# X3: 运动 (0=不运动, 1=规律运动)
# X4: 血压 (连续值)
# X5: 血糖 (连续值)
# Y: 心血管疾病风险 (0=低风险, 1=高风险)
# 未观测混杂因素
U1 = np.random.normal(0, 1, n_samples) # 遗传因素
U2 = np.random.normal(0, 1, n_samples) # 环境因素
# 吸烟 (受遗传和环境因素影响)
smoking_prob = 1 / (1 + np.exp(–(0.5 * U1 + 0.3 * U2)))
X1 = np.random.binomial(1, smoking_prob)
# 饮酒 (与吸烟相关，也受环境因素影响)
drinking_prob = 1 / (1 + np.exp(–(0.7 * X1 + 0.4 * U2)))
X2 = np.random.binomial(1, drinking_prob)
# 运动 (与吸烟负相关，受遗传因素影响)
exercise_prob = 1 / (1 + np.exp(–(–0.6 * X1 + 0.5 * U1)))
X3 = np.random.binomial(1, exercise_prob)
# 血压 (受吸烟、饮酒、运动和遗传因素影响)
X4 = 120 + 8 * X1 + 5 * X2 – 7 * X3 + 6 * U1 + np.random.normal(0, 5, n_samples)
# 血糖 (受吸烟、饮酒、运动和遗传因素影响)
X5 = 5.0 + 0.8 * X1 + 0.5 * X2 – 0.6 * X3 + 0.7 * U1 + np.random.normal(0, 0.5, n_samples)
# 心血管疾病风险 (受血压、血糖、吸烟、运动和遗传因素影响)
risk_score = (0.3 * X1 + 0.1 * X2 – 0.2 * X3 +
0.02 * (X4 – 120) + 0.5 * (X5 – 5.0) +
0.4 * U1 + 0.2 * U2 + np.random.normal(0, 0.5, n_samples))
disease_prob = 1 / (1 + np.exp(–risk_score))
Y = np.random.binomial(1, disease_prob)
# 创建数据框
data = pd.DataFrame({
\’smoking\’: X1,
\’drinking\’: X2,
\’exercise\’: X3,
\’blood_pressure\’: X4,
\’blood_sugar\’: X5,
\’heart_disease\’: Y
})
# 真实因果图
true_causal_graph = nx.DiGraph()
true_causal_graph.add_edges_from([
(\’U1\’, \’smoking\’), (\’U1\’, \’exercise\’),
(\’U1\’, \’blood_pressure\’), (\’U1\’, \’blood_sugar\’),
(\’U1\’, \’heart_disease\’),
(\’U2\’, \’smoking\’), (\’U2\’, \’drinking\’),
(\’U2\’, \’heart_disease\’),
(\’smoking\’, \’drinking\’),
(\’smoking\’, \’exercise\’),
(\’smoking\’, \’blood_pressure\’),
(\’smoking\’, \’blood_sugar\’),
(\’smoking\’, \’heart_disease\’),
(\’drinking\’, \’blood_pressure\’),
(\’drinking\’, \’blood_sugar\’),
(\’exercise\’, \’blood_pressure\’),
(\’exercise\’, \’blood_sugar\’),
(\’exercise\’, \’heart_disease\’),
(\’blood_pressure\’, \’heart_disease\’),
(\’blood_sugar\’, \’heart_disease\’)
])
print(f\”生成 {
n_samples} 个样本\”)
print(f\”变量: {
list(data.columns)}\”)
print(f\”疾病发生率: {
Y.mean():.2%}\”)
return data, true_causal_graph
def traditional_correlation_analysis(self, data):
\”\”\”传统相关性分析\”\”\”
print(\”\\n=== 传统相关性分析 ===\”)
# 计算相关系数
corr_matrix = data.corr()
# 可视化相关矩阵
fig, ax = plt.subplots(figsize=(10, 8))
cax = ax.matshow(corr_matrix, cmap=\’coolwarm\’, vmin=–1, vmax=1)
fig.colorbar(cax)
# 添加数值标签
for (i, j), val in np.ndenumerate(corr_matrix):
ax.text(j, i, f\’{
val:.2f}\’, ha=\’center\’, va=\’center\’, fontsize=10)
ax.set_xticks(range(len(corr_matrix.columns)))
ax.set_yticks(range(len(corr_matrix.columns)))
ax.set_xticklabels(corr_matrix.columns, rotation=45)
ax.set_yticklabels(corr_matrix.columns)
ax.set_title(\’医疗变量相关矩阵\’, fontsize=14, pad=20)
plt.tight_layout()
plt.show()
# 分析吸烟与心脏病的关系
print(\”\\n吸烟与心脏病相关性分析:\”)
smoking_disease_corr = corr_matrix.loc[\’smoking\’, \’heart_disease\’]
print(f\” 相关系数: {
smoking_disease_corr:.3f}\”)
# 分层分析
print(\”\\n分层分析 (按运动习惯):\”)
exercisers = data[data[\’exercise\’] == 1]
non_exercisers = data[data[\’exercise\’] == 0]
corr_exercisers = exercisers[\’smoking\’].corr(exercisers[\’heart_disease\’])
corr_non_exercisers = non_exercisers[\’smoking\’].corr(non_exercisers[\’heart_disease\’])
print(f\” 运动者中吸烟与心脏病相关性: {
corr_exercisers:.3f}\”)
print(f\” 非运动者中吸烟与心脏病相关性: {
corr_non_exercisers:.3f}\”)
return corr_matrix
def causal_discovery(self, data, method=\’pc\’):
\”\”\”因果发现：从数据中发现因果结构\”\”\”
print(f\”\\n=== 因果发现 (方法: {
method}) ===\”)
if method == \’pc\’:
# 使用PC算法进行因果发现
from causalnex.discovery import PC
from causalnex.network import BayesianNetwork
# 离散化连续变量（简化处理）
data_discrete = data.copy()
data_discrete[\’blood_pressure\’] = pd.cut(data[\’blood_pressure\’],
bins=3, labels=[0, 1, 2])
data_discrete[\’blood_sugar\’] = pd.cut(data[\’blood_sugar\’],
bins=3, labels=[0, 1, 2])
# 应用PC算法
pc = PC()
pc.fit(data_discrete)
# 获取因果图
causal_graph = pc.get_causal_graph()
# 转换为networkx图用于可视化
G = nx.DiGraph()
for edge in causal_graph.edges:
G.add_edge(edge[0], edge[1])
elif method == \’lingam\’:
# 使用LiNGAM算法（线性非高斯模型）
from lingam import DirectLiNGAM
# 准备数据（标准化）
X = data.values
X_standardized = (X – X.mean(axis=0)) / X.std(axis=0)
# 拟合LiNGAM模型
model = DirectLiNGAM()
model.fit(X_standardized)
# 获取因果顺序和邻接矩阵
causal_order = model.causal_order_
adjacency_matrix = model.adjacency_matrix_
# 构建因果图
G = nx.DiGraph()
variable_names = list(data.columns)
for i in range(len(causal_order)):
for j in range(i+1, len(causal_order)):
if abs(adjacency_matrix[causal_order[j], causal_order[i]]) > 0.1:
cause = variable_names[causal_order[i]]
effect = variable_names[causal_order[j]]
G.add_edge(cause, effect,
weight=adjacency_matrix[causal_order[j], causal_order[i]])
# 可视化发现的因果图
self._visualize_causal_graph(G, title=f\’因果发现结果 ({
method}算法)\’)
self.causal_discovery_results[method] = G
return G
def _visualize_causal_graph(self, G, title=\”因果图\”):
\”\”\”可视化因果图\”\”\”
plt.figure(figsize=(10, 8))
# 使用spring布局
pos = nx.spring_layout(G, seed=42)
# 绘制节点
node_colors = []
for node in G.nodes():
if \’heart\’ in node.lower():
node_colors.append(\’red\’)
elif \’blood\’ in node.lower():
node_colors.append(\’orange\’)
elif node in [\’smoking\’, \’drinking\’]:
node_colors.append(\’blue\’)
elif node == \’exercise\’:
node_colors.append(\’green\’)
else:
node_colors.append(\’gray\’)
nx.draw_networkx_nodes(G, pos, node_color=node_colors,
node_size=2000, alpha=0.8)
# 绘制边
edge_weights = [G.edges[edge].get(\’weight\’, 1.0) for edge in G.edges()]
nx.draw_networkx_edges(G, pos, width=[abs(w)*5 for w in edge_weights],
alpha=0.6, edge_color=\’gray\’, arrows=True, arrowsize=20)
# 绘制标签
nx.draw_networkx_labels(G, pos, font_size=12, font_weight=\’bold\’)
# 添加边权重标签（如果有）
if edge_weights and any(w != 1.0 for w in edge_weights):
edge_labels = {
(u, v): f\”{
G.edges[(u, v)].get(\’weight\’, 0):.2f}\”
for (u, v) in G.edges()}
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=9)
plt.title(title, fontsize=14, fontweight=\’bold\’)
plt.axis(\’off\’)
plt.tight_layout()
plt.show()
def causal_effect_estimation(self, data, treatment, outcome):
\”\”\”因果效应估计\”\”\”
print(f\”\\n=== 因果效应估计: {
treatment} → {
outcome} ===\”)
# 使用DoWhy进行因果推断
# 定义因果模型
model = CausalModel(
data=data,
treatment=treatment,
outcome=outcome,
common_causes=[\’drinking\’, \’exercise\’, \’blood_pressure\’, \’blood_sugar\’]
)
# 可视化因果模型
model.view_model()
# 识别因果效应
identified_estimand = model.identify_effect()
print(f\”\\n识别结果: {
identified_estimand}\”)
# 估计因果效应
# 方法1: 线性回归
estimate_reg = model.estimate_effect(
identified_estimand,
method_name=\”backdoor.linear_regression\”
)
print(f\”\\n线性回归估计:\”)
print(f\” 估计效应: {
estimate_reg.value:.4f}\”)
print(f\” 95%置信区间: [{
estimate_reg.get_confidence_intervals()[0]:.4f}, \”
f\”{
estimate_reg.get_confidence_intervals()[1]:.4f}]\”)
# 方法2: 倾向得分匹配
estimate_psm = model.estimate_effect(
identified_estimand,
method_name=\”backdoor.propensity_score_matching\”
)
print(f\”\\n倾向得分匹配估计:\”)
print(f\” 估计效应: {
estimate_psm.value:.4f}\”)
print(f\” 95%置信区间: [{
estimate_psm.get_confidence_intervals()[0]:.4f}, \”
f\”{
estimate_psm.get_confidence_intervals()[1]:.4f}]\”)
# 方法3: 双重稳健估计
estimate_dr = model.estimate_effect(
identified_estimand,
method_name=\”backdoor.doubly_robust\”
)
print(f\”\\n双重稳健估计:\”)
print(f\” 估计效应: {
estimate_dr.value:.4f}\”)
print(f\” 95%置信区间: [{
estimate_dr.get_confidence_intervals()[0]:.4f}, \”
f\”{
estimate_dr.get_confidence_intervals()[1]:.4f}]\”)
# 敏感性分析
print(\”\\n敏感性分析:\”)
sensitivity = model.refute_estimate(estimate_reg, \”placebo_treatment_refuter\”)
print(f\” 安慰剂检验: 效应值 = {
sensitivity.new_effect:.4f}\”)
# 解释因果效应
self._interpret_causal_effect(treatment, outcome, estimate_reg.value, data)
return {
\’linear_regression\’: estimate_reg,
\’propensity_score_matching\’: estimate_psm,
\’doubly_robust\’: estimate_dr
}
def _interpret_causal_effect(self, treatment, outcome, effect, data):
\”\”\”解释因果效应\”\”\”
print(f\”\\n因果效应解释:\”)
if treatment == \’smoking\’ and outcome == \’heart_disease\’:
print(f\” 吸烟对心脏病的平均因果效应: {
effect:.4f}\”)
if effect > 0:
print(\” 解释: 吸烟会增加心脏病风险\”)
# 计算人群归因分数
smoking_rate = data[\’smoking\’].mean()
risk_difference = effect
population_attributable_fraction = (smoking_rate * risk_difference) / data[\’heart_disease\’].mean()
print(f\” 吸烟人群心脏病风险增加: {
risk_difference:.2%}\”)
print(f\” 人群中归因于吸烟的心脏病比例: {
population_attributable_fraction:.2%}\”)
# 临床意义
if effect > 0.1:
print(\” 临床意义: 强因果关系，强烈建议戒烟\”)
elif effect > 0.05:
print(\” 临床意义: 中等因果关系，建议戒烟\”)
else:
print(\” 临床意义: 弱因果关系，但仍建议减少吸烟\”)
elif treatment == \’exercise\’ and outcome == \’heart_disease\’:
print(f\” 运动对心脏病的平均因果效应: {
effect:.4f}\”)
if effect < 0:
print(\” 解释: 规律运动会降低心脏病风险\”)
print(f\” 运动可降低心脏病风险: {
abs(effect):.2%}\”)
# 计算需要治疗的人数
nnt = 1 / abs(effect) if effect != 0 else float(\’inf\’)
print(f\” 需要治疗人数 (NNT): {
nnt:.1f} 人\”)
print(f\”\\n临床决策建议:\”)
print(\” 1. 基于因果证据而非仅相关性\”)
print(\” 2. 考虑混杂因素调整\”)
print(\” 3. 评估个体化因果效应\”)
print(\” 4. 结合机制理解进行解释\”)
def counterfactual_prediction(self, data, patient_id, intervention):
\”\”\”反事实预测：如果采取不同干预会怎样\”\”\”
print(f\”\\n=== 反事实预测 (患者 {
patient_id}) ===\”)
# 获取患者当前状态
patient = data.iloc[patient_id]
print(f\”患者当前状态:\”)
print(f\” 吸烟: {
\’是\’ if patient[\’smoking\’] == 1 else \’否\’}\”)
print(f\” 饮酒: {
\’是\’ if patient[\’drinking\’] == 1 else \’否\’}\”)
print(f\” 运动: {
\’是\’ if patient[\’exercise\’] == 1 else \’否\’}\”)
print(f\” 血压: {
patient[\’blood_pressure\’]:.1f}\”)
print(f\” 血糖: {
patient[\’blood_sugar\’]:.1f}\”)
print(f\” 心脏病风险: {
\’高风险\’ if patient[\’heart_disease\’] == 1 else \’低风险\’}\”)
# 构建反事实模型
from sklearn.ensemble import GradientBoostingClassifier
# 准备数据
X = data.drop(\’heart_disease\’, axis=1)
y = data[\’heart_disease\’]
# 训练模型
model = GradientBoostingClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
# 当前状态预测
current_pred = model.predict_proba(patient.drop(\’heart_disease\’).values.reshape(1, –1))[0, 1]
# 反事实状态
counterfactual_patient = patient.copy()
if intervention == \’quit_smoking\’:
counterfactual_patient[\’smoking\’] = 0
intervention_desc = \”戒烟\”
elif intervention == \’start_exercise\’:
counterfactual_patient[\’exercise\’] = 1
intervention_desc = \”开始规律运动\”
elif intervention == \’control_bp\’:
counterfactual_patient[\’blood_pressure\’] = 120 # 控制到正常值
intervention_desc = \”控制血压到正常水平\”
else:
intervention_desc = intervention
# 反事实预测
counterfactual_pred = model.predict_proba(
counterfactual_patient.drop(\’heart_disease\’).values.reshape(1, –1)
)[0, 1]
# 计算反事实效应
effect = current_pred – counterfactual_pred
print(f\”\\n反事实情景: 如果{
intervention_desc}\”)
print(f\” 当前心脏病风险: {
current_pred:.2%}\”)
print(f\” 反事实心脏病风险: {
counterfactual_pred:.2%}\”)
print(f\” 风险变化: {
effect:+.2%} ({
\’降低\’ if effect > 0 else \’增加\’})\”)
# 解释反事实结果
if effect > 0.1:
print(f\” 临床意义: 强烈建议{
intervention_desc}\”)
elif effect > 0.05:
print(f\” 临床意义: 建议{
intervention_desc}\”)
else:
print(f\” 临床意义: {
intervention_desc}可能带来有限益处\”)
return {
\’patient_id\’: patient_id,
\’intervention\’: intervention,
\’current_risk\’: current_pred,
\’counterfactual_risk\’: counterfactual_pred,
\’risk_change\’: effect
}
def demonstrate_mediation_analysis(self, data):
\”\”\”中介分析：揭示因果机制\”\”\”
print(\”\\n=== 中介分析: 吸烟→血压→心脏病 ===\”)
# 使用中介分析模型
import statsmodels.api as sm
# 模型1: 吸烟对血压的影响
X1 = sm.add_constant(data[\’smoking\’])
model1 = sm.OLS(data[\’blood_pressure\’], X1).fit()
# 模型2: 吸烟和血压对心脏病的影响
X2 = sm.add_constant(data[[\’smoking\’, \’blood_pressure\’]])
model2 = sm.Logit(data[\’heart_disease\’], X2).fit(disp=0)
print(\”\\n中介分析结果:\”)
print(f\”1. 吸烟对血压的效应: {
model1.params[\’smoking\’]:.3f}\”)
print(f\” (p值: {
model1.pvalues[\’smoking\’]:.4f})\”)
print(f\”\\n2. 血压对心脏病的效应: {
model2.params[\’blood_pressure\’]:.3f}\”)
print(f\” (p值: {
model2.pvalues[\’blood_pressure\’]:.4f})\”)
print(f\”\\n3. 吸烟对心脏病的直接效应: {
model2.params[\’smoking\’]:.3f}\”)
print(f\” (p值: {
model2.pvalues[\’smoking\’]:.4f})\”)
# 计算中介效应
indirect_effect = model1.params[\’smoking\’] * model2.params[\’blood_pressure\’]
direct_effect = model2.params[\’smoking\’]
total_effect = indirect_effect + direct_effect
proportion_mediated = indirect_effect / total_effect if total_effect != 0 else 0
print(f\”\\n中介效应分解:\”)
print(f\” 总效应: {
total_effect:.3f}\”)
print(f\” 直接效应: {
direct_effect:.3f} ({
direct_effect/total_effect:.1%})\”)
print(f\” 间接效应(通过血压): {
indirect_effect:.3f} ({
proportion_mediated:.1%})\”)
# 可视化中介模型
self._visualize_mediation_model(
total_effect, direct_effect, indirect_effect, proportion_mediated
)
return {
\’total_effect\’: total_effect,
\’direct_effect\’: direct_effect,
\’indirect_effect\’: indirect_effect,
\’proportion_mediated\’: proportion_mediated
}
def _visualize_mediation_model(self, total, direct, indirect, proportion):
\”\”\”可视化中介模型\”\”\”
fig, axes = plt.subplots(1, 2, figsize=(12, 5))
# 子图1: 效应分解
ax1 = axes[0]
effects = [direct, indirect]
labels = [\’直接效应\’, \’间接效应\\n(通过血压)\’]
colors = [\’lightcoral\’, \’lightblue\’]
bars = ax1.bar(labels, effects, color=colors, alpha=0.8)
ax1.set_ylabel(\’效应大小\’)
ax1.set_title(\’吸烟对心脏病效应的分解\’)
ax1.grid(True, alpha=0.3, axis=\’y\’)
# 添加数值标签
for bar, effect in zip(bars, effects):
height = bar
