基于 MONAI、FastAPI 与 3D Slicer 的医学影像自动化处理工作流实践

jaxzheng

154人浏览 · 2026-04-28 14:08:07

jaxzheng · 2026-04-28 14:08:07 发布

基于 MONAI、FastAPI 与 3D Slicer 的医学影像自动化处理工作流实践

在医学影像工程实践中，算法开发者经常在数据流转环节遇到阻碍。从 PACS 系统导出的 DICOM 序列往往伴随层厚不一、坐标系错乱等问题；同时，由于预处理脚本在训练与推理环境未实现强一致性，且 3D Slicer 难以直接加载未经物理空间对齐的预测数组，导致实验复现困难。为解决这些工程断层，本文将复盘如何整合 MONAI、FastAPI 与异步任务队列，构建端到端的处理原型。重点探讨工作流前移策略、预处理逻辑复用与结果回写机制，从而为临床科研提供稳定的工程基座。

核心环境与依赖推荐版本：

Python >= 3.9

MONAI == 1.3.0

FastAPI == 0.104.1

PyTorch == 2.1.0

pydicom == 2.4.3

临床数据处理痛点及解决方案

算法工程师通常需要投入大量精力处理数据准备工作。临床导出的 DICOM 序列常存在扫描参数不统一的情况，例如层厚差异、坐标系方向混杂（如 LPS 与 RAS 方向并存）以及单序列切片缺失等。若团队成员各自编写独立的脚本进行 NIfTI 格式转换，极易导致预处理逻辑在训练集构建和后续推理阶段产生分歧。

当模型生成预测 Mask 后，临床医生需要在 3D Slicer 中进行三维重建和评估。若工程师仅提供零散的 NumPy 数组或丢失了物理空间信息的 NIfTI 文件，结果将无法与原图精准叠加。这种数据链路的割裂直接拉低了自动化标注的实用价值。

要修复这一断层，需要将工作流前移，通过一个中心化的流水线统一接管数据解析、空间重采样与异步推理，并向临床工具提供携带标准空间坐标的返回结果。

架构设计：基于异步任务的端到端原型

为了保证数据处理链路的高可用与可扩展，工程部署上建议采用前后端解耦与异步任务队列结合的方案：

FastAPI 网关：作为外部调用的统一入口，接收 3D Slicer 插件传入的影像路径或直接上传的 DICOM 压缩包。
Celery + Redis：由于医学三维影像的重采样和推理属于 CPU/GPU 密集型的长耗时任务，通过 Celery 进行异步调度可避免网关阻塞，Redis 则用于状态追踪。
MONAI 预处理管线：将强度归一化、方向对齐等逻辑固化为声明式的 Transform 链，确保训练与推理环境的严格等价。
结果回写机制：推理结束后，服务将 Mask 逆向映射回原始物理空间，生成带标准 Affine 矩阵的 NRRD/NIfTI 文件，以支持 3D Slicer 原生加载。

核心实现：预处理复用与 Invertd 空间还原

以下核心代码展示了如何利用 MONAI 组装标准的预处理流水线，并将其嵌入 FastAPI 的异步端点中。重点在于使用 Invertd 变换替代粗粒度的重采样，以严谨还原物理空间信息。

import os
import torch
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from monai.transforms import (
    Compose, LoadImaged, EnsureChannelFirstd, Spacingd,
    Orientationd, ScaleIntensityRanged, SaveImaged,
    Invertd, AsDiscreted
)
from monai.data import Dataset, DataLoader
from monai.data.utils import decollate_batch

app = FastAPI(title="Medical Imaging AI Workflow")

# 定义严格复用的 MONAI 预处理管线
# 统一分辨率为 1.5x1.5x2.0 mm，并强制对齐到 RAS 坐标系
inference_transforms = Compose([
    LoadImaged(keys=["image"]),
    EnsureChannelFirstd(keys=["image"]),
    Spacingd(keys=["image"], pixdim=(1.5, 1.5, 2.0), mode="bilinear"),
    Orientationd(keys=["image"], axcodes="RAS"),
    ScaleIntensityRanged(
        keys=["image"], a_min=-175, a_max=250, 
        b_min=0.0, b_max=1.0, clip=True
    )
])

