MLOps实践¶
概述¶
MLOps是将机器学习模型从开发到生产部署的完整实践，包括模型训练、版本管理、部署、监控等环节。
MLOps流程¶
实验管理¶
MLflow¶
import mlflow
import mlflow.sklearn
# 开始实验
mlflow.set_experiment("my_experiment")
with mlflow.start_run():
# 记录参数
mlflow.log_param("n_estimators", 100)
mlflow.log_param("max_depth", 5)
# 训练模型
model = RandomForestClassifier(n_estimators=100, max_depth=5)
model.fit(X_train, y_train)
# 记录指标
accuracy = model.score(X_test, y_test)
mlflow.log_metric("accuracy", accuracy)
# 保存模型
mlflow.sklearn.log_model(model, "model")
Weights & Biases¶
import wandb
wandb.init(project="my_project")
# 记录参数
wandb.config.update({"lr": 0.001, "epochs": 10})
# 训练循环
for epoch in range(10):
train_loss = train_epoch(model)
val_loss = evaluate(model)
wandb.log({
"train_loss": train_loss,
"val_loss": val_loss,
"epoch": epoch
})
wandb.finish()
模型版本管理¶
# 使用DVC（Data Version Control）
# 初始化
dvc init
# 追踪数据
dvc add data/train.csv
# 追踪模型
dvc add model.pkl
# 提交
git add .
git commit -m "v1: initial model"
模型部署¶
Docker容器化¶
# Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
FastAPI服务¶
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
app = FastAPI()
model = joblib.load("model.pkl")
class PredictionRequest(BaseModel):
features: list
@app.post("/predict")
def predict(request: PredictionRequest):
features = [request.features]
prediction = model.predict(features)
return {"prediction": prediction.tolist()}
Kubernetes部署¶
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: ml-model
spec:
replicas: 3
selector:
matchLabels:
app: ml-model
template:
metadata:
labels:
app: ml-model
spec:
containers:
- name: model
image: my-model:latest
ports:
- containerPort: 8000
模型监控¶
# 数据漂移检测
from scipy import stats
import numpy as np
def detect_drift(reference_data, current_data, threshold=0.05):
drift_detected = []
for i in range(reference_data.shape[1]):
stat, p_value = stats.ks_2samp(
reference_data[:, i],
current_data[:, i]
)
if p_value < threshold:
drift_detected.append(i)
return drift_detected
# 性能监控
def monitor_performance(y_true, y_pred, threshold=0.8):
accuracy = np.mean(y_true == y_pred)
if accuracy < threshold:
send_alert(f"Model performance degraded: {accuracy:.2%}")
trigger_retraining()
CI/CD for ML¶
# .github/workflows/ml-pipeline.yml
name: ML Pipeline
on:
push:
branches: [main]
jobs:
train:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Train Model
run: |
python train.py
- name: Evaluate Model
run: |
python evaluate.py
- name: Deploy
if: success()
run: |
docker build -t my-model .
docker push my-model
最佳实践¶
# MLOps最佳实践
1. 代码版本化
- Git管理代码
- 实验追踪
2. 数据版本化
- DVC管理数据
- 数据血缘追踪
3. 自动化
- CI/CD管道
- 自动训练部署
4. 监控
- 性能监控
- 数据漂移检测
5. 文档
- 模型卡片
- API文档