MLOps流水线折腾手记

MLOps MLflow CI/CD
作者: 醉月思 发布日期: 2024-05-23 分类: 技术实践

去年这个时候，我们的模型部署状态大概是这样：三个模型分别跑在三个不同人的本地环境里，版本管理靠文件名后缀（、、），发布靠手动 rsync，出问题靠群里问谁改了代码。

MLOps 工具很多，但真正要选的时候得看约束条件：

  1. 团队规模不算大，上 Kubeflow 这种重型平台有点杀鸡用牛刀
  2. 模型数量目前只有几个，但未来可能会到十几个
  3. 既要能追踪实验，也要能管模型版本，最好还能自动部署
  4. 资源有限，不想维护一堆组件

最终选了 MLflow 做实验和模型管理，GitHub Actions 做 CI/CD，K8s 做部署。

问题的起点

去年这个时候，我们的模型部署状态大概是这样：三个模型分别跑在三个不同人的本地环境里，版本管理靠文件名后缀（v1_final.pkl、v1_final_real.pkl、v1_final_real_really_final.pkl），发布靠手动 rsync，出问题靠群里问谁改了代码。

第一次线上事故是因为推理环境和训练环境不一致导致的：训练用的 scikit-learn 版本是 1.3.2，线上部署的是 1.1.2，某个参数的默认值变了，预测结果全部偏了 15%。从发现问题到回滚用了四个小时，凌晨三点才把业务稳定下来。

那一刻我就知道，不能再这么干了。

先说选型

MLOps 工具很多，但真正要选的时候得看约束条件：

  1. 团队规模不算大，上 Kubeflow 这种重型平台有点杀鸡用牛刀
  2. 模型数量目前只有几个，但未来可能会到十几个
  3. 既要能追踪实验，也要能管模型版本，最好还能自动部署
  4. 资源有限，不想维护一堆组件

最终选了 MLflow 做实验和模型管理，GitHub Actions 做 CI/CD，K8s 做部署。这个组合不算最时髦，但胜在轻量、稳定，每个组件都能单独理解和替换。

搭建 MLflow 实验追踪

MLflow 是第一步，先把实验记录从 Jupyter 的魔法方法里拉出来。

初始化 Tracking Server

# 本地快速启动（适合个人开发）
mlflow ui --port 5000

# 生产环境建议用 PostgreSQL + S3/MinIO
# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:14
    environment:
      POSTGRES_DB: mlflow
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: yourpassword
    volumes:
      - postgres_data:/var/lib/postgresql/data

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.10.0
    ports:
      - "5000:5000"
    command: >
      mlflow server
      --backend-store-uri postgresql://mlflow:yourpassword@postgres:5432/mlflow
      --default-artifact-root s3://mlflow-artifacts/
      --host 0.0.0.0
    environment:
      AWS_ACCESS_KEY_ID: your_access_key
      AWS_SECRET_ACCESS_KEY: your_secret_key
      MLFLOW_S3_ENDPOINT_URL: http://minio:9000
    depends_on:
      - postgres

  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"
    command: server /data
    environment:
      MINIO_ROOT_USER: your_access_key
      MINIO_ROOT_PASSWORD: your_secret_key

这段配置踩过一个坑：S3 endpoint URL 环境变量的名字在早期版本里是 MLFLOW_S3_ENDPOINT_URL，但文档里有时候写成了 AWS_ENDPOINT_URL，导致第一次启动时一直连不上。看源码才确认是前者。

在训练代码里集成 MLflow

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, r2_score

mlflow.set_tracking_uri("http://your-mlflow-server:5000")
mlflow.set_experiment("house_price_prediction")

# 数据准备
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

with mlflow.start_run():
    # 记录超参数
    params = {
        "n_estimators": 200,
        "max_depth": 10,
        "random_state": 42
    }
    mlflow.log_params(params)

    # 训练模型
    model = RandomForestRegressor(**params)
    model.fit(X_train, y_train)

    # 评估
    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    r2 = r2_score(y_test, y_pred)

    # 记录指标
    mlflow.log_metrics({
        "mse": mse,
        "r2": r2
    })

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

    # 记录额外信息
    mlflow.log_text(
        "训练数据日期：2024-03-15\n数据来源：sales_db_v2",
        "dataset_info.txt"
    )

