MLOps流水线实战指南：从实验追踪到持续学习

MLOps MLflow 持续学习 CI/CD
作者: 醉月思 发布日期: 2026-02-18 分类: 技术实践

前言：MLOps 不是工具，是一套实践

很多团队的模型部署状态：三个模型跑在三个人本地，版本管理靠文件名后缀（v1_final.pkl、v1_final_real.pkl、v1_final_real_really_final.pkl），发布靠手动 rsync，出问题靠群里问谁改了代码。

MLOps 解决的核心问题：

  • 训练环境和推理环境不一致
  • 模型版本管理混乱
  • 实验无法追踪和复现
  • 部署依赖手动操作
  • 数据漂移导致模型退化

一、MLOps 的完整链路

graph TB subgraph 数据层 A[原始数据] --> B[数据预处理] B --> C[特征工程] end subgraph 训练层 C --> D[模型训练] D --> E[模型评估] end subgraph 部署层 E --> F[模型打包] F --> G[部署上线] G --> H[监控告警] end subgraph 管理层 I[实验追踪 MLflow] J[工作流编排] K[版本管理] end

二、环境一致性问题

2.1 一个真实的事故

训练用 scikit-learn 1.3.2，线上部署 1.1.2，某个参数默认值变了，预测结果全部偏了 15%。从发现问题到回滚用了 4 小时，凌晨三点才稳定下来。

2.2 解决方案：Docker 统一环境

FROM pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime

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

CMD ["python", "main.py"]

关键：训练和服务都用同一个 Docker 基础镜像。

三、实验追踪：MLflow

3.1 部署 MLflow

# 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

  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"
    command: server /data

3.2 集成到训练代码

import mlflow
import mlflow.sklearn

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

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

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

    # 评估并记录指标
    y_pred = model.predict(X_test)
    mlflow.log_metrics({
        "mse": mean_squared_error(y_test, y_pred),
        "r2": r2_score(y_test, y_pred)
    })

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

3.3 PyTorch 大模型的坑

mlflow.log_model 对 2GB 以上模型可能超时。解决：分开存权重和架构

mlflow.pytorch.log_model(model, "model")
torch.save(model.state_dict(), "model_weights.pth")
mlflow.log_artifact("model_weights.pth")

四、模型版本管理

4.1 注册模型

model_uri = f"runs:/{run.info.run_id}/model"
mlflow.register_model(model_uri, "house_price_predictor")

4.2 模型生命周期

from mlflow import MlflowClient

client = MlflowClient()

# 设为 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"
)

三个阶段： None → Staging → Production → Archived

4.3 权限控制

MLflow Model Registry 默认是全局权限控制，没有细粒度 ACL。删除 Production 模型需要两个人确认。

五、CI/CD 流水线

5.1 GitHub Actions 配置

name: MLOps Pipeline

on:
  push:
    branches: [main]

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 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 improved
        run: python scripts/register_if_improved.py

  deploy-staging:
    needs: train-and-evaluate
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Deploy to Staging
        env:
          KUBECONFIG: ${{ secrets.KUBECONFIG_STAGING }}
        run: |
          kubectl set image deployment/model-serving \
            serving=your-registry/predictor:staging \
            -n staging

5.2 模型服务化

from fastapi import FastAPI
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):
    X = np.array([[features["area"], features["bedrooms"]]])
    prediction = model.predict(X)
    return {"prediction": float(prediction[0])}

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

5.3 K8s 部署

apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-serving
spec:
  replicas: 2
  template:
    spec:
      containers:
      - name: serving
        image: your-registry/predictor:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30  # 大模型需要更长时间
          periodSeconds: 10

坑： initialDelaySeconds 必须根据模型大小调整，大模型可能需要 45 秒加载。

六、工作流编排

6.1 方案对比

工具优势适用场景
Shell 脚本简单直接原型、小项目
Airflow功能完整、生态成熟复杂工作流
Prefect现代化、Python 友好ML 工作流
Kubeflow PipelinesK8s 原生大规模生产

6.2 Airflow 实现

from airflow import DAG
from airflow.operators.python import PythonOperator

dag = DAG('ai_training_pipeline', schedule_interval='@daily')

preprocess_task = PythonOperator(
    task_id='preprocess_data',
    python_callable=preprocess_data,
    dag=dag,
)

train_task = PythonOperator(
    task_id='train_model',
    python_callable=train_model,
    dag=dag,
)

# 定义依赖
preprocess_task >> train_task

