心电图AI系统部署实战：从信号预处理到API服务全流程解析

一张10秒的12导联心电图，交给AI模型推理，单条结果秒级返回，输出的不是“正常”或“异常”这种笼统结论，而是包含心律失常类型、置信度和波形异常位置的辅助判读信息。这个场景现在已经不是论文里的演示，而是很多心电图AI系统在真实部署时的标准流程。这篇文章要把这套流程拆开讲：心电信号预处理、模型推理、结果输出、批量文件处理、API服务封装，每一步怎么落地都会过一遍。

先说明边界：文章讨论的是面向科研验证和辅助筛查的AI心电图判读工作流，不涉及具体医疗诊断决策，也不代表任何一种临床结论。所谓“隐藏的心脏疾病信号”，指的是阵发性房颤、早搏、ST段轻微偏移、QT间期异常这类容易在快速阅片时被忽略的波形变化，AI擅长做的不是取代医生，而是先把可疑片段找出来，让医生复核。如果你手头正好有心电数据，或者正在评估要不要引入心电图AI，这篇文章可以直接收藏。

1. 核心能力速览

从技术形态看，这套系统解决的是“心电图信号分类”问题：输入一段多导联心电记录，输出疾病标签和置信度。它和常见的图像分类模型不同，输入不是图片，而是时序信号，所以前处理和后处理都围绕导联、采样率、窗口长度和类别概率展开。

这套流程的关键点不在“模型多大”，而在“能不能稳定处理真实心电图”。真实数据和公开数据集差别很大：电极贴错、基线漂移、工频干扰、导联顺序不一致，任何一个问题都会让模型输出崩掉。所以后面的章节会把数据校验放在部署之前，顺序不能反。

2. 适用场景与使用边界

先讲能用在哪儿。第一类是科研场景：用PTB-XL、MIT-BIH心律失常数据库这类公开数据集做模型训练、对比实验和论文复现，重点是跑通“信号预处理-特征提取-分类器输出”的闭环。第二类是体检和健康管理的初筛场景：大量静息心电图需要被快速分拣，AI先把“高度怀疑异常”的记录挑出来，医生重点看这些，普通记录快速过。第三类是产品原型验证：公司想做心电图AI功能，先用小批量数据验证模型精度和性能，再决定是否投入医疗注册流程。

再讲不适合什么。心电图AI模型的输出不能直接用于诊断，尤其是涉及胸痛、晕厥、急性心梗这类高风险场景时，AI只能作为辅助参考，不能作为决策依据。医疗环境部署必须走对应国家或地区的医疗器械软件验证流程，比如NMPA注册、临床试验、算法性能评估，这个环节不是技术文章能替代的。另外，数据边界必须守住：心电信号属于敏感健康数据，训练、测试、部署时都要做去标识化处理，涉及患者数据要取得合法授权，不能拿未经许可的医院数据做模型训练。

还有一个容易被忽略的点：心电图AI模型有“人群迁移”问题。在A医院数据上训练出来的模型，换到B医院的设备、导联排布、采样设置后，准确率往往会下降。更稳妥的判断是：模型上线前必须在目标设备和目标人群上重新做验证，不能拿公开数据集上的精度直接当真实场景精度。

3. 环境准备与前置条件

如果你要从零开始部署一套心电图AI推理服务，建议先按下面的清单检查环境，避免跑到一半发现基础库版本冲突。

软件层面需要准备：

• 操作系统：Windows 10/11、Ubuntu 18.04及以上、macOS（仅CPU推理建议）。
• Python版本：3.8到3.11都可以，优先3.10，第三方库兼容性比较好。
• 深度学习框架：PyTorch或TensorFlow，模型如果是ONNX格式，可以只装onnxruntime，不装完整训练框架。
• 心电读取库：wfdb用于读取MIT格式心电记录，numpy和scipy做信号处理。
• 接口服务：fastapi和uvicorn，用于把推理封装成REST API。
• 数据处理：pandas、scikit-learn，批量结果整理和指标计算会用到。

硬件层面不做死规定，但建议看模型规模和使用频率选：

• 只做少量验证：CPU足够，模型加载和单条推理慢一点，但流程能跑通。
• 批量测试或接口服务：建议NVIDIA GPU，显存8GB起步，具体占用以实际模型为准。
• 纯测试小模型：4GB显存也有可能，但需要配合量化或降低batch size。

