11 - MLOps与模型部署¶

import joblib
from sklearn.ensemble import RandomForestClassifier

# 训练模型
model = RandomForestClassifier()
model.fit(X_train, y_train)

# 保存
joblib.dump(model, 'model.joblib')

# 加载
loaded_model = joblib.load('model.joblib')
predictions = loaded_model.predict(X_test)

import torch

# 保存整个模型
torch.save(model, 'model.pth')

# 保存模型参数 (推荐)
torch.save(model.state_dict(), 'model_weights.pth')

# 加载
model = MyModelClass()
model.load_state_dict(torch.load('model_weights.pth', weights_only=True))
model.eval()  # 设置为评估模式

import torch.onnx

# 导出为ONNX
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(
    model,
    dummy_input,
    'model.onnx',
    input_names=['input'],
    output_names=['output'],
    dynamic_axes={'input': {0: 'batch_size'},
                 'output': {0: 'batch_size'}}
)

# 加载并运行
import onnxruntime as ort
session = ort.InferenceSession('model.onnx')
outputs = session.run(None, {'input': dummy_input.numpy()})

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np
import pandas as pd

app = FastAPI(title="ML Model API")

# 加载模型
model = joblib.load('model.joblib')
scaler = joblib.load('scaler.joblib')

# 定义输入数据格式
class PredictionRequest(BaseModel):
    features: list[float]

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