这里有个踩坑点：mlflow.log_model 默认会把整个模型序列化，但 PyTorch 模型比较大时，有时候会超时。遇到过一次 2GB 的模型上传失败，后来改成只存模型权重和模型架构分开处理：

# 针对 PyTorch 大模型
mlflow.pytorch.log_model(model, "model")
torch.save(model.state_dict(), "model_weights.pth")
mlflow.log_artifact("model_weights.pth")

模型版本管理

有了实验记录，下一步是管模型版本。MLflow 的 Model Registry 做这个事。

注册模型

# 在训练完成后注册模型
model_uri = f"runs:/{run.info.run_id}/model"
mlflow.register_model(model_uri, "house_price_predictor")

设置模型生命周期

from mlflow import MlflowClient

client = MlflowClient("http://your-mlflow-server:5000")

# 给模型打标签
client.set_model_version_tag(
    name="house_price_predictor",
    version=1,
    key="performance_tier",
    value="baseline"
)

# 把某个版本设为 Staging
client.transition_model_version_stage(
    name="house_price_predictor",
    version=2,
    stage="Staging"
)

# 线上出问题时回滚
client.transition_model_version_stage(
    name="house_price_predictor",
    version=1,
    stage="Production"
)

Model Registry 有个不直观的地方：默认是全局权限控制，没有细粒度的 ACL。团队里如果有人误删了 Production 模型，会很麻烦。我们的做法是配合 Git 仓库的代码审查，删除操作需要两个人确认。

搭建 CI/CD 流水线

模型版本有了，下一步是自动部署。用 GitHub Actions 做了一个完整的流水线。

流水线结构

# .github/workflows/mlops-pipeline.yml
name: MLOps Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  workflow_dispatch:
    inputs:
      deploy_to_production:
        description: "Deploy to production?"
        required: true
        default: false
        type: boolean

jobs:
  train-and-evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install mlflow

      - name: Train model
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
        run: |
          python train_model.py

      - name: Run tests
        run: |
          pytest tests/

      - name: Register model if metrics improve
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
        run: |
          python scripts/register_if_improved.py

  deploy-staging:
    needs: train-and-evaluate
    runs-on: ubuntu-latest
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4

      - name: Deploy to Staging
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
          KUBECONFIG: ${{ secrets.KUBECONFIG_STAGING }}
        run: |
          kubectl set image deployment/model-serving \
            serving=your-registry/house-price-predictor:staging \
            -n staging

模型服务化

服务化这块我们用了 MLflow Model Server 加个自制的 FastAPI 包装器：

# app.py
from fastapi import FastAPI, HTTPException
from mlflow.pyfunc import load_model
import numpy as np

app = FastAPI()

# 启动时加载模型
model = load_model("models:/house_price_predictor/Production")

@app.post("/predict")
async def predict(features: dict):
    try:
        # 转换输入格式
        X = np.array([[
            features["area"],
            features["bedrooms"],
            features["bathrooms"],
            features["age"]
        ]])

        prediction = model.predict(X)
        return {"prediction": float(prediction[0])}

    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy", "model_version": "Production"}

Dockerfile：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

# 运行时只下载 Production 模型
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

K8s 部署配置

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-serving
  namespace: staging
spec:
  replicas: 2
  selector:
    matchLabels:
      app: model-serving
  template:
    metadata:
      labels:
        app: model-serving
        model: house-price-predictor
        version: "{{ .Values.model.version }}"
    spec:
      containers:
      - name: serving
        image: your-registry/house-price-predictor:{{ .Values.image.tag }}
        ports:
        - containerPort: 8000
        env:
        - name: MLFLOW_TRACKING_URI
          value: "{{ .Values.mlflow.trackingUri }}"
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5