# 定义后处理与逆变换管线
post_transforms = Compose([
    EnsureChannelFirstd(keys=["pred"]),
    AsDiscreted(keys=["pred"], argmax=True),
    # 核心：使用 Invertd 安全反转 Spacingd 和 Orientationd 带来的空间变化
    Invertd(
        keys="pred",
        transform=inference_transforms,
        orig_keys="image",
        meta_keys="pred_meta_dict",
        orig_meta_keys="image_meta_dict",
        meta_key_postfix="meta_dict",
        nearest_interp=True,
        to_tensor=True,
    ),
    SaveImaged(
        keys="pred",
        meta_keys="pred_meta_dict",
        output_dir="outputs/",
        output_postfix="seg",
        resample=False, # 已通过 Invertd 精确还原，此处无需再设为 True
        separate_folder=False
    )
])

class InferenceRequest(BaseModel):
    task_id: str
    input_filepath: str

def process_and_infer(req: InferenceRequest):
    """后台异步处理：涵盖数据加载、推理与物理空间回写"""
    data_dict = [{"image": req.input_filepath}]
    ds = Dataset(data=data_dict, transform=inference_transforms)
    loader = DataLoader(ds, batch_size=1, num_workers=0)
    
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    
    for batch_data in loader:
        inputs = batch_data["image"].to(device)
        
        # 模拟生成预测 Mask，实际应用中替换为模型推理
        outputs = torch.randint(0, 2, inputs.shape).to(device)
        batch_data["pred"] = outputs
        
        # 将 batch 解包并应用后处理与逆变换
        for item in decollate_batch(batch_data):
            post_transforms(item)
            
    print(f"Task {req.task_id} completed.")

@app.post("/api/v1/segment")
async def trigger_segmentation(req: InferenceRequest, background_tasks: BackgroundTasks):
    if not os.path.exists(req.input_filepath):
        return {"status": "error", "message": "Input file not found."}
        
    background_tasks.add_task(process_and_infer, req)
    return {"status": "accepted", "task_id": req.task_id}

在医学影像处理中，直接使用 SaveImaged(resample=True) 虽能提供基础的重采样，但在复杂的临床坐标系中存在误差风险。代码中引入的 Invertd 变换是目前最安全的空间反转方案。Spacingd 和 Orientationd 在执行时，会将仿射矩阵的变更历史记录在 image_meta_dict 的 applied_operations 字段中。Invertd 通过读取这些记录，能够逆向计算出精确的变换矩阵，确保最终输出的预测 Mask 完全还原至输入 DICOM/NIfTI 的原始像素间距与方向轴，从根本上避免空间错位。

工程踩坑与物理空间对齐经验

在实际对接 3D Slicer 时，工程实施上需要规避以下常见问题：

首先是 DICOM 到 NIfTI 转换时的坐标系丢失。临床数据源常带有倾斜机架角度（Gantry Tilt），前期清洗时如果使用非标准工具直接提取像素数组，会破坏空间信息。建议使用 dcm2niix 引擎或通过 MONAI 的 LoadImaged 原生读取，并在 Transform 链的早期引入 Orientationd(axcodes="RAS")。由于 3D Slicer 默认使用 RAS 坐标系，保持这一层对齐可消除大部分由坐标轴翻转导致的掩模错位。

其次是多进程数据加载导致的内存泄漏。在处理高分辨率 CT 序列（如 512x512x500）时，若在 FastAPI 后台任务或 Celery Worker 中将 DataLoader 的 num_workers 设为大于 0，可能因 PyTorch 的共享内存限制引发崩溃。实际部署中，建议将推理 Worker 的并发度前置到 Celery 层面控制，并将 DataLoader 的 num_workers 设为 0，或在任务结束时显式执行系统级内存回收。

最后是结果交互闭环。在 3D Slicer 端，开发 Python 宏脚本通过 HTTP 轮询 FastAPI 的任务状态是一种可靠的解法。任务完成后，脚本调用 slicer.util.loadVolume() 即可将原图与携带标准 Affine 矩阵的预测 Mask 加载到同一视图，供临床医生直接可视化审查或微调。