依赖安装可以先用一个通用的requirements.txt，版本号根据你本机环境调整：

txt复制numpy==1.24.3
scipy==1.10.1
pandas==2.0.3
wfdb==4.1.1
torch==2.1.0
onnxruntime==1.16.3
fastapi==0.104.1
uvicorn==0.24.0
scikit-learn==1.3.2

bash复制conda create -n ecg-ai python=3.10 -y
conda activate ecg-ai
pip install -r requirements.txt

数据方面，输入可以是标准MIT格式的.dat和.hea文件，也可以是已经转换成numpy数组的12导联信号。要提前确认三个参数：导联数量、采样率、记录时长。模型训练时的输入维度通常是[batch, leads, samples]或[batch, samples, leads]，导联顺序和采样率必须和训练时保持一致，这一步写进预处理函数，不要靠人工每次核对。

4. 模型加载与推理启动方式

先给一个最小可运行的推理脚本。下面代码里的模型路径、输入形状和预处理参数都是示例，实际部署时要替换成自己的模型文件。

python复制import numpy as np
import torch

# 示例：模型加载
# 实际项目中替换为你的模型权重文件路径
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = torch.load("ecg_model.pt", map_location=device)
model.eval()

# 示例：预处理
def preprocess_ecg(raw_signal: np.ndarray, target_fs: int = 500) -> torch.Tensor:
    """
    raw_signal: shape (leads, samples)，float32
    实际部署时按训练阶段的预处理流程替换
    """
    # 去基线漂移、滤波、重采样等操作写在这里
    signal = raw_signal.astype(np.float32)
    # 归一化，具体统计量以训练集为准
    signal = (signal - signal.mean(axis=-1, keepdims=True)) / (signal.std(axis=-1, keepdims=True) + 1e-5)
    tensor = torch.from_numpy(signal).float()
    tensor = tensor.unsqueeze(0)  # 增加 batch 维度
    return tensor.to(device)

# 示例：推理
def predict_ecg(raw_signal: np.ndarray) -> dict:
    tensor = preprocess_ecg(raw_signal)
    with torch.no_grad():
        logits = model(tensor)
    probs = torch.softmax(logits, dim=-1).cpu().numpy()[0]
    label_idx = int(np.argmax(probs))
    confidence = float(probs[label_idx])
    return {"label_id": label_idx, "confidence": confidence, "probabilities": probs.tolist()}

启动方式有三种。

第一种是命令行单文件推理，适合验证单条心电记录：

bash复制python infer_one.py --record_path ./data/record1.dat --output ./result.json

第二种是ONNX Runtime启动，适合不想装PyTorch完整训练环境的推理机。先把PyTorch模型导出为ONNX，再用onnxruntime加载：

bash复制python -m onnxruntime.tools.convert_onnx_models_to_ort model.onnx

python复制import onnxruntime as ort
import numpy as np

sess = ort.InferenceSession("ecg_model.onnx", providers=["CUDAExecutionProvider", "CPUExecutionProvider"])
input_name = sess.get_inputs()[0].name
output_name = sess.get_outputs()[0].name

def predict_onnx(signal: np.ndarray) -> dict:
    # 注意这里的输入shape和预处理要和导出时一致
    outputs = sess.run([output_name], {input_name: signal})
    return {"logits": outputs[0].tolist()}

第三种是FastAPI接口服务，适合把推理能力开放给其他模块调用。下面是一个最小示例：

python复制from fastapi import FastAPI, UploadFile, File
from pydantic import BaseModel

app = FastAPI()

class ECGSignal(BaseModel):
    signal: list
    fs: int = 500

@app.post("/predict_array")
def predict_array(data: ECGSignal):
    # 将JSON数组转为numpy，再调用预处理和推理
    signal = np.array(data.signal, dtype=np.float32)
    result = predict_ecg(signal)
    return result

@app.post("/predict_file")
async def predict_file(file: UploadFile = File(...)):
    content = await file.read()
    # 示例：先保存临时文件，再用 wfdb 读取
    temp_path = f"/tmp/{file.filename}"
    with open(temp_path, "wb") as f:
        f.write(content)
    # 实际读取逻辑需要根据文件格式调整
    return {"status": "ok", "message": "文件已接收，解析逻辑按实际格式补充"}

启动服务：

bash复制uvicorn main:app --host 0.0.0.0 --port 8000

