AI 工程化与 DevOps（十）：从零搭建 MLOps 平台，10 期精华全整合

开篇：10 期学完了，现在把它们串起来

如果你是从第一期一路跟下来的读者，恭喜你 : :  你已经在 AI 工程化的路上走完了从入门到实战的完整闭环。

回顾这 10 期的旅程，我们循序渐进地覆盖了 MLOps 的全景地图：

第一期：MLOps 成熟度模型与团队评估，定义了「我们在哪里，要去哪里」

第二期：实验追踪，用 MLflow 管理每一次训练的元数据

第三期：数据版本控制，用 DVC + Feast 让数据集可复现

第四期：训练流水线自动化，Kubeflow 与 Airflow 的实战用法

第五期：模型服务，BentoML、Triton Inference Server 和 vLLM 的部署方案

第六期：模型监控，用 Evidently 检测数据漂移

第七期：CI/CD for ML，GitHub Actions 驱动的自动化发布

第八期：LLMOps，大语言模型特有的 Prompt 管理、追踪与评估

第九期：可观测性，日志 + 指标 + 链路追踪三位一体

第十期：端到端整合，用一套代码把所有组件串联起来。

每一期都是一个独立的组件，但真正的 MLOps 平台是这些组件的有机结合。本期我们将用最简洁的方式，搭建一个涵盖模型注册 → 训练 → 评估 → 自动 Stage 过渡 → 金丝雀部署 → 监控 → 告警的完整 MLOps 平台。

这不仅仅是代码的堆叠，而是一套可复用的架构模板 : :  你可以直接拿来改改，用在真实项目里。

MLOps 平台架构总览：四层架构

一个生产级的 MLOps 平台，可以抽象为四个层次。每一层解决一类问题，层与层之间通过标准接口通信。

数据层

数据层是 MLOps 的地基，解决的是「数据从哪来、如何管理、如何复用」。

核心组件包括：

DVC（Data Version Control）：像 Git 管理代码一样管理数据集。对大文件（GB/TB 级）做指针式版本管理，数据集本身存储在 S3/MinIO 等远程存储中。

Feast：特征存储（Feature Store），让离线训练和在线推理使用同一套特征定义和特征值。消除「训练-推理特征不一致」这个经典问题。

MinIO：兼容 S3 协议的对象存储，常作为自建数据湖的核心。所有训练数据、模型 Artifacts 和日志都可以存放在这里。

数据层的关键原则：可复现。无论过多久，拉回正确的代码和正确的数据版本，就能复现出相同的结果。

训练层

训练层是模型产出的流水线工厂。

核心组件包括：

MLflow：实验追踪的标配工具。记录每次训练的 hyperparameters、metrics、artifacts，并提供 Model Registry 做模型版本管理和 Stage 过渡。

Kubeflow：Kubernetes 原生的 ML 工作流平台。用 Pipeline DSL 定义多步骤训练流程，天然支持分布式训练、GPU 调度。

Airflow：通用的工作流调度引擎。如果你已经有 Airflow 基础设施，用它编排 ML 流水线完全没有问题。ETL 到模型训练到模型评估，一个 DAG 搞定。

训练层的关键原则：可编排。单次训练可以手动跑，但重复性的训练、定时重训、触发式训练必须交给流水线。

服务层

服务层解决的是「模型训练好后，如何被业务系统调用」。

核心组件包括：

BentoML：Python 最友好的模型服务框架之一。自动生成 OpenAPI 文档、集成 Docker 打包、支持模型批处理。几行代码就能把 sklearn 或 PyTorch 模型变成一个 REST API。

Triton Inference Server：NVIDIA 出品的高性能推理服务器。支持多模型并发、动态批处理、GPU 模型并发。适合高吞吐场景。

vLLM：大语言模型推理引擎。PagedAttention 技术让 LLM 推理效率提升数倍。结合 Triton 或 BentoML 可以搭建 LLM 服务。

服务层的关键原则：可扩展。水平扩展、自动伸缩、版本路由，一个都不能少。

治理层

治理层是 MLOps 平台的运维大脑。

核心组件包括：

Evidently：模型监控库，支持数据漂移检测、模型性能衰减分析。可以对接 MLflow 或 Grafana 做可视化。

