レッスン 15: Capstone — エンドツーエンドの医療 AI パイプラインの構築

医療とヘルスケアにおける AI: 実戦アプリケーション

パート 4: 生産とコンプライアンス

xdev.asia

これはシリーズの最後のレッスンです。レッスン 1 から 14 までのすべての知識を統合して、生の DICOM → 臨床 API に至る完全な医療 AI システムを構築します。

---

Capstone プロジェクト: 胸部 X 線 AI システム

目標: エンドツーエンドの胸部 X 線分析システムを構築する:

1. PACS から DICOM を取り込む (シミュレート)
2. 前処理（ウィンドウ処理、正規化）
3. 14 の病理を分類する (CheXpert)
4. Grad-CAM ヒートマップを生成する
5. HIPAA 準拠の REST API 経由でサービスを提供する
6. ドリフト検出による監視

スタック: PyTorch + FastAPI + Docker + MLflow

---

フェーズ 1: データ パイプライン

import pydicom
import numpy as np
import cv2
from pathlib import Path
import torch
from torchvision import transforms

class ChestXrayPipeline:
    """End-to-end pipeline từ DICOM → tensor cho inference."""
    
    # 14 CheXpert pathologies
    PATHOLOGIES = [
        "No Finding", "Enlarged Cardiomediastinum", "Cardiomegaly",
        "Lung Opacity", "Lung Lesion", "Edema", "Consolidation",
        "Pneumonia", "Atelectasis", "Pneumothorax", "Pleural Effusion",
        "Pleural Other", "Fracture", "Support Devices"
    ]
    
    def __init__(self):
        self.transform = transforms.Compose([
            transforms.ToPILImage(),
            transforms.Resize((320, 320)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                               std=[0.229, 0.224, 0.225])
        ])
    
    def load_dicom(self, dicom_path: str) -> np.ndarray:
        """Load DICOM và convert to normalized 8-bit grayscale."""
        ds = pydicom.dcmread(dicom_path)
        pixel_array = ds.pixel_array.astype(np.float32)
        
        # Apply rescale slope/intercept
        slope = float(getattr(ds, 'RescaleSlope', 1.0))
        intercept = float(getattr(ds, 'RescaleIntercept', 0.0))
        pixel_array = pixel_array * slope + intercept
        
        # Invert if MONOCHROME1
        photometric = getattr(ds, 'PhotometricInterpretation', 'MONOCHROME2')
        if photometric == 'MONOCHROME1':
            pixel_array = pixel_array.max() - pixel_array
        
        # Normalize to 0-255
        p2, p98 = np.percentile(pixel_array, [2, 98])
        pixel_array = np.clip(pixel_array, p2, p98)
        pixel_array = ((pixel_array - p2) / (p98 - p2) * 255).astype(np.uint8)
        
        # Apply CLAHE
        clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
        pixel_array = clahe.apply(pixel_array)
        
        # Convert to 3-channel
        image_rgb = cv2.cvtColor(pixel_array, cv2.COLOR_GRAY2RGB)
        return image_rgb
    
    def preprocess(self, image: np.ndarray) -> torch.Tensor:
        """Image → normalized tensor."""
        return self.transform(image).unsqueeze(0)

---

フェーズ 2: モデル アーキテクチャ

import torch
import torch.nn as nn
from torchvision.models import densenet121, DenseNet121_Weights

class CheXpertModel(nn.Module):
    """DenseNet-121 fine-tuned cho CheXpert 14 pathologies."""
    
    def __init__(self, n_classes: int = 14, pretrained: bool = True):
        super().__init__()
        
        weights = DenseNet121_Weights.IMAGENET1K_V1 if pretrained else None
        backbone = densenet121(weights=weights)
        
        # Freeze early layers
        layers = list(backbone.features.children())
        for layer in layers[:6]:
            for param in layer.parameters():
                param.requires_grad = False
        
        self.features = backbone.features
        
        in_features = backbone.classifier.in_features
        self.classifier = nn.Sequential(
            nn.AdaptiveAvgPool2d((1, 1)),
            nn.Flatten(),
            nn.Linear(in_features, 512),
            nn.BatchNorm1d(512),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(512, n_classes)
        )
        
        # Store last conv layer for Grad-CAM
        self.last_conv = self.features[-1]
        self._activations = None
        self._gradients = None
    
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        features = self.features(x)
        return self.classifier(features)
    
    def register_hooks(self):
        """Register hooks cho Grad-CAM."""
        def save_activation(module, input, output):
            self._activations = output
        
        def save_gradient(module, grad_input, grad_output):
            self._gradients = grad_output[0]
        
        self.last_conv.register_forward_hook(save_activation)
        self.last_conv.register_backward_hook(save_gradient)


