医疗场景下的AI辅助诊断系统架构：从影像识别到报告生成的全链路实践

1. 引言与工程背景

医疗影像辅助诊断是AI落地的高价值场景。
CT、MRI、X光片日均产出数百万张。
影像科医师的阅片负荷持续攀升。
漏诊率与疲劳度正相关，这是客观规律。
AI辅助诊断系统旨在缓解这一矛盾。
它不是替代医师，而是提供第二意见。
本文从系统工程视角拆解完整链路。
从DICOM影像输入到诊断报告输出。
覆盖模型选型、推理服务、审核流程。
所有代码均面向生产环境设计。

2. 系统整体架构设计

2.1 全链路数据流

2.2 DICOM影像预处理管道

import pydicom
import numpy as np
from skimage.transform import resize
from skimage.filters import threshold_local

class DicomPreprocessor:
    """生产级DICOM影像预处理管道"""

    def __init__(self, target_size=(512, 512)):
        self.target_size = target_size
        self.valid_modality = {'CT', 'MR', 'XA', 'RF', 'CR'}

    def load_and_validate(self, dicom_path: str) -> dict:
        """加载DICOM文件并校验元信息"""
        ds = pydicom.dcmread(dicom_path)

        if ds.Modality not in self.valid_modality:
            raise ValueError(f"不支持的模态: {ds.Modality}")

        pixel_array = ds.pixel_array.astype(np.float32)

        # 窗宽窗位调整
        if hasattr(ds, 'WindowCenter') and hasattr(ds, 'WindowWidth'):
            wc = float(ds.WindowCenter)
            ww = float(ds.WindowWidth)
            pixel_array = np.clip(pixel_array, wc - ww/2, wc + ww/2)

        # 归一化到0-1范围
        pixel_array = (pixel_array - pixel_array.min()) / \
                      (pixel_array.max() - pixel_array.min() + 1e-8)

        return {
            "image": pixel_array,
            "modality": ds.Modality,
            "patient_id": ds.PatientID,
            "study_date": ds.StudyDate,
            "slice_thickness": getattr(ds, 'SliceThickness', None)
        }

    def preprocess(self, image: np.ndarray) -> np.ndarray:
        """标准化预处理流程"""
        # 统一分辨率
        image = resize(image, self.target_size,
                       anti_aliasing=True)

        # 局部自适应阈值增强对比度
        if image.ndim == 2:
            block_size = 31
            binary = threshold_local(image, block_size,
                                     method='gaussian')
            image = np.where(image > binary, image * 1.2, image * 0.8)

        # 维度标准化: (H, W) -> (1, H, W) 或 (H, W, C) -> (C, H, W)
        if image.ndim == 2:
            image = np.expand_dims(image, axis=0)
        elif image.ndim == 3:
            image = np.transpose(image, (2, 0, 1))

        return image.astype(np.float32)

3. 多模型并行推理引擎

3.1 推理服务架构

3.2 Triton推理服务配置

# model_config.pbtxt - Triton Inference Server配置
# 病灶检测模型配置示例

name: "lesion_detection_ct"
platform: "onnxruntime_onnx"
max_batch_size: 8

input [
  {
    name: "image"
    data_type: TYPE_FP32
    dims: [1, 512, 512]
  }
]

output [
  {
    name: "boxes"
    data_type: TYPE_FP32
    dims: [100, 5]  # [x1, y1, x2, y2, confidence]
  },
  {
    name: "labels"
    data_type: TYPE_INT32
    dims: [100]
  }
]

instance_group [
  {
    kind: KIND_GPU
    count: 2
    gpus: [0, 1]
  }
]

dynamic_batching {
  preferred_batch_size: [4, 8]
  max_queue_delay_microseconds: 5000
}

3.3 异步推理调度器

import asyncio
import tritonclient.grpc as grpc_client
from typing import List, Dict