这里有一个常见坑：FastAPI接收JSON数组时，如果心电信号是500Hz的10秒12导联数据，JSON体积会比较大，单条记录可能达到几MB，接口响应时间会被序列化拖慢。更推荐的做法是直接传原始心电文件或压缩成numpy的.npz格式，服务端读取后转数组。

5. 功能测试与效果验证

5.1 单条心电记录推理测试

测试目的：确认模型能完成最基本的“输入一段心电信号，输出分类结果”流程。

操作步骤：

1. 准备一条已标注的心电记录，最好是模型验证集里的样本，这样能对照真实标签。
2. 执行推理脚本，观察输出的label_id和置信度。
3. 把输出的softmax概率分布贴到结果文件里，不要只保留最高类。

预期结果：推理能正常结束，输出概率向量长度等于模型类别数，最高概率对应的类别和标注标签一致或接近。

判断成功标准：模型没有报错，输出概率之和为1，标签映射正确。如果输出的是归一化前logits，需要确认后续是否做softmax，避免直接把logits当概率看。

5.2 多导联与采样率稳定性测试

测试目的：验证预处理是不是能稳定处理不同来源的心电数据。

重点测三种情况：

• 原始采样率不是模型输入采样率，重采样后是否正常。
• 导联数量不是12，比如单导联或三导联，模型是否做兼容。
• 信号长度超过10秒或不足10秒，截断和补零策略是否正确。

我在实际项目里的经验是，这类问题比模型精度问题更容易暴露。很多模型在公开数据集上精度很高，但一接真实设备和真实数据，图像不输出或者输出异常，往往是预处理函数对“非标准输入”没有做防御性判断。

一个可用的防御性处理框架是这样的：

python复制def normalize_ecg_input(signal, input_fs, expected_fs=500, expected_leads=12):
    import scipy.signal as sps
    # 重采样
    if input_fs != expected_fs:
        resample_len = int(signal.shape[-1] * expected_fs / input_fs)
        signal = np.array([sps.resample(lead, resample_len) for lead in signal])
    # 导联数校验
    if signal.shape[0] < expected_leads:
        raise ValueError(f"导联数不足：当前{signal.shape[0]}，预期{expected_leads}")
    # 长度对齐
    if signal.shape[-1] > expected_len:
        signal = signal[:, :expected_len]
    elif signal.shape[-1] < expected_len:
        padding = expected_len - signal.shape[-1]
        signal = np.pad(signal, ((0, 0), (0, padding)), mode="constant")
    return signal.astype(np.float32)

5.3 批量心电文件推理测试

测试目的：确认模型的批量处理能力，以及长时间运行的稳定性。

操作步骤：

1. 准备一个目录，里面放几十条心电记录。
2. 写一个遍历脚本，逐条读取、预处理、推理。
3. 把结果汇总成CSV，记录文件名、label_id、置信度、耗时。

python复制import os
import csv
import time

input_dir = "./data/records"
output_csv = "./output/batch_result.csv"

results = []
for fname in os.listdir(input_dir):
    if not fname.endswith(".dat"):
        continue
    record_path = os.path.join(input_dir, fname)
    start = time.time()
    try:
        # 实际读取逻辑替换为符合文件格式的实现
        signal = load_ecg_record(record_path)
        result = predict_ecg(signal)
        elapsed = time.time() - start
        results.append({
            "file": fname,
            "label_id": result["label_id"],
            "confidence": result["confidence"],
            "elapsed": round(elapsed, 3),
        })
    except Exception as e:
        results.append({"file": fname, "error": str(e)})