这个配置踩过的坑：模型刚启动时还在加载，livenessProbe 马上开始检查会导致容器重启循环。initialDelaySeconds 必须根据模型大小调整，我们的大模型需要 45 秒才能加载完成。

持续训练

模型上线不是终点，数据会漂移，性能会退化。我们做了两件事：

自动化性能监控

# monitor.py
import mlflow
from mlflow.tracking import MlflowClient
import numpy as np
from datetime import datetime, timedelta

client = MlflowClient()

def check_model_performance(model_name: str, threshold: float = 0.05):
    """检查模型是否出现性能退化"""
    # 获取当前 Production 模型
    prod_version = client.get_latest_versions(
        model_name,
        stages=["Production"]
    )[0]

    # 获取最近 7 天的线上预测误差
    seven_days_ago = datetime.now() - timedelta(days=7)
    recent_errors = []

    # 这里假设你有把线上预测误差记录到某个地方
    # 实际中可以连数据库、日志系统或专门的监控平台
    for log in get_recent_prediction_logs(seven_days_ago):
        actual = log["actual_value"]
        predicted = log["predicted_value"]
        error = abs(actual - predicted) / actual
        recent_errors.append(error)

    if not recent_errors:
        print("没有足够的线上数据，无法判断")
        return

    # 计算平均误差
    current_mae = np.mean(recent_errors)

    # 和注册时的基准误差对比
    baseline_mae = float(
        client.get_model_version(
            model_name,
            prod_version.version
        ).tags.get("baseline_mae", 0)
    )

    degradation = (current_mae - baseline_mae) / baseline_mae

    if degradation > threshold:
        print(f"⚠️  模型性能退化 {degradation:.2%}，建议重新训练")
        # 可以在这里触发重新训练的 pipeline
    else:
        print(f"✅ 模型性能正常，误差退化 {degradation:.2%}")

check_model_performance("house_price_predictor")

数据漂移检测

# drift_detection.py
from scipy import stats
import pandas as pd

def detect_drift(new_data: pd.DataFrame, baseline_stats: dict):
    """检测特征分布是否发生漂移"""
    drift_detected = False

    for feature in baseline_stats.keys():
        new_mean = new_data[feature].mean()
        new_std = new_data[feature].std()

        baseline_mean = baseline_stats[feature]["mean"]
        baseline_std = baseline_stats[feature]["std"]

        # 使用 KS 检验
        _, p_value = stats.ks_2samp(
            new_data[feature],
            # 这里简化了，实际应该有 baseline 数据
            np.random.normal(baseline_mean, baseline_std, len(new_data))
        )

        if p_value < 0.05:
            print(f"⚠️  特征 {feature} 可能发生漂移 (p={p_value:.4f})")
            drift_detected = True

    return drift_detected

踩过的主要坑

1. 模型加载时机

一开始把模型加载放在 FastAPI 请求处理函数里，每次预测都重新加载。大模型的话，第一次请求可能要等几十秒，用户直接以为服务挂了。

解决：应用启动时加载，用全局变量持有模型对象。但要考虑多线程安全，模型本身一般不可变，但如果有缓存需要注意。

2. 环境隔离

训练时用的 pandas 版本比服务新，导致模型序列化/反序列化时出错。特征列顺序和类型不完全一致也会导致预测失败。

解决：

  • 把依赖版本固定在 requirements.txt 里
  • 训练和服务都用同一个 Docker 基础镜像
  • 模型注册时记录特征定义，服务加载时校验
# 训练时记录特征定义
feature_schema = {
    "area": {"type": "float", "range": [0, 10000]},
    "bedrooms": {"type": "int", "range": [0, 20]}
}
mlflow.log_dict(feature_schema, "feature_schema.json")

# 服务加载时校验
def validate_input(features: dict):
    schema = mlflow.artifacts.load_text(
        "feature_schema.json"
    )
    # ... 校验逻辑

3. 内存泄露

长时间运行后服务内存持续增长。排查发现是因为 MLflow 的日志对象没有正确释放，加上大模型加载后没有清理旧引用。

