原文：https://dev.to/naman_2004/building-a-production-grade-end-to-end-mlops-pipeline-from-scratch-l9h（作者 @naman_2004）

从零搭建生产级端到端 MLOps 流水线

本文是一篇手把手教程，带你用 DVC、MLflow、FastAPI、Docker、GitHub Actions、Prometheus 和 Evidently AI 搭建一套完整的 MLOps 流水线，覆盖从数据版本管理到漂移检测的整个机器学习生命周期。

大多数机器学习教程讲到 model.fit() 就结束了。但在真实世界里，训练一个模型只占全部工作量的 10% 左右，剩下 90% 都是让机器学习在生产环境里真正运转起来所必需的工作——数据版本管理、数据校验、实验追踪、API 服务、自动化测试、容器化、CI/CD、监控以及漂移检测。

本文将带你以经典的泰坦尼克号生存预测问题为例，搭建一套完整的、生产级 MLOps 流水线。重点不在构建一个多复杂的模型，而在于落地每一项生产实践——正是这些实践，把 Jupyter Notebook 里的一次实验和一套可部署的机器学习系统区分开来。

GitHub 仓库： End-to-end-MLOPS-Pipeline

架构总览

下面就是我们要搭建的完整系统：

Data (Titanic CSV)
 │
 ▼
DVC (Data Versioning)
 │
 ▼
Data Validation (Schema & Constraint Checks)
 │
 ▼
Scikit-learn Training Pipeline
 │
 ▼
MLflow (Experiment Tracking & Model Registry)
 │
 ▼
Pytest (Automated Testing)
 │
 ▼
Docker (Containerization)
 │
 ▼
GitHub Actions (CI/CD)
 │
 ▼
FastAPI (Inference Service)
 │
 ├── Prometheus (Monitoring)
 │
 └── Evidently AI (Drift Detection)

技术栈

版本控制：Git、GitHub
数据版本管理：DVC
模型训练：Scikit-learn
实验追踪：MLflow
API 框架：FastAPI
测试：Pytest
容器化：Docker
CI/CD：GitHub Actions
监控：Prometheus
漂移检测：Evidently AI

项目结构

End-to-end-MLOPS-Pipeline/
│
├── data/
│   └── raw/
│       └── titanic.csv
│
├── models/
│   └── model.joblib
│
├── src/
│   ├── data_ingestion.py
│   ├── data_validation.py
│   ├── data_preprocess.py
│   ├── train.py
│   └── drift_detection.py
│
├── api/
│   ├── __init__.py
│   └── main.py
│
├── tests/
│   ├── __init__.py
│   ├── test_api.py
│   └── test_model.py
│
├── monitoring/
│   ├── prometheus.yml
│   └── drift_report.html
│
├── .github/
│   └── workflows/
│       └── ci_cd.yml
│
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── .gitignore
├── .dockerignore
└── .dvc/

第 1 步：项目初始化与环境配置

首先初始化 Git 并创建虚拟环境：

git init
python -m venv venv

# 激活虚拟环境
# Windows PowerShell:
.\venv\Scripts\Activate.ps1
# Linux/macOS:
source venv/bin/activate

`.gitignore`

创建一个 .gitignore 文件，保持仓库整洁：

# 虚拟环境
venv/
env/
.env

# Python 缓存
__pycache__/
*.pyc