class InferenceScheduler:
    """多模型并行推理调度器"""

    def __init__(self, triton_url="localhost:8001"):
        self.client = grpc_client.InferenceServerClient(
            url=triton_url)
        self.model_map = {
            "CT": ["lesion_detection_ct",
                    "classification_ct",
                    "segmentation_ct"],
            "MR": ["lesion_detection_mr",
                    "classification_mr"],
            "CR": ["lesion_detection_cr",
                    "classification_cr"]
        }

    async def schedule_parallel(self,
                                 image: np.ndarray,
                                 modality: str) -> Dict:
        """并行调度同一模态下的所有模型"""
        models = self.model_map.get(modality, [])
        tasks = []

        for model_name in models:
            task = self._infer_single(model_name, image)
            tasks.append(task)

        results = await asyncio.gather(*tasks,
                                       return_exceptions=True)

        # 过滤异常结果
        valid_results = {}
        for model_name, result in zip(models, results):
            if not isinstance(result, Exception):
                valid_results[model_name] = result

        return self._aggregate_results(valid_results)

    async def _infer_single(self, model_name: str,
                            image: np.ndarray) -> Dict:
        """单模型推理调用"""
        inputs = grpc_client.InferInput("image",
                                        [1, 512, 512], "FP32")
        inputs.set_data_from_numpy(image)

        outputs = [
            grpc_client.InferRequestedOutput("boxes"),
            grpc_client.InferRequestedOutput("labels")
        ]

        response = await self.client.async_infer(
            model_name=model_name,
            inputs=[inputs],
            outputs=outputs
        )

        return {
            "boxes": response.as_numpy("boxes"),
            "labels": response.as_numpy("labels")
        }

4. 诊断报告生成引擎

4.1 结构化特征提取

from dataclasses import dataclass
from typing import List, Optional

@dataclass
class LesionFeature:
    """病灶结构化特征"""
    lesion_type: str
    location: str
    size_mm: float
    shape: str
    boundary: str
    density_signal: str
    confidence: float
    risk_level: str

class FeatureExtractor:
    """从推理结果提取结构化特征"""

    def extract(self, inference_results: Dict,
                dicom_meta: Dict) -> List[LesionFeature]:
        """特征提取与标准化"""
        features = []

        detection = inference_results.get(
            "lesion_detection_ct", {})
        boxes = detection.get("boxes", [])
        labels = detection.get("labels", [])

        segmentation = inference_results.get(
            "segmentation_ct", {})
        mask = segmentation.get("mask", None)

        classification = inference_results.get(
            "classification_ct", {})
        class_probs = classification.get("probs", [])

        for i, (box, label) in enumerate(zip(boxes, labels)):
            if box[4] < 0.3:  # 置信度阈值过滤
                continue

            # 像素坐标转物理坐标
            pixel_spacing = dicom_meta.get(
                "pixel_spacing", [0.5, 0.5])
            size_mm = max(
                (box[2]-box[1]) * pixel_spacing[0],
                (box[3]-box[0]) * pixel_spacing[1]
            )

            feature = LesionFeature(
                lesion_type=self._label_to_type(label),
                location=self._bbox_to_location(box),
                size_mm=round(size_mm, 1),
                shape=self._analyze_shape(mask, i),
                boundary=self._analyze_boundary(mask, i),
                density_signal=self._analyze_density(mask, i),
                confidence=round(box[4], 3),
                risk_level=self._calc_risk(
                    size_mm, box[4], label)
            )
            features.append(feature)

        return features

4.2 报告模板匹配与生成

from jinja2 import Template
import json

REPORT_TEMPLATE = Template("""
影像检查报告
========================
检查类型: {{ modality }}
检查日期: {{ study_date }}
患者编号: {{ patient_id }}

影像所见:
{% for lesion in lesions %}
- {{ lesion.location }}见{{ lesion.lestion_type }}，
  大小约{{ lesion.size_mm }}mm，
  形态{{ lesion.shape }}，
  边界{{ lesion.boundary }}，
  {{ lesion.density_signal }}特征。
  (AI置信度: {{ lesion.confidence }},
   风险等级: {{ lesion.risk_level }})
{% endfor %}

{% if not lesions %}
未见明显异常征象。
{% endif %}

AI辅助提示:
本报告由AI辅助诊断系统生成，
仅供医师参考，不作为最终诊断依据。
需经执业医师审核签发后方可生效。
""")

