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

1. MLOps 流程

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

2. FastAPI 部署

fromfastapiimportFastAPIfrompydanticimportBaseModelimportpickleimportnumpyasnp app=FastAPI()withopen('model.pkl','rb')asf:model=pickle.load(f)classPredictRequest(BaseModel):features:listclassPredictResponse(BaseModel):prediction:floatprobability:float@app.post('/predict',response_model=PredictResponse)defpredict(request:PredictRequest):features=np.array(request.features).reshape(1,-1)prediction=model.predict(features)[0]probability=model.predict_proba(features).max()returnPredictResponse(prediction=float(prediction),probability=float(probability))@app.get('/health')defhealth():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. 模型监控

# 数据漂移检测fromscipy.statsimportks_2sampdefcheck_drift(reference_data,new_data,threshold=0.05):drift_detected=Falseforcolinreference_data.columns:stat,p_value=ks_2samp(reference_data[col],new_data[col])ifp_value<threshold:drift_detected=Trueprint(f"漂移检测:{col}(p={p_value:.4f})")returndrift_detected

总结