@app.get("/")
def home():
    return {"message": "ML Model API is running"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

@app.post("/predict", response_model=PredictionResponse)
def predict(request: PredictionRequest):
    try:  # try/except捕获异常，防止程序崩溃
        # 转换输入
        features = np.array(request.features).reshape(1, -1)

        # 预处理
        features_scaled = scaler.transform(features)

        # 预测
        prediction = model.predict(features_scaled)[0]
        probability = model.predict_proba(features_scaled)[0].max()

        return PredictionResponse(
            prediction=float(prediction),
            probability=float(probability)
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# 批量预测
@app.post("/predict_batch")
def predict_batch(requests: list[PredictionRequest]):
    results = []
    for req in requests:
        features = np.array(req.features).reshape(1, -1)
        features_scaled = scaler.transform(features)
        prediction = model.predict(features_scaled)[0]
        results.append({"prediction": int(prediction)})
    return {"results": results}

import boto3
import sagemaker
from sagemaker.sklearn.model import SKLearnModel

# 创建SageMaker会话
sagemaker_session = sagemaker.Session()
role = "arn:aws:iam::account-id:role/your-role"

# 上传模型
model_data = sagemaker_session.upload_data(
    path="model.joblib",
    bucket="your-bucket",
    key_prefix="model"
)

# 部署模型
sklearn_model = SKLearnModel(
    model_data=model_data,
    role=role,
    entry_point="inference.py",
    framework_version="1.0-1"
)

predictor = sklearn_model.deploy(
    initial_instance_count=1,
    instance_type="ml.m5.large"
)

# 预测
result = predictor.predict(X_test)

from azureml.core import Workspace, Model, Webservice
from azureml.core.webservice import AciWebservice
from azureml.core.model import InferenceConfig

# 连接工作区
ws = Workspace.from_config()

# 注册模型
model = Model.register(
    workspace=ws,
    model_path="model.joblib",
    model_name="my-model"
)

# 配置推理
inference_config = InferenceConfig(
    entry_script="score.py",
    environment=my_env
)

# 部署
deployment_config = AciWebservice.deploy_configuration(
    cpu_cores=1,
    memory_gb=4
)

service = Model.deploy(
    ws,
    "myservice",
    [model],
    inference_config,
    deployment_config
)

service.wait_for_deployment(show_output=True)

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 定义指标
prediction_counter = Counter(
    'predictions_total',
    'Total number of predictions'
)

prediction_latency = Histogram(
    'prediction_latency_seconds',
    'Prediction latency'
)

model_confidence = Gauge(
    'model_confidence',
    'Model prediction confidence'
)

data_drift_score = Gauge(
    'data_drift_score',
    'Data drift detection score'
)

# 在预测时记录
import time

def predict_with_monitoring(features):
    start_time = time.time()

    # 预测
    result = model.predict(features)

    # 记录指标
    prediction_counter.inc()
    latency = time.time() - start_time
    prediction_latency.observe(latency)

    if hasattr(result, 'max'):  # hasattr检查对象是否有指定属性
        model_confidence.set(result.max())

    return result

# 启动监控服务器
start_http_server(8001)

from alibi_detect.cd import ChiSquareDrift, KSDrift

# 准备参考数据和当前数据
X_ref = X_train  # 训练数据作为参考
X_current = X_test  # 当前数据

# 检测分类变量漂移
dd = ChiSquareDrift(
    X_ref,
    p_val=0.05  # 显著性水平
)

labels = dd.predict(X_current)
print(f"漂移检测结果: {labels['data']['is_drift']}")

# 检测连续变量漂移（Kolmogorov-Smirnov检验）
cd = KSDrift(X_ref, p_val=0.05)
labels = cd.predict(X_current)

import mlflow

def log_metrics(model, X_test, y_test):
    y_pred = model.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)
    precision = precision_score(y_test, y_pred)
    recall = recall_score(y_test, y_pred)
    f1 = f1_score(y_test, y_pred)

    # 记录到MLflow
    mlflow.log_metrics({
        "accuracy": accuracy,
        "precision": precision,
        "recall": recall,
        "f1_score": f1
    })

    return {
        "accuracy": accuracy,
        "precision": precision,
        "recall": recall,
        "f1": f1
    }

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier

# 启动MLflow UI
# mlflow ui

# 设置实验
mlflow.set_experiment("My Experiment")

# 训练并记录
with mlflow.start_run():
    # 记录参数
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", 10)
    mlflow.log_param("learning_rate", 0.1)

    # 训练模型
    model = RandomForestClassifier(
        n_estimators=100,
        max_depth=10
    )
    model.fit(X_train, y_train)

    # 记录指标
    accuracy = model.score(X_test, y_test)
    mlflow.log_metric("accuracy", accuracy)

    # 记录模型
    mlflow.sklearn.log_model(model, "model")

    # 记录artifacts
    mlflow.log_artifact("config.json")

# 查看实验
# mlflow ui
# 访问 http://localhost:5000

from fastapi import FastAPI
import random

app = FastAPI()

# 加载不同版本
model_v1 = joblib.load("model_v1.joblib")
model_v2 = joblib.load("model_v2.joblib")

@app.post("/predict")
def predict(features: list[float]):
    # A/B测试: 50%流量到新模型
    if random.random() < 0.5:
        model = model_v1
        version = "v1"
    else:
        model = model_v2
        version = "v2"

    prediction = model.predict([features])[0]

    # 记录版本和结果
    log_prediction(version, features, prediction)

    return {"prediction": prediction, "model_version": version}

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):  # async def定义异步函数
    token = credentials.credentials
    if token != "your-secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )
    return token

@app.post("/predict")
async def predict(
    request: PredictionRequest,
    token: str = Depends(verify_token)
):
    # 预测逻辑
    pass

from pydantic import BaseModel, field_validator

class PredictionRequest(BaseModel):
    age: int
    income: float
    score: float

    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v < 0 or v > 120:
            raise ValueError('Age must be between 0 and 120')
        return v

    @field_validator('income')
    @classmethod
    def validate_income(cls, v):
        if v < 0:
            raise ValueError('Income cannot be negative')
        return v

@app.post("/predict_batch")
async def predict_batch(requests: list[PredictionRequest]):
    # 批量处理,提高吞吐量
    features_batch = np.array([req.features for req in requests])

    # 批量预测
    predictions_batch = model.predict(features_batch)

    return [
        {"prediction": int(pred)}
        for pred in predictions_batch
    ]

from functools import lru_cache
from fastapi import FastAPI

app = FastAPI()

@lru_cache(maxsize=1000)
def get_prediction(features_tuple):
    # 转换为tuple以支持缓存
    features = list(features_tuple)
    return model.predict([features])[0]

@app.post("/predict")
async def predict(request: PredictionRequest):
    features_tuple = tuple(request.features)
    prediction = get_prediction(features_tuple)
    return {"prediction": prediction}