class ReportGenerator:
    """诊断报告生成器"""

    def generate(self, features: List[LesionFeature],
                 dicom_meta: Dict) -> str:
        """生成结构化诊断报告"""
        lesions = [
            {
                "lesion_type": f.lestion_type,
                "location": f.location,
                "size_mm": f.size_mm,
                "shape": f.shape,
                "boundary": f.boundary,
                "density_signal": f.density_signal,
                "confidence": f.confidence,
                "risk_level": f.risk_level
            }
            for f in features
        ]

        report = REPORT_TEMPLATE.render(
            modality=dicom_meta["modality"],
            study_date=dicom_meta["study_date"],
            patient_id=dicom_meta["patient_id"],
            lesions=lesions
        )

        # 输出结构化JSON供下游系统使用
        structured_output = {
            "report_text": report,
            "lesions": lesions,
            "meta": dicom_meta,
            "ai_version": "v2.3.1",
            "generated_at": datetime.utcnow().isoformat()
        }

        return json.dumps(structured_output,
                          ensure_ascii=False)

5. 合规与部署实践

5.1 医师审核与签发流程

5.2 数据合规与隐私保护

from cryptography.fernet import Fernet
import hashlib

class MedicalDataCompliance:
    """医疗数据合规处理"""

    def __init__(self, encryption_key: bytes):
        self.fernet = Fernet(encryption_key)

    def anonymize_dicom(self, ds: pydicom.Dataset) \
            -> pydicom.Dataset:
        """DICOM脱敏处理"""
        sensitive_tags = [
            (0x0010, 0x0010),  # PatientName
            (0x0010, 0x0020),  # PatientID
            (0x0010, 0x0030),  # PatientBirthDate
            (0x0010, 0x1040),  # PatientAddress
        ]

        for group, element in sensitive_tags:
            if (group, element) in ds:
                original = ds[group, element].value
                # 哈希替代原始值
                hashed = hashlib.sha256(
                    str(original).encode()
                ).hexdigest()[:8]
                ds[group, element].value = f"ANON_{hashed}"

        return ds

    def encrypt_report(self, report: str) -> bytes:
        """报告加密存储"""
        return self.fernet.encrypt(
            report.encode('utf-8'))

    def audit_log(self, action: str,
                  operator: str,
                  patient_hash: str):
        """审计日志记录"""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "action": action,
            "operator": operator,
            "patient_hash": patient_hash,
            "ip_source": self._get_client_ip()
        }
        # 写入不可篡改的审计日志存储
        self._write_to_immutable_store(log_entry)

5.3 部署监控与SLA保障

生产环境关键指标：

• 推理延迟P99 < 500ms
• 单日阅片吞吐 > 5000例
• 漏检率 < 人工基线的5%
• 服务可用性 > 99.95%

监控维度覆盖模型精度漂移、
推理服务QPS与延迟、
GPU资源利用率、
医师修正率趋势。
模型每周在验证集上重评估。
精度下降超过2%触发自动告警。

核心要点

1. DICOM预处理是基石：窗宽窗位、归一化、分辨率标准化直接影响下游模型精度，必须严格校验元信息合法性
2. 多模型并行推理提效：病灶检测+分类+分割三模型并行，Triton动态批处理提升GPU利用率，模态路由避免无效推理
3. 结构化特征桥接AI与报告：像素坐标转物理坐标、置信度阈值过滤、风险等级计算，将模型输出转化为医师可理解的结构化描述
4. 医师审核是合规红线：AI报告不可直接签发，必须经执业医师审核修正，修正标注回流训练数据形成闭环
5. 数据脱敏与审计不可缺失：DICOM敏感字段哈希替代、报告加密存储、操作审计日志不可篡改，满足医疗数据合规要求