6.3 Prefect + MLflow

from prefect import flow, task
import mlflow

@task
def preprocess_data(input_path):
    with mlflow.start_run():
        processed = do_preprocess(input_path)
        mlflow.log_metric("samples_count", len(processed))
    return processed

@task
def train_model(data, config):
    with mlflow.start_run():
        mlflow.log_params(config)
        model = do_train(data, config)
        mlflow.pytorch.log_model(model, "model")
    return model

@flow(name="ai-training-pipeline")
def ai_training_pipeline(config):
    data = preprocess_data(config["input_path"])
    model = train_model(data, config["model_config"])
    return model

七、持续训练

7.1 性能监控

def check_model_performance(model_name, threshold=0.05):
    """检查模型是否出现性能退化"""
    client = MlflowClient()

    prod_version = client.get_latest_versions(
        model_name, stages=["Production"]
    )[0]

    # 获取最近 7 天的线上预测误差
    recent_errors = []
    for log in get_recent_prediction_logs():
        error = abs(log["actual"] - log["predicted"]) / log["actual"]
        recent_errors.append(error)

    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%}，建议重新训练")
        trigger_retraining()

7.2 数据漂移检测

from scipy import stats

def detect_drift(new_data, baseline_stats):
    drift_detected = False
    for feature in baseline_stats.keys():
        _, p_value = stats.ks_2samp(
            new_data[feature],
            np.random.normal(
                baseline_stats[feature]["mean"],
                baseline_stats[feature]["std"],
                len(new_data)
            )
        )
        if p_value < 0.05:
            print(f"⚠️  特征 {feature} 可能漂移 (p={p_value:.4f})")
            drift_detected = True
    return drift_detected

八、持续学习（Continual Learning）

8.1 灾难遗忘问题

电商商品分类：原 10 个类别准确率 95%，加入 3 个新类别重训后，旧类别准确率跌到 40%。

这是 catastrophic forgetting，神经网络的老问题。

8.2 方案一：全量重训

把所有历史数据攒起来重新训练。

问题：

  • 数据量线性增长，训练成本高
  • 数据权限和隐私管理复杂
  • 新噪声会影响已训练好的部分

8.3 方案二：重播缓冲区

训练新任务时混合一部分旧数据。

import random
from collections import deque

class ReplayBuffer:
    def __init__(self, buffer_size=1000):
        self.buffer = deque(maxlen=buffer_size)

    def add(self, data):
        self.buffer.append(data)

    def sample(self, batch_size):
        return random.sample(self.buffer, batch_size)

def incremental_training_with_replay(model, new_data, replay_buffer):
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    for new_batch in new_data:
        optimizer.zero_grad()

        # 新任务损失
        loss_new = model.loss_fn(new_batch)

        # 重播旧数据
        old_batch = replay_buffer.sample(len(new_batch))
        loss_old = model.loss_fn(old_batch)

        (loss_new + 0.3 * loss_old).backward()
        optimizer.step()

效果： 旧类别准确率从 95% 降到 75%（比忘干净好，但还是不够）。

8.4 方案三：EWC（Elastic Weight Consolidation）

对旧任务重要的参数少动，不重要的随便改。

class EWC:
    def __init__(self, model, old_data, lambda_ewc=5000):
        self.model = model
        self.lambda_ewc = lambda_ewc
        self.old_params = {name: param.clone().detach()
                          for name, param in model.named_parameters()}
        self.fisher = self._compute_fisher(model, old_data)

    def _compute_fisher(self, model, data):
        fisher = {name: torch.zeros_like(param)
                 for name, param in model.named_parameters()}

        model.eval()
        for batch in data:
            outputs = model(batch)
            loss = F.cross_entropy(outputs, batch.labels)
            model.zero_grad()
            loss.backward()

            for name, param in model.named_parameters():
                if param.grad is not None:
                    fisher[name] += param.grad.pow(2)

        return {name: f / len(data) for name, f in fisher.items()}

    def penalty(self, model):
        loss = 0
        for name, param in model.named_parameters():
            if name in self.fisher:
                loss += (self.fisher[name] *
                        (param - self.old_params[name]).pow(2)).sum()
        return loss * self.lambda_ewc

坑： lambda_ewc 很难调。太小管不住遗忘，太大又学不到新东西。

8.5 方案四：混合方案（推荐）

EWC + 重播缓冲区的组合：

