搭建心源性猝死预警Agent：架构、代码与风险
2026年5月，22岁巴西健美网红 Gabriel Ganley 因肥厚型心肌病（HCM）猝死。他的死因不是训练过度，而是遗传性心脏结构异常——这类问题在年轻运动员中常被忽视，直到心脏停跳。
这件事对开发者的冲击是：我们正在构建的健康AI产品，有没有可能提前一天、一小时甚至十分钟预警？ 如果没有，我们的“AI健康助手”最多是个聊天玩具。
本文不讨论健身补剂或悲剧本身。我将拆解一个可解释的心源性猝死预警Agent的完整架构，从心率数据采集到警报下发，附带一个基于HRV（心率变异性）的简化规则引擎实现，并重点分析三类陷阱：数据噪声、假阳性心理伤害、法律问责。读完你可以立刻开始搭建一个原型，并清楚知道哪些地方可能会害死人。
1. 为什么需要预警Agent，而不是单次检测？
猝死不是瞬间事件。心脏骤停前数小时到数天，身体会发出微弱信号：心率变异性（HRV）骤降、夜间心率异常升高、活动后恢复时间延长。这些信号在单次静息心电图里很难捕捉，但连续监测能看出趋势。
传统可穿戴设备（Apple Watch、Garmin）已经提供基础心律报警，但问题是：
- 误报率极高：运动中的心率波动被当成房颤，用户很快关掉通知。
- 缺乏上下文融合：不知道用户昨天是否熬夜、今天是否生病、训练强度是否异常。
- 没有多步推理：只做单点阈值判断，不会根据历史趋势和外部数据做综合决策。
这正是Agent的用武之地：规划 → 感知 → 推理 → 执行。
2. Agent 架构拆解
我把预警系统分成四个模块，每个模块对应Agent的一个能力：
| 模块 | 对应Agent能力 | 输入 | 输出 |
|---|---|---|---|
| 数据采集 & 清洗 | 感知（工具调用） | 心率、加速度、GPS、睡眠 | 规整的时间序列 |
| 特征工程 & 趋势分析 | 短期记忆 + 推理 | 时间序列窗口 | 风险指标（HRV、恢复指数） |
| 混合决策引擎 | 规划（含反思） | 风险指标 + 用户上下文 | 动作标签（正常/警告/警报） |
| 警报路由 & 解释 | 执行（工具输出） | 动作标签 + 证据 | 通知 / 呼叫 / 日志 |
与传统ML pipeline的区别在于：决策引擎内部可以调用多个子工具（大模型、规则、分类器）并执行失败重试。比如当HRV低于阈值时，Agent不是直接报警，而是先查询用户今天的睡眠时长、血氧，再决定是否升级。
3. 核心模块细节与代码
3.1 数据采集：小心苹果的隐私墙
大部分可穿戴设备通过HealthKit或Google Fit暴露数据，但采样频率被限制（HRV通常每5分钟一个点）。为了获取更高频数据，需要BLE直接对接设备（如Polar H10），但用户授权门槛极高。
简易方案：用heartpy库从PPG或ECG信号计算HRV。以下是从CSV读取心率并计算时域HRV的代码：
import heartpy as hp
import pandas as pd
def compute_hrv_from_ecg(ecg_signal, sample_rate=250):
"""
ecg_signal: list of raw ECG voltage values
sample_rate: Hz
返回: {'bpm':..., 'rmssd':..., 'sdnn':...}
"""
wd, m = hp.process(hp.scale_data(ecg_signal), sample_rate)
return {
'bpm': m['bpm'],
'rmssd': m['rmssd'], # 短期变异，交感/副交感平衡
'sdnn': m['sdnn'] # 整体变异
}
# 示例：读取5分钟窗口
window = pd.read_csv('ecg_5min.csv')['voltage'].values
hrv = compute_hrv_from_ecg(window)
print(hrv)踩坑：运动中ECG容易因肌肉电噪声而失真，必须加滤波器（带通5-40Hz）。多数公开数据集的信号质量标签不可用，导致假阳线。
3.2 特征工程：多窗口趋势 + 上下文
单点HRV没有意义，需要看滚动变化率。我设计三个窗口：
- 短期（5分钟 vs 1小时均值）→ 即时异常
- 中期（过去24小时 vs 过去7天基线）→ 日趋势
- 长期（本周 vs 上月）→ 基线迁移（可能提示心肌病变）
同时引入外部上下文：天气（气压变化诱发心梗）、睡眠时长、当天的运动负荷（来自GPS/加速度）。用TF-IDF风格给每个因素赋权，但权重要基于医学证据而非纯统计——这一点LLM做不到，需要领域专家。
def risk_score(hrv_now, hrv_baseline, sleep_hours, training_load, age, sex):
# 基于文献的简易加权
# 来源：Romagnoli et al. 2020, JACC
score = 0
# HRV drop > 30% from baseline: +40 points
if hrv_now < 0.7 * hrv_baseline:
score += 40
# sleep < 5h: +20
if sleep_hours < 5:
score += 20
# training load > 2x usual: +15
if training_load > 2 * baseline_load:
score += 15
# age > 45: +10 (男性)
if age > 45:
score += 10
return score # 0-100这个函数是「固定规则」，但Agent可以在规则输出边界值时（如得分在55-65之间），主动调用LLM分析用户的社交网络抱怨（“昨晚没睡好”“压力大”），做一次上下文增强。
3.3 混合决策引擎：规则 + ML + LLM 的三层管道
为了平衡速度和解释性，我设计了三层管道。
第一层：规则过滤器（延迟毫秒级）
- 如果心率 > 220-age 且持续 >30秒 → 直接紧急警报
- 如果HRV < 硬阈值（rmssd < 10ms）→ 进入第二层
第二层：LightGBM 分类器（延迟百毫秒级）
- 输入：20维特征（HRV时域/频域、心率、运动状态等）
- 输出：风险概率 + SHAP解释值
- 如果概率 >0.8 → 警报；如果0.5~0.8 → 第三层
第三层：LLM 作为二次验证（延迟1-3秒）
- 输入：最近24小时的心率曲线、用户自述、天气、事件摘要
- Prompt 设计强制要求 LLM 输出三个信号：
- 紧急程度（low/medium/high）
- 关键证据（引用特征）
- 建议行动（如“立即静息并监测15分钟”）
这里的关键是第三层不能独立决策，它只提供建议和解释。最终警报由规则引擎根据所有层的分数加权触发。我设置了一个人类在环选项：如果高分警报且用户是第一次出现，Agent不直接发警告，而是发一条“请确认是否感觉异常”的交互消息。
3.4 执行：警报路由的细分
不同风险等级对应不同的执行工具：
| 等级 | 示例分值 | 执行行为 |
|---|---|---|
| 绿色 | <30 | 无操作或存入日志 |
| 黄色 | 30-60 | 推送建议“今天减少强度，多睡眠” |
| 橙色 | 60-80 | 推送 + 询问用户状态，15分钟后复测 |
| 红色 | >80 | 推送 + 自动拨打紧急联系人（需用户预授权） |
注意：对于年轻人（如Gabriel的年龄），橙色级的规则要更保守，因为HCM的首次表现就是猝死，没有渐进过程。实际上，现有医学证据表明HCM患者运动时发生恶性心律失常的风险显著升高，但HRV的变化并非特异性。所以我的系统对22岁健美选手会额外标记为“高风险体质”，即使HRV正常也定期做心电图复查提醒。
4. 简化版动手实现：一个能跑的HRV预警Agent
这里我提供一个最小可复现的Python Agent，基于LangGraph框架（或纯函数管道）。它读取模拟的5分钟ECG数据，计算HRV，执行规则并输出警报。
# 简化版：省略了LLM和ML，仅规则
import time
import numpy as np
import heartpy as hp
from typing import Dict, Optional
class HCMRiskAgent:
def __init__(self, age=22, baseline_hrv=None):
self.age = age
self.baseline_hrv = baseline_hrv or {'rmssd': 40, 'sdnn': 70}
self.history = []
def perceive(self, ecg_signal, sample_rate=250) -> Dict:
# 模拟感知
try:
wd, m = hp.process(hp.scale_data(ecg_signal), sample_rate)
return {
'rmssd': m['rmssd'],
'sdnn': m['sdnn'],
'bpm': m['bpm'],
'timestamp': time.time()
}
except Exception as e:
return {'error': str(e)}
def reason(self, observation: Dict) -> str:
if 'error' in observation:
return 'ERROR: signal quality issue'
self.history.append(observation)
# 规则：rmssd < 50% baseline → warning
rmssd_drop_ratio = observation['rmssd'] / self.baseline_hrv['rmssd']
if rmssd_drop_ratio < 0.3:
return 'CRITICAL: extreme HRV drop, immediate medical attention'
elif rmssd_drop_ratio < 0.5:
return 'WARNING: significant HRV drop, rest and monitor'
else:
return 'NORMAL'
def act(self, decision: str):
if 'CRITICAL' in decision:
print(f"[ALARM] {decision}")
# 实际会调用Twilio发送短信
elif 'WARNING' in decision:
print(f"[ADVISORY] {decision}")
else:
print(f"[SILENT] No action")
# 模拟运行
if __name__ == '__main__':
agent = HCMRiskAgent(age=22)
# 生成一段模拟异常ECG（心率低变异性）
abnormal_ecg = np.sin(np.linspace(0, 100, 1250)) * 0.05 + np.random.normal(0,0.01,1250)
obs = agent.perceive(abnormal_ecg)
decision = agent.reason(obs)
agent.act(decision)运行后会输出[ALARM] ...或[ADVISORY] ...。你可以在GitHub上找到真实ECG数据集（如MIT-BIH）替换abnormal_ecg来测试。
5. 踩坑记录：这些设计可能害死人
5.1 数据噪声：运动伪迹导致的假警报
实测中，手臂摆动产生的加速度能完全淹没ECG信号，使算法误判为极端HRV。解决方案：必须融合加速度计数据，在运动>2MET时禁用HRV分析，只保留心率监测。
5.2 假阳性的心理伤害
如果Agent频繁发出“预警”，用户会产生报警疲劳，甚至出现焦虑障碍。对于健康年轻人，红色警报的每日触发率必须低于0.01%（每万天一次）。这意味着我们的阈值必须足够保守，并且在警报前先询问用户是否感觉异常。
5.3 法律问责：谁为错误的警报负责？
如果Agent没有预警（真阴性）导致死亡，还是发出假警报导致不必要的急诊，责任在开发者、设备厂商还是用户？目前美国FDA对这类软件按“医疗设备”监管（SaMD），需要510(k)预通知。你的项目如果面向真实用户，必须获得监管批准。我个人的建议是：不要把Agent设计成诊断工具，而是作为“健康建议助手”，并在UI中明确声明不提供医疗建议。
5.4 LLM的幻觉风险
第三层使用LLM时，我曾见过GPT-4基于错误的心率曲线编造“用户可能发生心肌炎”的结论，而实际只是传感器脱落。解决方案：LLM必须引用具体特征值（如“RMSSD从40降至15”），并通过正则校验确保输出格式可控。
6. 对开发者的建议
- 领域知识 > 模型复杂度。心源性猝死的病理机制非常清楚，规则引擎往往比任何深度学习都可靠。ML只能辅助覆盖多变量非线性关系。
- 不要忽视环境因素。气压变化、感染、饮水不足都会影响心率。你的Agent应该接入天气API和睡眠追踪。
- 伦理设计优先。在代码中嵌入“人类在环”（Human-in-the-Loop）逻辑，让用户拥有最终决策权。
- 保留原始数据。所有决策都必须可追溯，留下审计日志。一旦出现法律纠纷，这是唯一的保护。
Gabriel Ganley的悲剧提醒我们：技术可以做的比现在更多。但只有清醒地认识到技术边界，我们才能安全地推动边界。如果你现在就想开始，请先下载一个干净的ECG数据集，跑通上面的简化Agent，然后逐步加入你关注的维度。
不要等到事故发生后，才明白预警信号的含义。