ml-mlops

ML MLOps

概述

将此技能用于机器学习运维：实验跟踪、可复现性、编排、注册表、CI/CD、模型部署治理、监控、漂移响应和重新训练。MLOps 可将笔记本和脚本转变为可审计、可重复且权责明确的系统，并支持回滚。

MLOps 不变量

每个生产机器学习工作流都应回答以下问题：

• 哪些数据、代码、配置、环境和硬件生成了此模型？
• 哪些指标、数据切片和测试为晋级提供了依据？
• 模型制品存储在哪里？如何回滚？
• 如何验证传入数据和服务特征？
• 如何监控质量、漂移、延迟、成本、安全性和公平性？
• 谁负责处理事件、重新训练、审批和弃用？

如果有任何问题没有答案，请先完善生命周期，再添加更多基础设施。

实验跟踪

跟踪参数、指标、制品、数据集版本、代码修订版本、环境、随机种子、运行备注和标签。对于开源模型注册表工作流，MLflow 是可靠的默认选择；Weights & Biases 擅长提供丰富的实验仪表板和参数搜索；Neptune、Comet 和 ClearML 也能满足类似的协作和制品管理需求。应根据治理要求、托管方式、集成能力和团队工作流进行选择，而不应只看仪表板的外观。

为实验和运行采用一致的命名方式。记录最终指标和学习曲线。保存混淆矩阵、校准图、特征重要性、验证集预测结果和有代表性的错误。除非跟踪后端获准处理相应类别的数据，否则请勿记录密钥、原始个人身份信息（PII）或专有示例。

模型注册表和版本控制

注册表条目应包括制品 URI、模型签名或架构、输入/输出示例、预处理器/分词器引用、依赖环境、训练数据集标识符、评估报告、审批状态、负责人、变更日志和预期用途。

标准化模型注册表元数据架构（JSON）

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "ModelRegistryMetadata",
  "type": "object",
  "properties": {
    "model_name": { "type": "string" },
    "version": { "type": "string" },
    "artifact_uri": { "type": "string", "format": "uri" },
    "framework": {
      "type": "object",
      "properties": {
        "name": { "type": "string" },
        "version": { "type": "string" }
      },
      "required": ["name", "version"]
    },
    "signature": {
      "type": "object",
      "properties": {
        "inputs": {
          "type": "array",
          "items": {
            "type": "object",
            "properties": {
              "name": { "type": "string" },
              "type": { "type": "string" },
              "shape": { "type": "array", "items": { "type": "integer" } }
            },
            "required": ["name", "type"]
          }
        },
        "outputs": {
          "type": "array",
          "items": {
            "type": "object",
            "properties": {
              "name": { "type": "string" },
              "type": { "type": "string" },
              "shape": { "type": "array", "items": { "type": "integer" } }
            },
            "required": ["name", "type"]
          }
        }
      },
      "required": ["inputs", "outputs"]
    },
    "provenance": {
      "type": "object",
      "properties": {
        "git_commit": { "type": "string" },
        "dataset_snapshot_uri": { "type": "string" },
        "environment_lockfile_uri": { "type": "string" }
      },
      "required": ["git_commit", "dataset_snapshot_uri"]
    }
  },
  "required": ["model_name", "version", "artifact_uri", "framework", "signature", "provenance"]
}

阶段转换应由自动化质量门禁控制；在风险需要时，还应经过人工审批。仅当含义得到落实时，才使用 candidate、staging、production 和 archived 等语义模型标签。使用 DVC、湖仓表版本、对象存储清单或特征存储物化时间戳等工具为数据集添加版本。仅有模型版本并不足够，数据和特征代码也是模型的一部分。

流水线编排

根据运行环境选择编排工具：

平台	最适用场景
Kubeflow Pipelines	原生支持 Kubernetes 的机器学习工作流、可移植组件、复杂 DAG
SageMaker Pipelines	原生支持 AWS 的训练、处理、模型注册表、端点、Feature Store、Clarify、Model Monitor
Vertex AI Pipelines	原生支持 GCP 的训练、端点、Feature Store、Model Monitoring、AutoML 集成
Azure ML Pipelines	Azure 工作区、托管计算、注册表、托管端点、Responsible ML
Databricks Workflows	以湖仓为中心的特征工程、MLflow 注册表、Spark、Delta、笔记本/作业
Airflow/Prefect/Dagster	通用数据/机器学习编排，拥有广泛的生态系统集成

保持流水线模块化：数据验证、特征生成、训练、评估、注册、部署和监控设置应拆分为独立步骤，并明确输入和输出。在安全的情况下，使步骤具备幂等性并支持缓存。对于 CI 中的 Azure ML 代码资产注册、ADF 到 Azure ML 的版本传递、指针 Blob 或 ADF WebActivity 路径的运行时验证，请加载 ml-azureml-adf-automation。

机器学习 CI/CD

CI 应运行快速检查：代码 lint、预处理单元测试、架构测试、确定性小数据集训练冒烟测试、模型加载测试、序列化测试、推理契约测试和安全扫描。CD 只能部署通过评估门禁的制品。