Grafana + Prometheus：指标收集与展示的标准方案。Prometheus 采集模型服务的 latency、throughput、error rate，Grafana 提供 Dashboard。

Langfuse：LLM 的可观测性工具。追踪每次 Prompt 的调用链、Token 消耗、响应质量。

ELK / Loki：日志聚合。训练日志、服务日志、告警日志统一管理和检索。

治理层的关键原则：可观测。看不到的运行就是黑盒，黑盒里的模型出了问题你永远不会第一时间知道。

四层架构的整合图

    数据层                 训练层                服务层                治理层
  ┌─────────┐          ┌──────────┐          ┌──────────┐          ┌──────────┐
  │ DVC     │          │ MLflow   │          │ BentoML  │          │ Evidently│
  │ Feast   │    →     │ Kubeflow │    →     │ Triton   │    →     │ Grafana  │
  │ MinIO   │          │ Airflow  │          │ vLLM     │          │ Langfuse │
  └────┬────┘          └────┬─────┘          └────┬─────┘          └────┬─────┘
       │                    │                    │                    │
       └────────────────────┴────────────────────┴────────────────────┘
                              │
                         CI/CD Pipeline
                   (GitHub Actions + Docker + K8s)

每一期对应的就是这个四层架构中的一个或多个组件。第一期讲的是战略，第二到三期讲数据层和实验层，第四期讲训练流水线，第五期讲服务层，第六、九期讲监控和可观测性（治理层），第七期讲底层的 CI/CD 自动化，第八期讲 LLM 专属治理。

本期代码，则用最小可行的方式把这四层抽象成了一个完整的 Python 平台类。

完整流水线演示：模型注册 → 训练 → 评估 → 自动 Stage 过渡 → 金丝雀部署 → 监控 → 告警

接下来我们看一段端到端的流水线演示。这段代码来自本期仓库的 scripts/mlops_platform.py，它模拟了一次完整的 MLOps 工作流。

模型注册表（Model Registry）

首先，我们需要一个地方来管理所有模型版本。MLflow 自带的 Model Registry 是生产级方案，但为了演示清晰的逻辑，我们先写一个轻量级的实现：

class ModelStage(Enum):
    NONE = "None"
    STAGING = "Staging"
    PRODUCTION = "Production"
    ARCHIVED = "Archived"


@dataclass
class ModelVersion:
    """模型版本"""
    name: str
    version: int
    stage: ModelStage = ModelStage.NONE
    metrics: Dict[str, float] = field(default_factory=dict)
    created_at: str = ""
    run_id: str = ""

每个模型版本包含四个关键信息：模型名称、版本号、当前所处阶段、性能指标。Stage 的流转是 MLOps 的核心机制：

None：新训练的模型，未经评估，不可上线

Staging：通过评估阈值的模型，进入预发布队列

Production：通过了金丝雀验证，正式上线服务

Archived：退役模型，仅保留历史记录用于审计

训练 + 自动 Stage 过渡

有了注册表，我们定义「训练并自动评估」的流程：

def train_and_register(self, model_name: str, metrics: Dict[str, float]):
    """训练 + 自动注册模型"""

    # 1. 模拟训练
    print(f"  [1/3] 训练中...")

    # 2. 获取当前版本号，递增
    existing = self.registry.models.get(model_name, [])
    next_version = (max(v.version for v in existing) if existing else 0) + 1

    # 3. 注册新版本
    print(f"  [2/3] 注册模型 v{next_version}")
    mv = ModelVersion(
        name=model_name,
        version=next_version,
        stage=ModelStage.NONE,
        metrics=metrics,
        created_at=str(datetime.now()),
    )

    # 4. 自动 Stage 过渡 : :  核心逻辑
    threshold_met = metrics.get("f1", 0) >= 0.85
    if threshold_met:
        print(f"  [3/3] 自动 Stage: None → Staging (F1={metrics['f1']:.4f} >= 0.85)")
        self.registry.promote(model_name, next_version, ModelStage.STAGING)
    else:
        print(f"  [3/3] 保留在 None (F1={metrics['f1']:.4f} < 0.85)")

