机器学习系统设计：从原型到生产

1. MLOps 流程

MLOps 全流程：
├── 数据管理
│   ├── 数据版本控制（DVC）
│   ├── 数据质量检查
│   └── 数据血缘追踪
├── 模型开发
│   ├── 实验跟踪（MLflow/W&B）
│   ├── 版本管理
│   └── 自动化训练
├── 模型部署
│   ├── 容器化（Docker）
│   ├── API 服务（FastAPI）
│   ├── A/B 测试
│   └── 灰度发布
└── 监控维护
    ├── 性能监控
    ├── 数据漂移检测
    └── 模型再训练

2. FastAPI 部署

from fastapi import FastAPI
from pydantic import BaseModel
import pickle
import numpy as np

app = FastAPI()

with open('model.pkl', 'rb') as f:
    model = pickle.load(f)

class PredictRequest(BaseModel):
    features: list

class PredictResponse(BaseModel):
    prediction: float
    probability: float

@app.post('/predict', response_model=PredictResponse)
def predict(request: PredictRequest):
    features = np.array(request.features).reshape(1, -1)
    prediction = model.predict(features)[0]
    probability = model.predict_proba(features).max()
    return PredictResponse(prediction=float(prediction), probability=float(probability))

@app.get('/health')
def health():
    return {'status': 'ok'}

3. Docker 部署

FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

4. 模型监控

# 数据漂移检测
from scipy.stats import ks_2samp

def check_drift(reference_data, new_data, threshold=0.05):
    drift_detected = False
    for col in reference_data.columns:
        stat, p_value = ks_2samp(reference_data[col], new_data[col])
        if p_value < threshold:
            drift_detected = True
            print(f"漂移检测: {col} (p={p_value:.4f})")
    return drift_detected

总结