# 数据文件（由 DVC 管理，不进 Git）
data/raw/*
!data/raw/*.dvc
!data/raw/*.csv
data/processed/

# 模型与 MLflow
models/
mlruns/

`requirements.txt`

# 数据与机器学习
pandas>=2.0.0
numpy>=1.24.0
scikit-learn>=1.3.0
joblib>=1.3.0

# 数据校验与实验追踪
dvc>=3.0.0
mlflow>=2.10.0

# API
fastapi>=0.100.0
uvicorn[standard]>=0.22.0
pydantic>=2.0.0

# 测试
pytest>=7.4.0
httpx>=0.24.0
prometheus-client>=0.17.0
evidently

安装全部依赖：

pip install -r requirements.txt

第 2 步：用 DVC 做数据版本管理

为什么不直接把数据集提交到 Git？

Git 是为源代码——也就是小型文本文件——设计的。而数据集动辄几百 MB 甚至几个 GB，直接提交到 Git 会让仓库变得臃肿、克隆缓慢，规模一大根本没法管理。

DVC（Data Version Control） 通过以下方式解决这个问题：

在 Git 中只保存轻量的 .dvc 元数据文件（只有几个字节）；
把真正的数据文件存放在远程存储（S3、GCS 或本地缓存）；
可以像 git checkout 一样，把数据集回滚到任意历史版本。

配置步骤

下载 Titanic 数据集，放到 data/raw/titanic.csv。

# 初始化 DVC
dvc init

# 用 DVC 追踪数据集
dvc add data/raw/titanic.csv

# 提交元数据
git add .
git commit -m "Initialize project and track dataset with DVC"

DVC 会生成一个 data/raw/titanic.csv.dvc 文件——一个极小的元数据指针，Git 追踪的是它，而不是真正的 CSV 文件。

步骤 3：数据摄取（`src/data_ingestion.py`）

数据摄取模块负责从磁盘加载原始数据，并带有防御性检查——确保文件存在且不为空之后，才把数据传给下游。

import os
import logging
import pandas as pd

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def load_data(file_path: str) -> pd.DataFrame:
    """
    Load raw CSV data and validate it.
    """
    if not os.path.exists(file_path):
        logging.error(f"File not found at path: {file_path}")
        raise FileNotFoundError(f"File does not exist: {file_path}")

    logging.info(f"Loading data from {file_path}")
    df = pd.read_csv(file_path)

    if df.empty:
        logging.error("Loaded dataframe is empty")
        raise ValueError("Dataframe is empty")

    logging.info(f"Successfully loaded dataset with shape: {df.shape}")
    return df


if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    print(df.head())

为什么用 `logging` 而不是 `print()`？

在生产环境的 MLOps 里，print() 打印出来的东西就像石沉大海。logging 能给你带来：

每个操作都带时间戳。
严重级别（INFO、WARNING、ERROR），方便过滤噪音。
可以把日志写入文件，或者发送到监控系统。

步骤 4：数据校验（`src/data_validation.py`）

在碰任何机器学习模型之前，我们先校验传入的数据是否符合预期。这是抵御上游数据变更导致模型静默失效的第一道防线。

import os
import sys
import logging
import pandas as pd

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# Define expected schema and column rules
REQUIRED_COLUMNS = [
    "Survived", "Pclass", "Name", "Sex", "Age", 
    "SibSp", "Parch", "Ticket", "Fare", "Embarked"
]
VALID_PCLASS = {1, 2, 3}
VALID_SEX = {"male", "female"}


def validate_data(df: pd.DataFrame) -> bool:
    """
    Validates dataset integrity, schema, and column value constraints.
    """
    logging.info("Starting data validation checks...")

    # 1. Check for required columns
    missing_cols = [col for col in REQUIRED_COLUMNS if col not in df.columns]
    if missing_cols:
        logging.error(f"Validation failed: Missing columns: {missing_cols}")
        raise ValueError(f"Missing required columns: {missing_cols}")

    # 2. Check categorical domains
    invalid_pclass = set(df["Pclass"].dropna().unique()) - VALID_PCLASS
    if invalid_pclass:
        logging.error(f"Validation failed: Unexpected Pclass values: {invalid_pclass}")
        raise ValueError(f"Invalid values in Pclass: {invalid_pclass}")

    invalid_sex = set(df["Sex"].dropna().unique()) - VALID_SEX
    if invalid_sex:
        logging.error(f"Validation failed: Unexpected Sex values: {invalid_sex}")
        raise ValueError(f"Invalid values in Sex: {invalid_sex}")

    # 3. Check target column for missing values
    if df["Survived"].isnull().any():
        logging.error("Validation failed: Target column 'Survived' contains null values.")
        raise ValueError("Target column 'Survived' cannot contain null values.")

    # 4. Range checks
    if (df["Fare"] < 0).any():
        logging.error("Validation failed: Found negative values in 'Fare'.")
        raise ValueError("Fare values cannot be negative.")

    logging.info("All data validation checks passed successfully!")
    return True


if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    validate_data(df)

我们到底在防御什么？

必填列必须存在：如果上游有人把 Sex 改名成了 Gender，流水线会在这里就报错，而不是等到模型训练深处才抛出一段莫名其妙的错误信息。
`Pclass` ∈ {1, 2, 3}：未知类别会让 one-hot 编码直接崩溃，或者产出垃圾特征。
`Survived` 不能有空值：监督学习的目标变量必须完整。
`Fare >= 0`：负数票价在逻辑上不可能存在，一旦出现就说明数据已经损坏。

运行方式：

python src/data_validation.py
# Output: All data validation checks passed successfully!

第 5 步：数据预处理与特征工程（`src/data_preprocess.py`）

import os
import sys
import logging
import pandas as pd
from sklearn.model_selection import train_test_split

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data
from src.data_validation import validate_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def preprocess_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    用简单的 pandas 操作清洗泰坦尼克号原始数据：
    1. 填充缺失值（Age 用中位数，Embarked 用众数）。
    2. 删除对建模无用的列（PassengerId、Name、Ticket、Cabin）。
    3. 通过独热编码（get_dummies）把 Sex 和 Embarked 转成数字。
    """
    df = df.copy()

    # 1. 填充缺失值
    df["Age"] = df["Age"].fillna(df["Age"].median())
    df["Fare"] = df["Fare"].fillna(df["Fare"].median())
    df["Embarked"] = df["Embarked"].fillna(df["Embarked"].mode()[0])

    # 2. 删除建模用不到的列
    drop_cols = ["PassengerId", "Name", "Ticket", "Cabin"]
    df = df.drop(columns=[col for col in drop_cols if col in df.columns])

    # 3. 把文本/分类列转换成数字（独热编码）
    df = pd.get_dummies(df, columns=["Sex", "Embarked"], drop_first=True, dtype=int)

    return df


def split_data(df: pd.DataFrame, test_size: float = 0.2, random_state: int = 42):
    """
    把清洗后的数据框切分成训练集和测试集。
    """
    X = df.drop(columns=["Survived"])
    y = df["Survived"]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, random_state=random_state, stratify=y
    )

    return X_train, X_test, y_train, y_test


if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    validate_data(df)

    cleaned_df = preprocess_data(df)
    logging.info(f"Cleaned data shape: {cleaned_df.shape}")
    print("\nCleaned Data Preview:")
    print(cleaned_df.head())

    X_train, X_test, y_train, y_test = split_data(cleaned_df)
    logging.info(f"Train samples: {len(X_train)}, Test samples: {len(X_test)}")

关键决策解析

用中位数/众数做 `fillna`：中位数对异常值更稳健（均值则不然），众数则用出现次数最多的值来填补分类字段的空缺。
删掉的列：PassengerId 是唯一标识符（没有预测价值），Name 和 Ticket 是高基数的文本字段，Cabin 的缺失率超过 70%。
`pd.get_dummies(drop_first=True)`：把 Sex（male/female）转成 Sex_male（1/0），把 Embarked（C/Q/S）转成 Embarked_Q、Embarked_S。设置 drop_first 是为了避免虚拟变量陷阱。
`stratify=y`：确保训练集和测试集都保持同样的 38/62% 生还/遇难比例。

运行方式：

python src/data_preprocess.py
# 输出：
# Cleaned data shape: (891, 9)
# Train samples: 712, Test samples: 179

第 6 步：用 MLflow 进行模型训练与实验跟踪（`src/train.py`）

大多数教程到这里就止步了。但我们不只是训练一个模型——我们还要记录每一次实验，这样才能对比各次运行、复现结果，并把表现最好的模型推向生产环境。

import os
import sys
import logging
import joblib
import mlflow
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score
)

sys.path.append(
    os.path.abspath(
        os.path.join(os.path.dirname(__file__), "..")
    )
)
from src.data_ingestion import load_data
from src.data_preprocess import preprocess_data, split_data

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)


def train_model(
    n_estimators: int = 100,
    max_depth: int = 5,
    random_state: int = 42
):
    """
    Train a Random Forest model and log results to MLflow.
    """
    # Load raw data
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)

    # Preprocess data
    cleaned_df = preprocess_data(df)

    # Train-test split
    X_train, X_test, y_train, y_test = split_data(
        cleaned_df,
        random_state=random_state
    )

    # Configure MLflow
    mlflow.set_tracking_uri("sqlite:///mlflow.db")
    mlflow.set_experiment("Titanic-Survival-Prediction")

    with mlflow.start_run():
        logging.info("Training Random Forest Classifier...")

        # Train model
        model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            random_state=random_state
        )
        model.fit(X_train, y_train)

        # Predictions
        y_pred = model.predict(X_test)

        # Metrics
        acc = accuracy_score(y_test, y_pred)
        prec = precision_score(y_test, y_pred)
        rec = recall_score(y_test, y_pred)
        f1 = f1_score(y_test, y_pred)

        logging.info(
            f"Accuracy={acc:.4f}, "
            f"Precision={prec:.4f}, "
            f"Recall={rec:.4f}, "
            f"F1={f1:.4f}"
        )

        # Log parameters
        mlflow.log_param("n_estimators", n_estimators)
        mlflow.log_param("max_depth", max_depth)
        mlflow.log_param("random_state", random_state)

        # Log metrics
        mlflow.log_metric("accuracy", acc)
        mlflow.log_metric("precision", prec)
        mlflow.log_metric("recall", rec)
        mlflow.log_metric("f1_score", f1)

        # Save model locally
        os.makedirs("models", exist_ok=True)
        model_path = os.path.join("models", "model.joblib")
        joblib.dump(model, model_path)
        logging.info(f"Model saved successfully at: {model_path}")

        # Log model artifact to MLflow
        mlflow.log_artifact(model_path, artifact_path="model")
        logging.info("Experiment successfully logged to MLflow.")

    return model, acc


if __name__ == "__main__":
    train_model(n_estimators=100, max_depth=5)

MLflow 给我们带来了什么

每次 train_model() 运行时，MLflow 都会生成一个唯一的 Run ID，并保存以下内容：

参数（n_estimators、max_depth）：方便你对比每条结果是由哪种配置跑出来的。
指标（accuracy、f1_score）：跨实验追踪模型性能表现。
工件（model.joblib）：真正训练出来的模型二进制文件，带版本记录、可直接下载。

跑一下：

python src/train.py
# Output: Accuracy=0.7821, Precision=0.7885, Recall=0.5942, F1=0.6777

# Launch the MLflow dashboard:
python -m mlflow ui
# Open http://localhost:5000 in your browser

现在，你就可以在 MLflow UI 中并排直观地对比多次训练运行的超参数和各项指标了。

第 7 步：FastAPI 推理服务（`api/main.py`）

终端用户不会去运行 Python 脚本，他们发送的是 HTTP 请求。我们的 FastAPI 服务会：

使用 Pydantic 校验传入的 JSON（格式不合法的数据会被 422 错误直接拒绝）。
把输入转换成模型所期望的确切特征格式。
返回预测结果，附带生存概率。
记录 Prometheus 指标，用于监控。

import os
import time
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

# 1. 初始化 FastAPI 应用
app = FastAPI(
    title="Titanic Survival Prediction API",
    description="Production-ready inference service with Prometheus monitoring.",
    version="1.0.0"
)

# 2. Prometheus 指标定义
REQUEST_COUNT = Counter(
    "api_request_count_total", 
    "Total HTTP requests received", 
    ["method", "endpoint", "status"]
)
REQUEST_LATENCY = Histogram(
    "api_request_latency_seconds", 
    "Histogram of request latencies in seconds", 
    ["endpoint"]
)
PREDICTION_COUNT = Counter(
    "model_prediction_count_total", 
    "Total predictions generated by the model", 
    ["prediction"]
)

# 3. 训练好的模型路径
MODEL_PATH = os.path.join("models", "model.joblib")
if not os.path.exists(MODEL_PATH):
    raise FileNotFoundError(f"Trained model not found at {MODEL_PATH}!")

model = joblib.load(MODEL_PATH)


# 4. Pydantic Schema
class PassengerInput(BaseModel):
    Pclass: int = Field(..., ge=1, le=3, description="Ticket class (1 = 1st, 2 = 2nd, 3 = 3rd)")
    Sex: str = Field(..., description="Gender: 'male' or 'female'")
    Age: float = Field(..., ge=0, le=120, description="Age in years")
    SibSp: int = Field(..., ge=0, description="Number of siblings/spouses aboard")
    Parch: int = Field(..., ge=0, description="Number of parents/children aboard")
    Fare: float = Field(..., ge=0.0, description="Passenger fare")
    Embarked: str = Field(..., description="Port of Embarkation: 'C', 'Q', or 'S'")


class PredictionResponse(BaseModel):
    survived: bool
    survival_probability: float


# 5. Prometheus 抓取端点
@app.get("/metrics")
def get_metrics():
    """Exposes Prometheus application metrics."""
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.get("/")
def health_check():
    return {"status": "healthy", "service": "titanic-survival-prediction"}


@app.post("/predict", response_model=PredictionResponse)
def predict(passenger: PassengerInput):
    start_time = time.time()
    try:
        input_data = pd.DataFrame([{
            "Pclass": passenger.Pclass,
            "Age": passenger.Age,
            "SibSp": passenger.SibSp,
            "Parch": passenger.Parch,
            "Fare": passenger.Fare,
            "Sex_male": 1 if passenger.Sex.lower() == "male" else 0,
            "Embarked_Q": 1 if passenger.Embarked.upper() == "Q" else 0,
            "Embarked_S": 1 if passenger.Embarked.upper() == "S" else 0,
        }])

        prediction = model.predict(input_data)[0]
        probability = model.predict_proba(input_data)[0][1]

        # 记录指标
        PREDICTION_COUNT.labels(prediction=str(int(prediction))).inc()
        REQUEST_COUNT.labels(method="POST", endpoint="/predict", status="200").inc()
        REQUEST_LATENCY.labels(endpoint="/predict").observe(time.time() - start_time)

        return PredictionResponse(
            survived=bool(prediction == 1),
            survival_probability=round(float(probability), 4)
        )

    except Exception as e:
        REQUEST_COUNT.labels(method="POST", endpoint="/predict", status="500").inc()
        raise HTTPException(status_code=500, detail=str(e))

特征对齐陷阱

注意这里是手动构造 Sex_male、Embarked_Q 和 Embarked_S 这几个特征。它们必须与训练阶段通过 pd.get_dummies(..., drop_first=True) 生成的哑变量列完全一致。如果你忘了加 Embarked_Q，或者本该加 Sex_male 却加了 Sex_female，模型会悄无声息地输出垃圾预测——不会报任何错误，只会给出错误的数字。

API 示例

请求：

{
  "Pclass": 1,
  "Sex": "female",
  "Age": 29.0,
  "SibSp": 0,
  "Parch": 0,
  "Fare": 100.0,
  "Embarked": "S"
}

响应：

{
  "survived": true,
  "survival_probability": 0.9200
}

运行方式：

uvicorn api.main:app --reload --port 8000
# 打开 http://localhost:8000/docs 查看交互式 Swagger UI

第 8 步：使用 Pytest 进行自动化测试

测试是 CI/CD 的护栏：任何一条测试失败，部署就会自动中止。

`tests/test_model.py` —— 模型完整性测试

import os
import joblib
import pandas as pd


def test_model_file_exists():
    """验证训练好的模型文件是否存在于磁盘上。"""
    model_path = os.path.join("models", "model.joblib")
    assert os.path.exists(model_path), "Model artifact 'model.joblib' is missing!"


def test_model_prediction_output():
    """验证模型能接收一条样本输入并输出 0 或 1。"""
    model_path = os.path.join("models", "model.joblib")
    model = joblib.load(model_path)

    sample = pd.DataFrame([{
        "Pclass": 3,
        "Age": 22.0,
        "SibSp": 1,
        "Parch": 0,
        "Fare": 7.25,
        "Sex_male": 1,
        "Embarked_Q": 0,
        "Embarked_S": 1
    }])

    pred = model.predict(sample)
    assert pred[0] in [0, 1], f"Unexpected prediction value: {pred[0]}"

`tests/test_api.py` —— API 契约测试

from fastapi.testclient import TestClient
from api.main import app

client = TestClient(app)


def test_health_check():
    """测试 GET / 是否返回健康状态。"""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"status": "healthy", "service": "titanic-survival-prediction"}


def test_predict_endpoint_valid_input():
    """用合法的乘客 JSON 测试 POST /predict。"""
    payload = {
        "Pclass": 1,
        "Sex": "female",
        "Age": 29.0,
        "SibSp": 0,
        "Parch": 0,
        "Fare": 100.0,
        "Embarked": "S"
    }
    response = client.post("/predict", json=payload)
    assert response.status_code == 200
    
    data = response.json()
    assert "survived" in data
    assert "survival_probability" in data
    assert isinstance(data["survived"], bool)
    assert 0.0 <= data["survival_probability"] <= 1.0


def test_predict_endpoint_invalid_input():
    """测试传入非法数据时 POST /predict 是否正确报错。"""
    invalid_payload = {
        "Pclass": 1,
        "Sex": "female",
        "Age": -5.0,  # 负数年龄应被 Pydantic 的 ge=0 校验拒绝
        "SibSp": 0,
        "Parch": 0,
        "Fare": 50.0,
        "Embarked": "S"
    }
    response = client.post("/predict", json=invalid_payload)
    assert response.status_code == 422  # Unprocessable Entity（不可处理的实体）

运行：

pytest tests/ -v
# tests/test_model.py::test_model_file_exists PASSED
# tests/test_model.py::test_model_prediction_output PASSED
# tests/test_api.py::test_health_check PASSED
# tests/test_api.py::test_predict_endpoint_valid_input PASSED
# tests/test_api.py::test_predict_endpoint_invalid_input PASSED

第 9 步：Docker 容器化

为什么要用 Docker？

你的模型在你自己的 Windows、Python 3.13 环境下跑得好好的，但换到 Ubuntu 云服务器上就可能挂掉——原因可能是缺少 C++ 构建工具、包版本冲突，或者文件路径不一致。Docker 会把你的 Python 版本、依赖、模型文件和 FastAPI 服务一起打包进一个隔离、可复现的 Linux 容器。

`Dockerfile`

# 1. 使用官方的轻量级 Python 运行时
FROM python:3.11-slim

# 2. 设置环境变量
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

# 3. 设置容器内的工作目录
WORKDIR /app

# 4. 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 5. 复制必要的应用代码和模型文件
COPY api/ ./api/
COPY src/ ./src/
COPY models/ ./models/

# 6. 暴露 FastAPI 监听的端口
EXPOSE 8000

# 7. 使用 Uvicorn 运行应用的启动命令
CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

层缓存技巧

注意到 COPY requirements.txt 和 RUN pip install 写在 COPY api/ 和 COPY src/ 之前了吗？Docker 会对构建层做缓存：如果你只改了源码（没动依赖），Docker 会直接跳过整个 pip install 步骤，大约 2 秒就能完成重新构建，而不是 2 分钟。

`.dockerignore`

venv/
.env
__pycache__/
*.pyc
.git/
.dvc/
data/raw/
mlruns/
mlflow.db
tests/
notebooks/

`docker-compose.yml`

services:
  titanic-api:
    image: titanic-mlops
    container_name: titanic-mlops-api
    ports:
      - "8000:8000"
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    container_name: titanic-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped

构建与运行

# 构建镜像
docker build -t titanic-mlops .

# 用 Docker Compose 启动（API + Prometheus）
docker compose up -d

现在可以访问：

API：http://localhost:8000/docs
Prometheus：http://localhost:9090

第 10 步：使用 GitHub Actions 实现 CI/CD

每次向 main 分支推送代码，都会触发一条自动化流水线，完成数据校验、模型训练、测试执行和 Docker 镜像构建——全部运行在 GitHub 的免费云端 runner 上。

`.github/workflows/ci_cd.yml`

name: Titanic MLOps Pipeline

on:
  push:
    branches: [ "main", "master" ]
  pull_request:
    branches: [ "main", "master" ]

jobs:
  mlops_pipeline:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout Code
      uses: actions/checkout@v4

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

    - name: Install Dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt

    - name: Run Data Validation
      run: |
        python src/data_validation.py

    - name: Train Model
      run: |
        python src/train.py

    - name: Run Pytest Suite
      env:
        PYTHONPATH: .
      run: |
        pytest tests/ -v

    - name: Build Docker Image
      run: |
        docker build -t titanic-mlops-api:latest .

关于 `PYTHONPATH: .` 的修复

在 Linux CI runner 上，项目根目录并不会被自动加入 Python 的模块搜索路径。如果不设置 PYTHONPATH: .，pytest 就找不到 api 模块，并报出 ModuleNotFoundError: No module named 'api' 错误。

推送代码，然后观察流水线运行：

git add .
git commit -m "Add CI/CD pipeline"
git push origin main

打开 GitHub 仓库的 Actions 标签页，就能实时看到流水线的执行过程。

第 11 步：用 Prometheus 做生产环境监控

Prometheus 追踪哪些指标？

在 api/main.py 中，我们定义了三种指标类型：

api_request_count_total：Counter 类型，按端点和状态码统计的 HTTP 请求总数
api_request_latency_seconds：Histogram 类型，以秒为单位记录推理响应时间
model_prediction_count_total：Counter 类型，统计预测为生还（1）与死亡（0）各自的次数

`monitoring/prometheus.yml`

global:
  scrape_interval: 5s

scrape_configs:
  - job_name: "titanic-api"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["titanic-api:8000"]

Prometheus 每 5 秒从 FastAPI 的 /metrics 端点拉取一次数据，并存储为时间序列。

为什么这在生产环境中很重要

假设你的模型已经部署上线，每天要处理 1 万次请求。Prometheus 能帮你发现这些问题：

流量激增：请求数突然涨到原来的 10 倍——这是自然增长，还是机器人攻击？
延迟劣化：平均推理时间从 15ms 飙升到 800ms——说明服务器内存不够用了。
预测分布偏移：平时 survived=0 的预测只占 38%，突然变成 99%——说明输入数据出了问题。

查询 Prometheus

通过 API 跑几次预测之后，访问 http://localhost:9090，然后查询：

api_request_count_total
model_prediction_count_total
rate(api_request_latency_seconds_sum[5m]) / rate(api_request_latency_seconds_count[5m])

第 12 步：使用 Evidently AI 检测数据漂移

什么是数据漂移？

一个在 1912 年泰坦尼克号乘客数据上训练出来的模型，会默认某些人口结构分布——中位年龄约 28 岁、以三等舱乘客为主、以男性为主。如果模型部署上线后，线上流量突然发生变化（比如进来的乘客全是 60 岁以上、票价超过 300 美元），模型并不会崩溃——它只会悄无声息地输出不可靠、不准确的预测结果。

这种现象叫做数据漂移（Data Drift，也称协变量偏移 Covariate Shift），它是生产级机器学习系统的头号隐形杀手。

`src/drift_detection.py`

import os
import sys
import logging
import pandas as pd

# Support both new and older Evidently versions
try:
    from evidently.report import Report
    from evidently.metric_preset import DataDriftPreset, DataQualityPreset
except ModuleNotFoundError:
    from evidently.legacy.report import Report
    from evidently.legacy.metric_preset import DataDriftPreset, DataQualityPreset

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data
from src.data_preprocess import preprocess_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def generate_drift_report(
    reference_data: pd.DataFrame, 
    current_data: pd.DataFrame, 
    output_html_path: str = "monitoring/drift_report.html"
):
    """
    Compares reference (training) data with current (production) data
    and generates an interactive HTML drift dashboard.
    """
    logging.info("Building Evidently Data Drift and Data Quality Report...")

    features = [col for col in reference_data.columns if col != "Survived"]
    ref_df = reference_data[features]
    curr_df = current_data[features]

    report = Report(metrics=[
        DataDriftPreset(),
        DataQualityPreset()
    ])

    report.run(reference_data=ref_df, current_data=curr_df)

    os.makedirs(os.path.dirname(output_html_path), exist_ok=True)
    report.save_html(output_html_path)

    logging.info(f"Drift report successfully generated at: {output_html_path}")
    return output_html_path


if __name__ == "__main__":
    # Load baseline reference data
    raw_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_path)
    reference_df = preprocess_data(df)

    # Simulate "Current" production data with intentional drift!
    current_df = reference_df.copy()
    current_df["Age"] = current_df["Age"] + 25.0  # Simulated drift in Age
    current_df["Fare"] = current_df["Fare"] * 3.5  # Simulated drift in Fare

    # Generate drift report
    report_file = generate_drift_report(reference_df, current_df)
    print(f"\nReport ready! Open in browser: {os.path.abspath(report_file)}")

Evidently 是如何检测漂移的

Evidently 会对每一个特征运行统计检验：

数值特征（Age、Fare）：Kolmogorov-Smirnov 检验——判断两个分布是否来自同一个总体
类别特征（Pclass、Sex_male）：卡方检验——判断类别占比是否发生了变化

如果 p 值低于阈值（默认 0.05），该特征就会被标记为已漂移。

运行脚本：

python src/drift_detection.py
# Drift report successfully generated at: monitoring/drift_report.html

在浏览器中打开 monitoring/drift_report.html，就能看到 Evidently 的交互式仪表盘，里面有每个特征的分布图、p 值以及漂移标记。

完整全景

让我们退后一步，看看这一路搭建出来的整套体系：

┌──────────────────────────────────────────────────────┐
│                  Developer Pushes Code               │
└────────────────────────┬─────────────────────────────┘
                         │
                         ▼
┌──────────────────────────────────────────────────────┐
│              GitHub Actions CI/CD Pipeline            │
│                                                      │
│     Install Dependencies                             │
│     Validate Data Schema                             │
│     Train Model + Log to MLflow                      │
│     Run Pytest Suite (Model + API tests)             │
│     Build Docker Image                               │
└────────────────────────┬─────────────────────────────┘
                         │
                         ▼
┌──────────────────────────────────────────────────────┐
│            Docker Container (Production)             │
│                                                      │
│  FastAPI (:8000)                                     │
│  ├── POST /predict  → Model Inference                │
│  ├── GET  /metrics  → Prometheus Scrape Endpoint     │
│  └── GET  /docs     → Interactive Swagger UI         │
└────────────────────────┬─────────────────────────────┘
                         │
              ┌──────────┴──────────┐
              ▼                     ▼
┌────────────────────┐  ┌────────────────────┐
│  Prometheus (:9090) │  │  Evidently AI       │
│                    │  │                    │
│  Request Counts    │  │  Data Drift        │
│  Latency Tracking  │  │  Feature Drift     │
│  Prediction Dist.  │  │  Quality Metrics   │
└────────────────────┘  └────────────────────┘

核心要点

数据版本管理（DVC）：永远不要把原始数据直接提交到 Git。用 DVC 跟踪数据，保证可复现和可回滚。
训练前先做校验：Schema 检查、约束检查和缺失值防护，能在数据问题悄悄污染模型之前就把它们拦下来。
记录每一次实验（MLflow）：每次训练运行都应该记录参数、指标和产物。“我记得上周二跑出了 82% 的准确率”这种话在生产环境是不可接受的。
用 API 而不是脚本对外提供服务：终端用户发的是 HTTP 请求。FastAPI 加上 Pydantic 校验，可以得到类型安全、自带文档的端点。
一切测试自动化：用 Pytest 做模型完整性测试 + API 契约测试，并在 CI 中自动执行。
容器化保证可复现：Docker 彻底消灭了“在我机器上能跑”的问题。
用 CI/CD 实现自动化：GitHub Actions 确保每次推送在进入生产环境之前，都经过校验、测试、训练和容器构建。
在生产环境做监控：Prometheus 让你实时掌握流量、延迟和预测行为。没有它，你就等于在盲飞。
检测数据漂移：Evidently AI 能抓住 ML 系统的隐形杀手——生产数据的分布偏离了模型训练时的分布。

后续改进方向

自动重训练流水线：当 Evidently 检测到显著漂移时，自动触发模型重训练。
特征库集成：用 Feast 或 Hopsworks 做集中式特征管理。
A/B 测试：同时部署两个模型版本，对比线上实际表现。
Kubernetes 部署：从单个 Docker 容器扩展到编排集群。
Grafana 仪表盘：在 Prometheus 指标之上加一层美观的可视化。
实时流处理：用实时事件驱动的预测取代批量推理。

参考资料

原文：https://dev.to/naman_2004/building-a-production-grade-end-to-end-mlops-pipeline-from-scratch-l9h（作者 @naman_2004）

MLOps 实战：从零搭建生产级端到端流水线（DVC+MLflow+FastAPI+Docker）

从零搭建生产级端到端 MLOps 流水线

架构总览

技术栈

项目结构

第 1 步：项目初始化与环境配置

.gitignore

requirements.txt

第 2 步：用 DVC 做数据版本管理

为什么不直接把数据集提交到 Git？

配置步骤

步骤 3：数据摄取（src/data_ingestion.py）

为什么用 logging 而不是 print()？

步骤 4：数据校验（src/data_validation.py）

我们到底在防御什么？

第 5 步：数据预处理与特征工程（src/data_preprocess.py）

关键决策解析

第 6 步：用 MLflow 进行模型训练与实验跟踪（src/train.py）

MLflow 给我们带来了什么

第 7 步：FastAPI 推理服务（api/main.py）

特征对齐陷阱

API 示例

第 8 步：使用 Pytest 进行自动化测试

tests/test_model.py —— 模型完整性测试

tests/test_api.py —— API 契约测试

第 9 步：Docker 容器化

为什么要用 Docker？

Dockerfile

层缓存技巧

.dockerignore

docker-compose.yml

构建与运行

第 10 步：使用 GitHub Actions 实现 CI/CD

.github/workflows/ci_cd.yml

关于 PYTHONPATH: . 的修复

第 11 步：用 Prometheus 做生产环境监控

Prometheus 追踪哪些指标？

monitoring/prometheus.yml

为什么这在生产环境中很重要

查询 Prometheus

第 12 步：使用 Evidently AI 检测数据漂移

什么是数据漂移？

src/drift_detection.py

Evidently 是如何检测漂移的

完整全景

核心要点

后续改进方向

参考资料

`.gitignore`

`requirements.txt`

步骤 3：数据摄取（`src/data_ingestion.py`）

为什么用 `logging` 而不是 `print()`？

步骤 4：数据校验（`src/data_validation.py`）

第 5 步：数据预处理与特征工程（`src/data_preprocess.py`）

第 6 步：用 MLflow 进行模型训练与实验跟踪（`src/train.py`）

第 7 步：FastAPI 推理服务（`api/main.py`）

`tests/test_model.py` —— 模型完整性测试

`tests/test_api.py` —— API 契约测试

`Dockerfile`

`.dockerignore`

`docker-compose.yml`

`.github/workflows/ci_cd.yml`

关于 `PYTHONPATH: .` 的修复

`monitoring/prometheus.yml`

`src/drift_detection.py`