AI运维工程师（MLOps）入门指南：从零搭建机器学习部署流水线

1次阅读

共计 2136 个字符，预计需要花费 6 分钟才能阅读完成。

关注点不同 ：DevOps 聚焦代码交付和基础设施管理，MLOps 还需处理数据管道、模型版本和监控（Data Pipeline/Model Versioning/Monitoring）
迭代周期差异 ：传统 DevOps 以周 / 天为单位，MLOps 因涉及模型重训练可能按小时级更新（Retraining Cycle）
环境复杂性 ：ML 系统需要管理数据、模型、代码三重依赖（Dependency Management）

Apache Airflow：适合调度型任务（如定时数据预处理），优势在于丰富的 Operator 和可视化 DAG（Directed Acyclic Graph）
Kubeflow：Kubernetes 原生 ML 平台，提供从训练到服务的全生命周期管理（ML Pipeline/TF Serving）
MLflow：轻量级实验跟踪工具（Experiment Tracking），适合快速迭代的算法团队

# 阶段一：构建环境
FROM python:3.8-slim as builder
COPY requirements.txt .
RUN pip install --user -r requirements.txt

# 阶段二：运行时镜像（优化体积）FROM python:3.8-slim
WORKDIR /app
# 只拷贝必要文件
COPY --from=builder /root/.local /root/.local
COPY model.pkl .
COPY app.py .

# 模型服务入口
ENV PATH=/root/.local/bin:$PATH
EXPOSE 5000
CMD ["gunicorn", "--bind=0.0.0.0:5000", "app:app"]

关键优化：多阶段构建减少镜像体积（从 1.2GB→180MB）

name: Model CI
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.8'
      - name: Run unit tests
        run: |
          pip install -r requirements.txt
          python -m pytest tests/ --cov=src
      - name: Model validation
        run: python validate.py  # 验证 AUC>0.9

触发条件：代码推送时自动运行测试和模型验证

from prometheus_client import Counter, Gauge

# 定义指标
REQUEST_COUNT = Counter('inference_requests', 'Total API calls')
LATENCY = Gauge('inference_latency', 'ms per request')

@app.route('/predict', methods=['POST'])
def predict():
    start_time = time.time()
    REQUEST_COUNT.inc()
    # ... 模型推理代码...
    LATENCY.set((time.time()-start_time)*1000)

监控重点：请求量、延迟、错误率（Error Rate）

预热加载 ：服务启动时预加载模型到内存（Preload Model）
缓存机制 ：对高频特征组合预计算（Feature Precomputation）
精简依赖 ：移除训练专用库（如 TensorFlow Serving 比完整 TF 小 70%）

水平扩展 ：HPA（Horizontal Pod Autoscaler）基于 CPU/ 内存自动扩缩
批处理优化 ：通过请求队列（Request Queue）合并推理任务
分级降级 ：流量激增时返回轻量级模型（如降级到决策树）

# Kubeflow 版本回退命令
kubectl rollout undo deployment/model-service --to-revision=3

必须同步回滚对应的特征处理器（Feature Processor）

# 计算特征分布差异
from scipy import stats

def detect_drift(current, baseline):
    p_value = stats.ks_2samp(current, baseline).pvalue
    return p_value < 0.01  # 显著性检验

监控指标：PSI（Population Stability Index）>0.25 时报警