def compute_gradcam(model, image_tensor, class_idx):
    """Generate Grad-CAM heatmap."""
    model.register_hooks()
    model.eval()
    
    output = model(image_tensor)
    model.zero_grad()
    output[0, class_idx].backward()
    
    activations = model._activations.detach()
    gradients = model._gradients.detach()
    
    weights = gradients.mean(dim=[2, 3], keepdim=True)
    cam = (weights * activations).sum(dim=1, keepdim=True)
    cam = torch.relu(cam).squeeze()
    
    # Normalize
    cam = cam - cam.min()
    cam = cam / (cam.max() + 1e-8)
    
    # Upsample
    H, W = image_tensor.shape[2:]
    cam_np = cam.cpu().numpy()
    cam_resized = cv2.resize(cam_np, (W, H))
    
    return cam_resized

---

フェーズ 3: トレーニング パイプライン

import mlflow
import mlflow.pytorch
from torch.cuda.amp import GradScaler, autocast

def train_chexpert(
    model: CheXpertModel,
    train_loader,
    val_loader,
    n_epochs: int = 30,
    lr: float = 1e-4,
    device: str = "cuda"
):
    """Production training với MLflow tracking."""
    
    with mlflow.start_run(run_name="chexpert-densenet121"):
        # Log hyperparameters
        mlflow.log_params({
            "model": "densenet121",
            "lr": lr,
            "n_epochs": n_epochs,
            "batch_size": train_loader.batch_size,
        })
        
        optimizer = torch.optim.AdamW(
            filter(lambda p: p.requires_grad, model.parameters()),
            lr=lr, weight_decay=1e-5
        )
        scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=n_epochs)
        scaler = GradScaler()
        
        # Weighted BCE: upweight positive cases
        pos_weights = compute_positive_weights(train_loader)
        criterion = nn.BCEWithLogitsLoss(pos_weight=pos_weights.to(device))
        
        best_val_auc = 0
        
        for epoch in range(n_epochs):
            model.train()
            train_losses = []
            
            for batch in train_loader:
                images = batch["image"].to(device)
                labels = batch["labels"].float().to(device)
                
                optimizer.zero_grad()
                with autocast():
                    logits = model(images)
                    loss = criterion(logits, labels)
                
                scaler.scale(loss).backward()
                scaler.unscale_(optimizer)
                nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
                scaler.step(optimizer)
                scaler.update()
                train_losses.append(loss.item())
            
            # Validation
            val_metrics = evaluate_chexpert(model, val_loader, device)
            scheduler.step()
            
            mlflow.log_metrics({
                "train_loss": np.mean(train_losses),
                "val_auc_mean": val_metrics["mean_auc"],
                **{f"val_auc_{p}": v for p, v in val_metrics["per_class_auc"].items()}
            }, step=epoch)
            
            # Save best model
            if val_metrics["mean_auc"] > best_val_auc:
                best_val_auc = val_metrics["mean_auc"]
                mlflow.pytorch.log_model(model, "best_model")
                print(f"Epoch {epoch}: New best AUC = {best_val_auc:.4f}")
        
        mlflow.log_metric("best_val_auc", best_val_auc)
        print(f"Training complete. Best val AUC: {best_val_auc:.4f}")


def compute_positive_weights(loader) -> torch.Tensor:
    """Compute class weights for imbalanced multilabel dataset."""
    label_counts = None
    total = 0
    for batch in loader:
        labels = batch["labels"].numpy()
        if label_counts is None:
            label_counts = labels.sum(axis=0)
        else:
            label_counts += labels.sum(axis=0)
        total += labels.shape[0]
    
    pos_weight = (total - label_counts) / (label_counts + 1e-6)
    return torch.tensor(pos_weight, dtype=torch.float32)

---

フェーズ 4: 評価

from sklearn.metrics import roc_auc_score, average_precision_score
import pandas as pd