with open(output_csv, "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["file", "label_id", "confidence", "elapsed", "error"])
    writer.writeheader()
    writer.writerows(results)

批量任务的判断标准不是“能不能跑”，而是“有没有一个文件拖垮整个任务”。实际部署时一定要做隔离：单条数据出错只记录error字段，不中断整体流程。上面代码里每个文件都有try-except，就是这个目的。

5.4 模型指标验证

单条测试通过不等于模型可用。正确做法是准备一个带真实标签的测试集，计算整体指标：

• Accuracy：整体准确率，但类别不平衡时需要结合其他指标看。
• Precision和Recall：重点关注灵敏度，对疾病筛检，“漏报”的代价通常高于“误报”。
• F1分数：精确率和召回率的调和平均。
• AUC：用于评估模型区分正负类的能力。

python复制from sklearn.metrics import classification_report, roc_auc_score

# y_true: 真实标签，y_pred: 预测类别，y_prob: 预测概率
print(classification_report(y_true, y_pred, target_names=LABELS))
auc = roc_auc_score(y_true, y_prob, multi_class="ovr")
print(f"Multi-class AUC: {auc:.4f}")

如果测试集上的指标和训练时差距很大，优先怀疑预处理不一致或数据分布差异，不要急着调模型。

6. 接口API与批量任务

6.1 接口设计

心电图AI服务最常见的API设计是：接收一段心电信号，返回异常标签和置信度。下面是一个示例请求结构：

json复制{
  "signal": [[0.01, 0.02, 0.01], [0.01, 0.02, 0.01]],
  "fs": 500,
  "channels": 12,
  "patient_id": "demo_001"
}

返回结果建议包含原始概率，而不是只返回最高类别，方便调用方自行决定阈值：

json复制{
  "patient_id": "demo_001",
  "label_id": 1,
  "label_name": "AFIB",
  "confidence": 0.93,
  "probabilities": {
    "NORMAL": 0.02,
    "AFIB": 0.93,
    "STTC": 0.05
  },
  "timestamp": "2025-01-01T12:00:00Z"
}

6.2 curl调用示例

bash复制curl -X POST http://127.0.0.1:8000/predict_array \
  -H "Content-Type: application/json" \
  -d '{
    "signal": [[0.01, 0.02, 0.01], [0.01, 0.02, 0.01]],
    "fs": 500
  }'

6.3 Python调用示例

python复制import requests

url = "http://127.0.0.1:8000/predict_array"
payload = {
    "signal": signal.tolist(),  # shape (leads, samples)
    "fs": 500,
}

resp = requests.post(url, json=payload, timeout=30)
resp.raise_for_status()
result = resp.json()

print("预测类别:", result["label_name"])
print("置信度:", result["confidence"])

6.4 批量任务队列

批量处理不推荐一次性把所有文件塞进API请求，内存和超时都会出问题。更稳妥的做法是设计一个简单的任务队列：

1. 输入目录存放待处理文件。
2. 任务脚本逐个读取文件，调用模型推理。
3. 每处理一条记录，把原始文件名、输出结果、耗时写入结果CSV，并立即保存。
4. 对失败任务记录错误原因，不中断整体队列。
5. 所有记录处理完后，生成汇总报告。

批量处理要关注三个问题。第一是内存：不要把整个目录的文件同时读入内存，按需加载，用完释放。第二是缓存：相同患者的多条重复记录可以跳过，避免重复推理。第三是断点续传：已经处理成功的记录写入done清单，程序中断后重新启动能跳过已处理文件。

6.5 手动调用测试

接口部署完成后，建议先用手动测试确认三个点。第一，正常请求能返回200和正确结构。第二，异常输入能返回明确错误信息，而不是直接服务崩溃。第三，并发请求下服务是否稳定。如果预期会有多个调用方，可以先用Stress测试工具做小规模并发验证，再决定是否增加排队和限流。

7. 资源占用与性能观察

题目里说的“AI不到2秒读出结果，对应的通常是单条10秒12导联记录在GPU上的推理耗时量级。CPU上能不能达到这个速度，取决于模型参数量、输入长度和是否做了量化，不能一概而论。部署前要做的不是问“能不能2秒”，而是用自己的模型和显卡实测一个延迟区间。

资源占用可以从三个层面观察：

• 显存：用nvidia-smi动态监测推理进程的Resource利用率。
• 内存：批量任务时用系统监视器观察Python进程占用的物理内存。
• 耗时：在脚本里对预处理、模型推理、后处理分别打时间戳，避免混在一起。

bash复制# 每2秒刷新一次GPU状态
nvidia-smi -l 2

影响性能的主要因素有四个。第一是输入序列长度，500Hz的10秒记录就是5000个采样点，12导联合计6万个浮点数，这个规模对模型来说不算大，但如果是60秒长记录，数据量会翻6倍。第二是batch size，批量推理时GPU利用率更高，但batch越大显存占用越高，要找到平衡点。第三是模型结构，纯CNN推理快但依赖感受野设计，带Transformer层的模型精度可能更高但更慢。第四是线程和优化，ONNX Runtime开启CPU线程数、GPU执行Provider、半精度推断，都能明显改变延迟。