这里的关键设计是自动 Stage 过渡。传统的做法是人肉去 MLflow UI 手动拖拽模型晋升，但生产环境中这个动作必须自动化。我们设置了一个可配置的 F1 阈值（0.85），只要自动评估的指标达标，模型就自动进入 Staging 阶段。

这是 CI/CD for ML 的精髓 : :  不再依赖人工判断，而是用预定义的评估标准自动化流水线。

金丝雀部署（Canary Deployment）

模型进入 Staging 后并不会立刻全量上线。我们会用金丝雀部署策略逐步放量：

def deploy_canary(self, model_name: str, version: int, traffic_pct: int = 10):
    """金丝雀部署：先放 10% 流量"""
    deploy_id = uuid.uuid4().hex[:8]
    self.deployments[deploy_id] = {
        "model": model_name,
        "version": version,
        "traffic_pct": traffic_pct,
        "status": "canary",
        "started_at": str(datetime.now()),
    }
    print(f"  🟡 金丝雀部署: {model_name} v{version} → {traffic_pct}% 流量")
    return deploy_id

金丝雀部署的逻辑是：

1. 新版本先接收 10% 的线上流量，旧版本继续处理 90%
2. 在观察期内，监控系统持续检查各项指标（延迟、错误率、数据漂移）
3. 如果指标正常，逐步放量到 100%（全量发布）
4. 如果指标异常，自动回滚到旧版本

def promote_to_full(self, deploy_id: str):
    """全量发布"""
    self.deployments[deploy_id]["traffic_pct"] = 100
    self.deployments[deploy_id]["status"] = "production"
    dep = self.deployments[deploy_id]
    print(f"  🟢 全量发布: {dep['model']} v{dep['version']} → 100% 流量")


def rollback(self, deploy_id: str):
    """自动回滚"""
    self.deployments[deploy_id]["traffic_pct"] = 0
    self.deployments[deploy_id]["status"] = "rolled_back"
    print(f"  🔴 回滚: {self.deployments[deploy_id]['model']} 已回滚")

在真实生产环境中，金丝雀策略的流量控制通过 Kubernetes Service Mesh（如 Istio）或 API Gateway 实现。例如在 Istio 中，用一个 VirtualService 就能做到：

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: churn-model
spec:
  hosts:
  - churn-model
  http:
  - route:
    - destination:
        host: churn-model-v2
        subset: v2
      weight: 10
    - destination:
        host: churn-model-v1
        subset: v1
      weight: 90

监控 + 告警

模型全量上线不代表结束，真正的挑战刚刚开始 : :  线上环境的输入分布会随时间变化，模型性能会衰减。

class MonitorDashboard:
    """监控面板模拟"""

    def check_drift(self, model_name: str, drift_score: float, threshold: float = 0.3):
        """数据漂移检测"""
        if drift_score > threshold:
            severity = "critical" if drift_score > 0.5 else "warning"
            print(f"  [DRIFT] {model_name}: drift_score={drift_score:.3f} > {threshold}")

    def check_latency(self, model_name: str, p99_ms: float, threshold_ms: float = 500):
        """P99 延迟监控"""
        if p99_ms > threshold_ms:
            print(f"  [LATENCY] {model_name}: P99={p99_ms:.0f}ms > {threshold_ms}ms")

这里我们监控两个关键指标：数据漂移和延迟。在生产环境中，这些指标会由 Prometheus 采集，Grafana 展示，AlertManager 发送告警到钉钉、飞书或 PagerDuty。

完整流水线一次执行

把上述所有步骤串起来，就是一次完整的 MLOps 流水线执行：

def full_pipeline_demo(self):
    """完整流水线演示"""

    # 1. 查看当前注册表状态
    self.registry.dashboard()

    # 2. 训练新模型（自动评估 + Stage 过渡）
    v = self.train_and_register("customer_churn",
                                 {"accuracy": 0.962, "f1": 0.94,
                                  "precision": 0.95, "recall": 0.93})

    # 3. 金丝雀部署 → 全量发布
    self.deploy_model("customer_churn", v, traffic=10)

    # 4. 训练另一个新模型
    self.train_and_register("recommendation_engine",
                            {"accuracy": 0.78, "f1": 0.72,
                             "precision": 0.74, "recall": 0.70})

    # 5. 全局监控检查（自动触达告警）
    self.monitor_all()

    # 6. 查看最终状态
    self.registry.dashboard()