class HybridIncrementalLearning:
    def __init__(self, model, buffer_size=500, lambda_ewc=3000):
        self.model = model
        self.replay_buffer = ReplayBuffer(buffer_size)
        self.ewc = None
        self.lambda_ewc = lambda_ewc

    def subsequent_task(self, new_data):
        optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)

        for new_batch in new_data:
            optimizer.zero_grad()

            loss_new = self._compute_loss(new_batch)
            loss_ewc = self.ewc.penalty(self.model)
            old_batch = self.replay_buffer.sample(len(new_batch))
            loss_replay = self._compute_loss(old_batch)

            loss = loss_new + 0.5 * loss_ewc + 0.3 * loss_replay
            loss.backward()
            optimizer.step()

        # 更新缓冲区和 EWC
        for batch in new_data:
            self.replay_buffer.add(batch)
        self.ewc = EWC(self.model, new_data, self.lambda_ewc)

实际效果：

  • 旧类别准确率维持 85%
  • 新类别学习速度不会太慢
  • 重播缓冲区只保留 5% 旧数据

8.6 持续学习的坑

坑一：类别不平衡

新类别数据通常比旧类别多，模型倾向新类别。解决：加权损失、难例采样。

坑二：长尾遗忘

某些本来就难的类别用任何方案都容易忘。解决：定期监控，掉太多就单独补训。

坑三：超参数管理

每个阶段的参数都要调，不同任务不通用。经验难迁移到下一个项目。

坑四：评估成本

要知道遗忘程度必须测所有历史任务。任务多了评估比训练还长。

九、踩坑总结

坑一：模型加载时机

把模型加载放在请求处理函数里，每次预测重新加载，第一次请求等几十秒。

解决： 应用启动时加载，用全局变量持有。

坑二：环境隔离

训练用 pandas 新版本，模型反序列化时出错。

解决：

  • 依赖版本固定在 requirements.txt
  • 训练和服务用同一 Docker 基础镜像
  • 模型注册时记录特征定义

坑三：内存泄露

长时间运行后内存持续增长。MLflow 日志对象没释放 + 大模型没清理旧引用。

解决：

  • 定期重启（K8s rolling update）
  • 避免请求处理里创建大对象

坑四：模型版本不一致

部署脚本从错误的 stage 拉模型。

解决：

  • 部署流水线增加版本校验
  • 部署后自动跑回归测试
  • 重要部署需要人工确认

坑五：数据依赖

训练跑了一晚上，第二天发现数据源中途更新了。

解决：数据快照

def save_data_snapshot(data_path, snapshot_dir):
    data_hash = get_data_hash(data_path)
    snapshot_path = os.path.join(snapshot_dir, f"{data_hash}.pkl")
    if not os.path.exists(snapshot_path):
        shutil.copy(data_path, snapshot_path)
    return snapshot_path

坑六：资源竞争

同时跑多个实验，GPU 显存不够。

解决： Prefect 资源限制 + K8s 资源配额。

十、工具选型

10.1 按项目规模选

项目规模推荐方案
个人/小项目Shell 脚本 + Jupyter
小团队MLflow + GitHub Actions
中型团队MLflow + Prefect + K8s
大型企业Kubeflow Pipelines + 全套

10.2 不要为了 MLOps 而 MLOps

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

十一、实际效果

某团队上 MLOps 后的效果：

指标改造前改造后
完整流水线时间手动 2 小时自动 15 分钟
实验迭代速度基线提升 3 倍
环境不一致问题频繁减少 80%
新人上手时间一周两天
MTTR（故障恢复）数小时分钟级

十二、写在最后

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

几条核心原则：

  1. 先解决最痛的问题，再逐步扩展
  2. 工具要轻量，每个组件能独立理解和替换
  3. 可观测性很重要，看不到真实表现就无法判断是否需要重训
  4. 自动化是为了减少错误，不是为了省人力
  5. 持续学习是妥协的艺术，设定可接受阈值，不追求完美
  6. 一开始就要想"以后怎么办"，不要先堆功能再重构

技术部分可以继续搞，组织部分需要时间沉淀。MLOps 这件事没有终点，只有不断的优化和调整。

本文整合了 5 篇 MLOps 流水线相关文章，涵盖实验追踪、模型版本管理、CI/CD、工作流编排、持续训练、持续学习、灾难遗忘等核心技术。

版权声明: 本文首发于 指尖魔法屋-MLOps流水线实战指南：从实验追踪到持续学习（https://blog.thinkmoon.cn/post/ai-mlops-pipeline-comprehensive-guide/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

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

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