def evaluate_chexpert(model, loader, device):
    model.eval()
    all_labels = []
    all_probs = []
    
    with torch.no_grad():
        for batch in loader:
            images = batch["image"].to(device)
            labels = batch["labels"].numpy()
            
            logits = model(images)
            probs = torch.sigmoid(logits).cpu().numpy()
            
            all_labels.append(labels)
            all_probs.append(probs)
    
    all_labels = np.concatenate(all_labels)
    all_probs = np.concatenate(all_probs)
    
    per_class_auc = {}
    for i, pathology in enumerate(ChestXrayPipeline.PATHOLOGIES):
        if all_labels[:, i].sum() > 0:
            auc = roc_auc_score(all_labels[:, i], all_probs[:, i])
            per_class_auc[pathology] = round(float(auc), 4)
    
    # Clinical metrics per pathology at threshold=0.5
    threshold = 0.5
    preds = (all_probs >= threshold).astype(int)
    
    clinical_report = []
    for i, pathology in enumerate(ChestXrayPipeline.PATHOLOGIES):
        if all_labels[:, i].sum() > 0:
            tp = ((preds[:, i] == 1) & (all_labels[:, i] == 1)).sum()
            tn = ((preds[:, i] == 0) & (all_labels[:, i] == 0)).sum()
            fp = ((preds[:, i] == 1) & (all_labels[:, i] == 0)).sum()
            fn = ((preds[:, i] == 0) & (all_labels[:, i] == 1)).sum()
            
            sensitivity = tp / (tp + fn + 1e-8)
            specificity = tn / (tn + fp + 1e-8)
            
            clinical_report.append({
                "pathology": pathology,
                "sensitivity": round(float(sensitivity), 4),
                "specificity": round(float(specificity), 4),
                "auc": per_class_auc.get(pathology, 0)
            })
    
    return {
        "per_class_auc": per_class_auc,
        "mean_auc": round(float(np.mean(list(per_class_auc.values()))), 4),
        "clinical_report": pd.DataFrame(clinical_report)
    }

---

フェーズ 5: API 統合 (レッスン 14 から)

from fastapi import FastAPI
import mlflow.pytorch
import io
from PIL import Image

app = FastAPI(title="CheXpert AI API v1.0")

# Load model từ MLflow
MODEL_URI = "runs:/best_run_id/best_model"
ai_model = mlflow.pytorch.load_model(MODEL_URI)
ai_model.eval()

pipeline = ChestXrayPipeline()

@app.post("/api/v1/analyze-xray")
async def analyze_xray(dicom_path: str, generate_heatmap: bool = True):
    """Analyze chest X-ray DICOM file."""
    # Load & preprocess
    image = pipeline.load_dicom(dicom_path)
    tensor = pipeline.preprocess(image).to("cuda")
    
    # Inference
    with torch.no_grad():
        logits = ai_model(tensor)
        probs = torch.sigmoid(logits).squeeze().cpu().numpy()
    
    # Format findings
    findings = []
    for i, (pathology, prob) in enumerate(zip(ChestXrayPipeline.PATHOLOGIES, probs)):
        if prob > 0.3:  # Threshold: tunable
            finding = {
                "pathology": pathology,
                "confidence": round(float(prob), 4),
                "severity": "high" if prob > 0.7 else "moderate" if prob > 0.5 else "low"
            }
            # Generate Grad-CAM for this finding
            if generate_heatmap:
                finding["heatmap"] = compute_gradcam(ai_model, tensor, i).tolist()
            findings.append(finding)
    
    return {
        "findings": sorted(findings, key=lambda x: x["confidence"], reverse=True),
        "total_pathologies_detected": len(findings),
        "disclaimer": "For clinical decision support only. Not for standalone diagnosis.",
    }

---

まとめシリーズ

おめでとうございます。AI in Health & Healthcare シリーズを完了しました。

学んだこと:

パート	トピックス	スキル
パート 1	プラットフォーム	DICOM、FHIR、医療データの前処理
パート 2	医用画像処理	CNN、U-Net、YOLO、WSI 分析
パート 3	臨床AI	NLP、創薬 GNN、ゲノミクス
パート 4	制作	Federated Learning、XAI、FDA、MLOps
キャップストーン	エンドツーエンド	DICOMからAPIまでのフルシステム

次のステップ:

1. Kaggle コンテスト: RSNA Pneumonia、PadChest、CheXpert
2. データセット: MIMIC-IV、PhysioNet、英国バイオバンク
3. カンファレンス: MICCAI、NeurIPS 医用画像ワークショップ
4. 認定: AWS Healthcare、GCP Healthcare API
5. オープンソース: monai、nnU-Net、medperf に貢献

---

最終的なキャップストーンの割り当て

最小要件 (完成したシリーズとみなされます):

1. 検証セットの平均 AUC ≥ 0.80 になるように CheXpert モデルをトレーニングします。
2. Docker を使用して API をデプロイします。 10 個の X 線 DICOM サンプルを使用してテストします。
3. 予測ごとに Grad-CAM を生成します。元の画像にオーバーレイします。
4. ドリフト検出を追加します。100 リクエスト後の予測信頼度分布と検証セットを比較します。
5. 使用目的、パフォーマンス指標、制限事項、倫理的考慮事項などの簡単なモデル カード (1 ページ) を作成します。

ボーナス (上級):

• ViT-B/16 (ビジョントランスフォーマー) で微調整します。 DenseNet-121と比較してください。
• フェデレーション バージョンの実装: CheXpert サブセットを使用して 3 つの病院をシミュレートします。
• 結果を CheXpert リーダーボードに送信します。