运行这个演示的输出大致如下：

============================================================
  Model Registry Dashboard
============================================================
  模型名称                       版本   阶段               F1
-------------------------------------------------------
  customer_churn              v1    Production     0.8800
  customer_churn              v2    Staging        0.9100
  customer_churn              v3    None           0.9250
  sentiment_analysis          v1    Production     0.8500
  fraud_detection             v1    Production     0.7800
  fraud_detection             v2    Staging        0.8200

Training: customer_churn v4 ...
  ✅ 自动 Stage: None → Staging (F1=0.94 >= 0.85)
  🟡 金丝雀部署: customer_churn v4 → 10% 流量
  🟢 全量发布: customer_churn v4 → 100% 流量
  ✅ customer_churn v4: Staging → Production

[DRIFT] customer_churn: drift_score=0.230 正常
[DRIFT] sentiment_analysis: drift_score=0.150 正常
[LATENCY] fraud_detection P99=520ms > 500ms
[DRIFT] fraud_detection: drift_score=0.550 > 0.50 CRITICAL

[Active Alerts: 2]
  [WARNING] high_latency: fraud_detection
  [CRITICAL] data_drift: fraud_detection

这套流水线体现的核心原则是：从训练到上线到监控，每个步骤都可以自动化，且通过可配置的阈值驱动决策，而非人工干预。

平台各组件整合方式

前面说的是平台「做什么」，现在我们说「怎么整合」。以下是各组件之间的通信方式和整合要点。

MLflow ↔ DVC：实验与数据的版本绑定

DVC 管理的数据集版本会通过 dvc.lock 文件记录在 Git 仓库里。在 MLflow 实验的 run_id 下，可以用 tag 记录对应的 DVC 数据版本哈希：

# 训练前，拉取指定版本的数据
dvc checkout data/train.csv.dvc

# 训练时，记录数据版本到 MLflow
mlflow.log_param("data_version", $(dvc status --md 2>/dev/null | head -1))

这样，任意一个 MLflow run 都可以追溯到它使用的数据版本，做到完整的可复现性。

MLflow ↔ BentoML：从 Model Registry 直接部署

BentoML 原生支持从 MLflow Model Registry 拉取模型：

import bentoml
import mlflow

# 从 MLflow 拉取指定生产模型的某个版本
model_uri = "models:/customer_churn/Production"
bento_model = bentoml.mlflow.import_model("churn_bento", model_uri)

# 构建服务
svc = bentoml.Service("churn_service", runners=[bento_model.to_runner()])

# 定义 API
@svc.api(input=bentoml.io.JSON(), output=bentoml.io.JSON())
def predict(input_data):
    result = bento_model.predict(input_data)
    return {"prediction": result.tolist()}

这样一来，MLflow 的 Stage 晋升会自动影响 BentoML 的模型拉取版本 : :  当你 promote 一个新版本到 Production 时，BentoML 服务会自动重新加载新模型。

Evidently ↔ MLflow/Grafana：监控指标的回流

Evidently 检测出的漂移指标可以写入 MLflow 或 Prometheus：

import evidently
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

# 计算漂移
report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=ref, current_data=cur)
drift_score = report.as_dict()["metrics"][0]["result"]["drift_score"]

# 回写到 MLflow
with mlflow.start_run(run_id=target_run_id):
    mlflow.log_metric("data_drift_score", drift_score)

# 同时暴露为 Prometheus 指标
DRIFT_GAUGE = prometheus_client.Gauge("model_data_drift", "Data Drift Score")
DRIFT_GAUGE.set(drift_score)

监控数据回流到 MLflow 方便历史对比，暴露为 Prometheus 指标则用于实时告警。两者互补，而不是替代关系。

Langfuse ↔ LLM 服务：LLM 的完整调用链

对 LLM 服务，Langfuse 提供了 OpenAI SDK 的无侵入 Trace 能力：

from langfuse import Langfuse
from openai import OpenAI

langfuse = Langfuse()
client = OpenAI()