生产级 GitHub Actions MLOps 工作流（.github/workflows/mlops.yml）

name: MLOps CI/CD Workflow

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
        cache: 'pip'
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install ruff pytest pytest-cov -r requirements.txt
    - name: Lint with Ruff
      run: ruff check src/
    - name: Run unit tests
      run: pytest tests/unit/ --cov=src --cov-report=xml
      
  model-smoke-test:
    needs: lint-and-test
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    - name: Run small-scale training smoke test
      run: |
        pip install -r requirements.txt
        python src/train.py --config configs/smoke_test.yaml
    - name: Verify model serialization & load contract
      run: python src/verify_contract.py --model_path outputs/model.pt

  promote-and-deploy:
    needs: model-smoke-test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Deploy model to Staging endpoint
      env:
        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
      run: |
        pip install awscli
        # Trigger staging endpoint update
        python scripts/deploy.py --stage staging

对于模型服务，请使用蓝绿、金丝雀、影子或 A/B 部署。先将少量流量导入金丝雀版本，监控防护指标，然后逐步扩大流量。影子模式适合进行无风险对比，但除非实际采用其决策，否则无法衡量对用户产生影响的指标。始终确保回滚简单易行且经过测试。

数据验证

在训练和服务之前验证数据。Great Expectations、TFDV 和 Deequ 可以强制执行架构和分布预期。

Great Expectations 验证套件设置

import great_expectations as ge
from great_expectations.core.expectation_suite import ExpectationSuite

def create_ingestion_validation_suite(suite_name: str = "ingestion_suite") -> ExpectationSuite:
    context = ge.get_context()
    suite = context.create_expectation_suite(suite_name, overwrite_existing=True)
    
    # Assert column names exist
    suite.add_expectation({
        "expectation_type": "expect_table_columns_to_match_ordered_list",
        "kwargs": {
            "column_list": ["user_id", "timestamp", "feature_a", "feature_b", "label"]
        }
    })
    
    # Assert feature_a ranges
    suite.add_expectation({
        "expectation_type": "expect_column_values_to_be_between",
        "kwargs": {
            "column": "feature_a",
            "min_value": 0.0,
            "max_value": 100.0
        }
    })
    
    # Assert label is never null
    suite.add_expectation({
        "expectation_type": "expect_column_values_to_not_be_null",
        "kwargs": {
            "column": "label"
        }
    })
    
    context.save_expectation_suite(suite)
    return suite

确保时间穿越联接能够防止未来信息泄漏，并监控在线特征的新鲜度。

监控与漂移

在以下四个层面进行监控：

1. 系统：可用性、延迟、吞吐量、错误、GPU/CPU/内存利用率、队列长度、成本。
2. 数据：架构、空值、范围、类别基数、嵌入范数、文本长度、图像尺寸、特征分布漂移。
3. 预测：分数分布、置信度、校准、类别分布、阈值结果、拒绝/安全标签。
4. 结果：延迟标签、业务指标、公平性切片、人工审核率、事件。

生产监控告警阈值参考

指标类别	目标指标	建议告警阈值	诊断操作
系统	p99 延迟	连续 5 分钟 > 200ms	检查服务容器线程池、GPU/CPU 瓶颈、网络存储
系统	错误率	任意 1 分钟窗口内 > 1.0%	检查服务后端日志，排查原始输入验证导致的 OOM 或架构违规
数据	特征漂移（PSI）	总体稳定性指数 > 0.25	触发调查：检查源数据架构变化或上游筛选器变更
预测	分数漂移（KS）	Kolmogorov-Smirnov 检验 p 值 < 0.01	针对潜在标签/概念漂移发出告警；将预测分布与基线进行比较
结果	校准漂移	Brier 分数较基线下降 > 15%	检查概念漂移；标记以收集日志并加入人工标注队列

漂移是调查信号，并不能自动证明重新训练会改善结果。

治理、安全与合规

记录预期用途、范围外用途、训练数据限制、模型局限性、已知风险和评估切片。对数据、制品、注册表和端点实施最小权限访问控制。扫描容器和依赖项。在托管式密钥存储中保护密钥。对传输中和静态存储的数据及制品进行加密。对于敏感系统，保留数据访问、审批、部署和事件的审计日志。

成本控制

按项目和负责人标记资源，使用自动扩缩容，为端点选择合适规格，关闭闲置笔记本和集群；仅在使用检查点时优先选择竞价/可抢占式训练；将批量推理安排在非高峰时段；在延迟允许时评估无服务器或 CPU 推理。跟踪每次训练运行的成本、每 1,000 次预测的成本，以及每次被接受的模型改进所需的成本。

来源

• MLflow 文档：https://mlflow.org/docs/latest/index.html
• Weights & Biases 文档：https://docs.wandb.ai/
• Kubeflow Pipelines 文档：https://www.kubeflow.org/docs/components/pipelines/
• AWS SageMaker MLOps 文档：https://docs.aws.amazon.com/sagemaker/
• Vertex AI MLOps 文档：https://cloud.google.com/vertex-ai/docs
• Azure Machine Learning 文档：https://learn.microsoft.com/azure/machine-learning/