降低延迟可以从这几个方向尝试：

python复制# ONNX Runtime GPU执行，按实际环境调整
sess_options = ort.SessionOptions()
sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
sess = ort.InferenceSession(
    "ecg_model.onnx",
    sess_options=sess_options,
    providers=["CUDAExecutionProvider", "CPUExecutionProvider"],
)

如果模型导出为FP16，GPU推理延迟通常会有明显下降，但精度可能有微弱损失。判断标准是在验证集上对比FP32和FP16的指标差异，差异可接受再切。

批量任务里的耗时统计要分开看：预处理耗时、推理耗时、IO耗时。很多心电图AI系统真正的瓶颈不在模型，而在磁盘和文件解析。从.dat文件读取心电信号的时间可能比模型推理还长，这时候要考虑把高频使用的数据预解析成numpy缓存格式，避免每次重复解析。

8. 常见问题与排查方法

心电图AI部署中最常见的坑集中在数据读取、预处理不一致、资源溢出三类。下面按现象列出排查思路：

依赖安装失败是另一个高频问题。torch和torchaudio版本对CUDA版本要求严格，建议先确认显卡驱动对应的CUDA版本，再选择合适的torch版本。如果只是做推理，不必安装完整torch，直接使用onnxruntime可以减少很多冲突。

9. 最佳实践与合规使用建议

如果是第一次部署心电图AI，建议按下面这套顺序走，能少踩很多坑。

第一，先准备一套小规模的验证数据，不要一开始就上全部数据。20到50条已知标签的心电记录就够，先把预处理和推理流程跑通。

第二，建立一套最小可运行的配置。把requirements.txt、推理脚本、模型文件路径、输入输出目录固定下来，写进项目README。这样换机器、换环境时能快速复现。

第三，模型文件、原始心电数据、输出结果分目录存放，不要混在同一层。建议结构如下：

text复制ecg_ai_project/
├── models/          # 模型权重和配置文件
├── data/
│   ├── raw_records/ # 原始心电记录
│   └── processed/   # 预处理后的numpy数组
├── scripts/         # 推理和批量处理脚本
├── outputs/         # 推理结果、日志、报告
└── requirements.txt

第四，批量任务必须写日志。每条记录处理完，除了写结果CSV，还要写一行结构化日志，包含文件名、开始时间、结束时间、耗时、结果摘要。任务中断时，日志能告诉你处理到哪一条、哪一条出了问题。

第五，接口服务要限制访问范围。如果只是内部工具，建议监听127.0.0.1，用Nginx或网关做转发；如果需要在局域网内提供API，至少要加简单的身份验证，避免任何设备都能直接调用。

第六，涉及医疗数据时，必须做去标识化处理。患者姓名、住院号、出生日期等信息不要和心电信号放在同一个文件里，文件名统一用脱敏ID。模型训练和测试集发布前，要按对应法规要求做合规审查。

第七，发布或商用前，要做效果复核。公开数据集上的成绩只代表历史验证结果，实际部署前需要用目标设备、目标人群、目标科室的数据做一次独立验证，并记录验证报告。模型输出只能辅助医生判读，不能直接替代诊断流程。

10. 总结与下一步

心电图AI的完整工作流其实没有太复杂的概念：读数据、预处理、推理、输出标签，难点都在“真实数据不听话”这件事上。导联顺序、采样率、信号长度、基线漂移，每一个都能让模型从“准确”变“不可用”。所以如果你准备上手，第一个要做的不是调模型结构，而是把数据读取和预处理这一层做扎实。

优先验证的最小闭环可以这样定义：准备10条已知标签的心电记录，跑通预处理和单条推理，再跑通目录批量生成CSV，然后封装成API。这一步跑通后，再考虑模型精度、并发性能、量化加速这些进阶内容。最容易踩的坑是拿着模型直接往真实数据上怼，遇到输出NaN或类别错乱就开始怀疑模型，结果最后发现是采样率没对齐。

后续可以扩展的方向也很多：把输出报告从简单标签扩展成带波形截图的PDF；把单模型推理改为多模型投票，提高心律失常分类稳定性；配合心电文件解析库，做成完整的“上传心电图-极速判读-可疑片段标注”工具链。如果这篇文章对你有帮助，建议收藏备用。