# Langfuse 自动捕获所有 OpenAI 请求
@langfuse.observe(as_type="generation")
def llm_predict(prompt: str):
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
    )
    # Langfuse 自动记录 Token 数、延迟、模型名
    return response.choices[0].message.content

在生产环境中，Langfuse 的 Trace 数据会与前端的 Nginx 日志、后端的业务日志关联，形成端到端的可观测性链路。

生产级扩展建议

前面几期我们已经深入探讨了每一个组件的详细用法。本期我们把这些扩展建议归纳为几个关键方向。

K8s 部署（Helm Charts）

用 Helm Chart 管理模型服务的部署，可以做到声明式管理和一键回滚：

# values.yaml
model:
  name: customer-churn
  version: 4
  image: registry.example.com/churn-model:v4

deployment:
  replicas: 3
  resources:
    requests:
      cpu: "1"
      memory: "2Gi"
    limits:
      cpu: "2"
      memory: "4Gi"

canary:
  enabled: true
  trafficWeight: 10

K8s 的优势是天然的滚动更新、水平伸缩和资源管理。一个模型服务就是一个 Deployment，配上 Service 和 Ingress 就能对外暴露。

GPU 节点池

对于深度学习模型，GPU 是必需品。在 K8s 中配置 GPU 节点池可以实现成本优化：

# 标记 GPU 节点
kubectl label nodes gpu-node-1 accelerator=nvidia-gpu

# 在 Deployment 中指定 GPU 需求
# resources:
#   limits:
#     nvidia.com/gpu: 1

对于大模型推理（7B+ 参数），通常需要 A100 或 H100 级别的 GPU。对中等规模模型（如 BERT），T4 或 L4 就足够了。建议将 GPU 节点用 taint 隔离，只有真正需要 GPU 的 Pod 才能调度上去。

多环境隔离（Dev / Staging / Prod）

用 K8s Namespace 做环境隔离是最标准的方式：

apiVersion: v1
kind: Namespace
metadata:
  name: ml-dev
---
apiVersion: v1
kind: Namespace
metadata:
  name: ml-staging
---
apiVersion: v1
kind: Namespace
metadata:
  name: ml-prod

不同环境的配置可以通过 Helm 的 values 文件区分：

• dev：最小资源上限，关闭自动缩放，无监控告警
• staging：与 prod 一致但无真实流量，用于金丝雀前的最终验证
• prod：完整资源分配，启用 HPA，绑定监控和告警

Web UI（Streamlit / Gradio）

给非技术人员提供一个模型管理界面，用 Streamlit 几十分钟就能搭建：

import streamlit as st
from mlops_platform import MLOpsPlatform

st.set_page_config(page_title="MLOps Dashboard", layout="wide")
platform = MLOpsPlatform()

st.title("🔬 MLOps 平台管理中心")

# 模型注册表概览
col1, col2, col3 = st.columns(3)
col1.metric("模型总数", len(platform.registry.models))
col2.metric("生产版本", sum(1 for v in platform.registry.models.values()
                              for m in v if m.stage.value == "Production"))
col3.metric("待部署", sum(1 for v in platform.registry.models.values()
                            for m in v if m.stage.value == "Staging"))

# 部署操作
model_name = st.selectbox("选择模型", list(platform.registry.models.keys()))
if st.button("部署最新版本"):
    versions = platform.registry.models[model_name]
    latest = max(versions, key=lambda v: v.version)
    platform.deploy_model(model_name, latest.version)

这个 Streamlit Dashboard 对接了我们的 MLOpsPlatform 类，可以实现：

• 查看模型注册表状态
• 手动触发训练
• 部署和回滚操作
• 查看监控告警

10 期知识图谱回顾

在最后这一期，我们用一张表格把整条学习路径浓缩展示。你可以把它当作快速检索手册，在遇到具体问题时快速定位到对应的期数和解决方案。

这张表的价值在于：它不是知识点的罗列，而是一条有逻辑的递进链。每一期的「输出」都是下一期的「输入」。

行动指南：下一步做什么

学完了 10 期内容，你现在可以做什么？以下是按难度排序的行动路线。

立即可以做的事（今天就能上手）

用 MLflow 管理下个模型训练：无论你是做分类、回归还是 NLP 任务，在训练脚本里加三行代码就能开启实验追踪：