解决：

  • 定期重启（用 K8s 的 rolling update）
  • 避免在请求处理里创建大对象
  • 如果有必要，用弱引用或手动清理

4. 模型版本不一致

某次部署后线上预测结果突然变差。查下来是部署脚本从错误的 stage 拉模型。

解决：

  • 在部署流水线里增加校验步骤，确认拉取的模型版本
  • 部署后自动跑一次回归测试
  • 重要模型部署需要人工确认
# deploy_validation.py
def validate_deployment(model_name: str, expected_version: str):
    actual_version = get_current_deployment_version()

    if actual_version != expected_version:
        raise RuntimeError(
            f"版本不一致！期望 {expected_version}，实际 {actual_version}"
        )

    # 跑一套回归测试
    test_results = run_regression_tests()

    if not test_results["passed"]:
        raise RuntimeError(
            f"回归测试失败：{test_results['details']}"
        )

经验总结

折腾了一年，这条链路算是基本稳住了。回过头看，有几件事是对的：

  1. 先解决最痛的问题，再逐步扩展。我们没有一开始就上 Kubeflow Pipeline，而是先把实验追踪和模型管理做扎实，再上自动化部署。

  2. 工具要轻量。MLflow + GitHub Actions + K8s 这个组合，每个组件都能独立理解和替换，出问题时排查成本相对低。

  3. 可观测性很重要。模型上线后如果看不到真实表现，就无法判断是否需要重新训练。我们把预测准确度、响应时间、请求量都接入了监控系统。

  4. 自动化是为了减少错误，不是为了省人力。流水线还是要有人看、有人维护，自动化只负责把重复劳动和人为失误降下来。

  5. 不要为了 MLOps 而 MLOps。如果只有一两个模型且更新不频繁，手动部署可能反而更高效。等痛苦积累到一定程度，再上自动化工具。

留给未来的坑

现在这套方案还有几个地方没完全解决：

  1. A/B 测试：多版本模型同时上线，按流量比例分配，这个还没做。目前是 Staging 验证过就切 Production，有一定风险。

  2. 数据标注成本：监控到性能退化后，重新训练需要新数据。标注数据的成本目前比较高，这块自动化不了。

  3. 多云部署：我们的服务现在只跑在一个云上，灾备和多区域部署还没做。

  4. 模型解释：业务方有时候问"为什么这个预测是这样"，目前只能给个大致解释，没有更细粒度的特征归因。

这些问题有些是技术问题，有些是组织问题。技术部分可以继续搞，组织部分需要时间沉淀。

路还长

MLOps 不是一套工具或一个流水线，而是一套把模型从实验变成稳定服务的实践。工具会换，流水线会改，但核心问题不变：如何让模型在生产环境持续提供可靠的价值。

这件事没有终点，只有不断的优化和调整。

希望这篇文章对正在路上的人有点用。

主要参考：

  • MLflow 官方文档：https://mlflow.org/docs/latest/
  • MLOps 实践指南：https://www.mlops.org/
  • GitHub Actions 文档：https://docs.github.com/en/actions

版权声明: 本文首发于 指尖魔法屋-MLOps流水线折腾手记（https://blog.thinkmoon.cn/post/152-mlops-pipeline-practice-experiment-to-production/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章

gitlab中vue前端项目CI/CD部署笔记
本文详细记录了在GitLab中实现Vue前端项目的CI/CD持续集成与部署过程。涵盖了GitLab Runner的安装配置、.gitlab-ci.yml文件的编写（包括本地部署与远程推送两种方案），以及常见的大小写敏感问题排查，为前端自动化部署提供实践参考。
📁 技术实践
📅 2020-06-23
使用github Action自动部署博客
本文详细介绍了如何利用GitHub Actions实现博客的自动化部署。文章从背景痛点分析入手，提供了完整的Action配置流程、服务器端Webhook监听及自动拉取重启脚本，是一套实用的CI/CD解决方案，有效解决了本地编译繁琐和服务器资源不足的问题。
📁 教程分享
📅 2022-06-14