從筆記本到量產差距龐大。在醫療保健領域，這一差距還包括 HIPAA、審計日誌、臨床整合和 99.9% 的正常運作時間。
1. 醫療AI生產架構
┌─────────────────┐
DICOM images → │ DICOM Gateway │ ← Hospital PACS
│ (Orthanc) │
└────────┬────────┘
│ HL7 v2 / FHIR
▼
┌─────────────────┐
│ AI Engine │ ← Load balancer
│ (FastAPI + │
│ GPU server) │
└────────┬────────┘
│ Results
▼
┌─────────────────┐
│ Results Store │ ← PostgreSQL + MinIO
│ + Audit Log │
└────────┬────────┘
│
▼
┌─────────────────┐
│ RIS/EMR │ ← Radiologist workstation
│ Integration │
└─────────────────┘
2. 符合 HIPAA 的 FastAPI 伺服器
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import jwt
import logging
import uuid
import hashlib
from datetime import datetime, timezone
import os
app = FastAPI(
title="Medical AI API",
# Không expose docs publicly trong production
docs_url=None if os.getenv("ENV") == "production" else "/docs",
redoc_url=None if os.getenv("ENV") == "production" else "/redoc",
)
# HIPAA Audit logging: mỗi access PHI phải được log
class AuditLogger:
def __init__(self, db_connection):
self.db = db_connection
def log_access(
self,
user_id: str,
action: str, # READ, WRITE, PREDICT, EXPORT
resource_type: str, # PATIENT_IMAGE, PREDICTION, etc
resource_id: str, # Patient ID (de-identified)
ip_address: str,
success: bool,
details: dict = None
):
# KHÔNG log trực tiếp PHI (Protected Health Information)
# Log resource ID (anonymized) + action + user + timestamp
log_entry = {
"log_id": str(uuid.uuid4()),
"timestamp": datetime.now(timezone.utc).isoformat(),
"user_id": user_id,
"action": action,
"resource_type": resource_type,
"resource_id": self._hash_resource_id(resource_id), # Hash PHI
"ip_address": ip_address,
"success": success,
"details": details or {}
}
# self.db.insert("audit_logs", log_entry)
logging.getLogger("hipaa.audit").info(str(log_entry))
def _hash_resource_id(self, resource_id: str) -> str:
"""One-way hash: không thể reverse về patient ID từ log."""
return hashlib.sha256(
(resource_id + os.getenv("AUDIT_SALT", "")).encode()
).hexdigest()[:16]
# JWT Authentication
security = HTTPBearer()
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Verify JWT. Role-based access control."""
try:
payload = jwt.decode(
credentials.credentials,
os.getenv("JWT_SECRET"),
algorithms=["HS256"]
)
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expired")
except jwt.InvalidTokenError:
raise HTTPException(status_code=401, detail="Invalid token")
class PredictionRequest(BaseModel):
study_uid: str # DICOM Study Instance UID (pseudonym, not patient name)
modality: str # CR, CT, MR, US
requesting_physician: str
class PredictionResponse(BaseModel):
prediction_id: str
findings: list[dict]
confidence_scores: dict
model_version: str
inference_time_ms: float
disclaimer: str = (
"AI output is for decision support ONLY. "
"Final clinical decision must be made by licensed physician."
)
@app.post("/api/v1/predict", response_model=PredictionResponse)
async def predict(
request: Request,
body: PredictionRequest,
token: dict = Depends(verify_token)
):
# Check role
if "radiologist" not in token.get("roles", []) and "admin" not in token.get("roles", []):
raise HTTPException(status_code=403, detail="Insufficient permissions")
start_time = datetime.now(timezone.utc)
prediction_id = str(uuid.uuid4())
# Tải ảnh từ DICOM server
# image = load_from_dicom_server(body.study_uid)
# Inference
# findings = ai_model.predict(image)
findings = [{"finding": "Cardiomegaly", "confidence": 0.87, "region": "cardiac_silhouette"}]
inference_time = (datetime.now(timezone.utc) - start_time).total_seconds() * 1000
# Log access
# audit_logger.log_access(
# user_id=token["user_id"],
# action="PREDICT",
# resource_type="PATIENT_IMAGE",
# resource_id=body.study_uid,
# ip_address=request.client.host,
# success=True
# )
return PredictionResponse(
prediction_id=prediction_id,
findings=findings,
confidence_scores={"cardiomegaly": 0.87},
model_version=os.getenv("MODEL_VERSION", "1.0.0"),
inference_time_ms=round(inference_time, 1)
)
3. Docker 和 Kubernetes 部署
# Dockerfile
FROM nvidia/cuda:12.1-runtime-ubuntu22.04
# Non-root user (security best practice)
RUN groupadd -r medai && useradd -r -g medai medai
WORKDIR /app
# Install Python
RUN apt-get update && apt-get install -y python3.11 python3-pip --no-install-recommends && rm -rf /var/lib/apt/lists/*
# Dependencies
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt
# Copy app (không copy model weights vào image — mount via volume)
COPY --chown=medai:medai . .
USER medai
EXPOSE 8000
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 CMD curl -f http://localhost:8000/health || exit 1
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]
# kubernetes/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: medical-ai
namespace: healthcare-ai
spec:
replicas: 2
selector:
matchLabels:
app: medical-ai
template:
spec:
containers:
- name: medical-ai
image: registry.hospital.vn/medical-ai:1.2.0
resources:
limits:
nvidia.com/gpu: "1"
memory: "8Gi"
requests:
memory: "4Gi"
env:
- name: MODEL_VERSION
value: "1.2.0"
- name: JWT_SECRET
valueFrom:
secretKeyRef:
name: ai-secrets
key: jwt-secret
volumeMounts:
- name: model-weights
mountPath: /models
readOnly: true
volumes:
- name: model-weights
persistentVolumeClaim:
claimName: model-weights-pvc
4. 模型監控與漂移偵測
import numpy as np
from scipy import stats
class ModelMonitor:
"""Monitor model performance và data drift trong production."""
def detect_data_drift(
self,
reference_distribution: np.ndarray,
current_distribution: np.ndarray,
threshold: float = 0.05
) -> dict:
"""
KS Test: detect shift trong input distribution.
PSI (Population Stability Index): industry standard.
"""
ks_stat, ks_pvalue = stats.ks_2samp(reference_distribution, current_distribution)
# PSI: sum((actual% - expected%) * ln(actual%/expected%))
# PSI < 0.1: no shift
# 0.1-0.25: moderate shift
# > 0.25: significant shift — investigate
psi = self._calculate_psi(reference_distribution, current_distribution)
return {
"ks_statistic": round(float(ks_stat), 4),
"ks_pvalue": round(float(ks_pvalue), 4),
"drift_detected_ks": ks_pvalue < threshold,
"psi": round(float(psi), 4),
"psi_severity": (
"No shift" if psi < 0.1 else
"Moderate shift — monitor" if psi < 0.25 else
"Significant shift — ALERT"
)
}
def _calculate_psi(self, expected: np.ndarray, actual: np.ndarray, bins: int = 10) -> float:
"""Population Stability Index."""
expected_hist, bin_edges = np.histogram(expected, bins=bins)
actual_hist, _ = np.histogram(actual, bins=bin_edges)
expected_pct = expected_hist / expected_hist.sum() + 1e-8
actual_pct = actual_hist / actual_hist.sum() + 1e-8
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return psi
def check_prediction_drift(
self,
reference_preds: np.ndarray, # Confidence scores từ validation
current_preds: np.ndarray, # Last 7-day predictions
) -> dict:
"""Detect nếu model confidence distribution thay đổi."""
return self.detect_data_drift(reference_preds, current_preds)
5. 練習
-
為第 4 課中的胸部 X 光 AI 建立 FastAPI 服務。新增身份驗證、審核日誌記錄和速率限制。使用 Locust 進行測試（負載測試）以確保 < 2s response time.
-
實現漂移偵測管道：透過改變影像亮度分佈來模擬資料漂移。當PSI > 0.25時發出警報。
-
使用 HPA（Horizontal Pod Autoscaler）在 Kubernetes 上部署：當 CPU > 70% 時進行擴充。測量延遲 P50/P95/P99。
第 15 課：Capstone — 從頭到尾建構醫療人工智慧系統。