import mlflow

mlflow.set_experiment("my_first_mlops")
with mlflow.start_run():
    mlflow.log_params({"lr": 0.01, "epochs": 50})
    # ... 训练代码 ...
    mlflow.log_metrics({"accuracy": 0.92, "f1": 0.88})

用 DVC 版本化你的数据集：在项目根目录执行 dvc init，然后用 dvc add data/ 接管数据目录。从此所有历史数据版本都有迹可循。

一周内可以完成的事

搭建一个模型服务：用 BentoML 包装你最近的模型，生成 Docker 镜像并部署到 K8s 或 Docker Compose 上。

pip install bentoml
bentoml serve service.py:svc --reload

配置模型监控：用 Evidently 对比训练数据的分布和线上数据的分布，设置一个简单的漂移阈值告警。

一个月内可以做的事

实现完整的 CI/CD Pipeline：参照第七期的内容，用 GitHub Actions 搭建从代码提交 → 模型训练 → 自动评估 → 部署到环境的完整链路。

搭建 Grafana Dashboard：把模型服务的延迟、吞吐量、错误率、数据漂移分数汇聚到一个 Dashboard 上。

三个月内可以挑战的事

实现自动化重训：当监控检测到数据漂移超过阈值时，自动触发重新训练流程。

搭建多模型 A/B 测试平台：同时运行多个版本，按实验设计分配流量，自动收集效果数据。

致谢 + 全系列总结

10 期，10 个主题，从 MLOps 的战略思考到每一行代码的落地实现。

这一路走来，我们覆盖了以下这些核心认知：

MLOps 不是工具，而是工程文化。MLflow、BentoML、Kubeflow 只是工具，真正的 MLOps 是你愿意为模型的可复现性、可编排性、可扩展性和可观测性投入工程时间。工具帮你完成 80% 的工作，但剩下 20% 的架构决策和流程设计，需要你基于对项目实际需求的理解来做。

从最小可行做起。不要一开始就想搭建完美的四层架构。从 MLflow 实验追踪开始 : :  这只需要半天时间。然后加 DVC 数据版本 : :  又是一天。逐步迭代，而不是一步到位。MLOps 平台的建设应该像模型的训练一样：从小规模开始，观察效果，不断迭代。

自动化是杠杆，监控是保障。自动化的价值不在于替代人工，而在于让团队从重复劳动中解放出来去做更有价值的事。但自动化如果没有监控配合，就是一台高速运转的失控机器。自动化和监控是一对孪生兄弟，缺一不可。

LLMOps 不是全新的领域，而是 MLOps 的子集加扩展。大语言模型带来的 Prompt 管理、Token 追踪和幻觉检测是新的挑战，但模型版本管理、CI/CD、可观测性和监控的基础框架完全复用。如果你已经掌握了 ML 模型的 MLOps，切换到 LLMOps 的学习成本会非常低。

做难而正确的事。在项目初期就引入 MLOps 实践，看起来是「增加了工作量」。但从几个月的时间尺度来看，MLOps 节省的时间远远超过它的投入。不做 MLOps 的 AI 项目，第一周效率很高，第二个月开始陷入混乱，第三个月开始返工重做。做了 MLOps 的项目，前期慢一点，之后越来越快。

如果你读到了这里，我由衷地感谢你的陪伴。10 期文章，每期都在 5000 字以上，这不是一个小工程。希望这些内容真的帮到了你，无论你是一位 AI 工程师、一名技术负责人，还是正在踏入 AI 工程领域的学生。

最后，我想引用一个观点作为全系列的收尾：

AI 的价值不在于模型，而在于系统。 一个 80 分精度的模型 + 完善的工程体系，远比一个 95 分精度但无法运维的模型有价值。工程化的能力，是区分「AI 实验室」和「AI 产品」的分水岭。

全系列代码仓库和项目规划在下方，欢迎 Star、Fork、提 Issue。我们下个专题再见。

📦 全部代码：https://gitee.com/genesisesNoun/ai-mlops-tutorial

🔗 专题规划：https://gitee.com/genesisesNoun/ai-mlops-tutorial

📚 专题十八完整目录：追更请关注「CV技术笔记」