09-模型部署与全链路优化：从实验室到产线的最后一公里

第 9 篇：模型部署与全链路优化——从实验室到产线的最后一公里

本文为「从零到落地：机器学习分析数据实战系列」第 9 篇，完整系列持续更新中。

---

前言

前四篇跑通了四大业务场景的模型，但模型做得再好，不能上线也是白搭。这篇解决最实际的问题：怎么把 Jupyter Notebook 里的模型，变成产线上 7×24 小时运行的服务？

回顾一下第 5-8 篇的成果：

• 第 5 篇：XGBoost 故障预警，召回率 71%
• 第 6 篇：随机森林多分类 + SHAP 根因分析
• 第 7 篇：Optuna 贝叶斯优化最优工艺参数
• 第 8 篇：LSTM 预测涡扇发动机剩余寿命（RUL），RMSE ~25

这些模型都跑在 Jupyter Notebook 里——交互式运行、手动加载数据、一次性出结果。但在工厂里，你需要的是一个7×24 小时不间断运行的服务：传感器数据实时流入 → 自动计算特征 → 模型推理 → 超过阈值就报警 → 运维人员手机上收到通知。

从 Notebook 到产线服务，中间隔着四大工程问题：

1. 模型怎么保存和加载？ —— 序列化（ONNX / Joblib）
2. 其他系统怎么调用模型？ —— API 服务化（FastAPI）
3. 数据怎么实时进来？ —— 数据管道（MQTT / Kafka）
4. 上线后怎么知道模型还准不准？ —— 监控与维护

本篇按这个顺序逐一解决，最后给出一套完整的设备故障预警系统部署方案。

本篇学完你将掌握：

• 模型序列化的三种方案（Joblib / Pickle / ONNX）及各自适用场景
• 用 FastAPI 把模型封装成 HTTP 推理服务
• 用 Docker 容器化部署，实现「一次构建，到处运行」
• MQTT / Kafka 实时数据管道的原理和基础用法
• 模型漂移检测与自动重训练触发机制
• A/B 测试和影子模式的上线策略
• 一套完整的故障预警系统部署架构

---

一、环境与工具准备

1.1 安装部署相关依赖

conda activate ml-data

# API 服务框架
pip install fastapi==0.111.* uvicorn[standard]==0.30.*

# 模型序列化
pip install onnx==1.16.* onnxruntime==1.18.* skl2onnx==1.17.*

# 消息队列（轻量级 MQTT）
pip install paho-mqtt==2.1.*

# 监控与日志
pip install prometheus-client==0.20.*

# HTTP 客户端（测试 API 用）
pip install httpx==0.27.*

💡 提示：Docker 需要单独安装（不是 pip 包）。Windows 用户安装 Docker Desktop，macOS 同理。Linux 用户按发行版安装 Docker Engine 即可。安装后在终端跑 docker --version 确认版本 ≥ 24.0。

1.2 本篇用到的前置模型

本篇不训练新模型，而是复用前几篇训练好的模型。为了完整演示，我们先训练一个 XGBoost 故障预警模型并保存：

import pandas as pd
import numpy as np
import warnings
warnings.filterwarnings('ignore')

# ---------- 复用第 5 篇的数据加载和特征工程函数 ----------
def load_and_prepare(filepath='ai4i2020.csv'):
    """加载 AI4I 2020 数据集并完成特征工程。"""
    df = pd.read_csv(filepath)
    
    # 构造特征（和第 4-5 篇一致）
    df['Temperature_diff'] = df['Process temperature [K]'] - df['Air temperature [K]']
    df['Power_approx'] = df['Rotational speed [rpm]'] * df['Torque [Nm]']
    df['Speed_torque_ratio'] = df['Rotational speed [rpm]'] / (df['Torque [Nm]'] + 1e-8)
    
    # 磨损阶段分箱
    bins = [0, 50, 120, 180, 253]
    labels = [0, 1, 2, 3]
    df['Wear_stage_code'] = pd.cut(df['Tool wear [min]'], bins=bins, labels=labels).astype(float)
    
    # Type 编码
    type_map = {'L': 0, 'M': 1, 'H': 2}
    df['Type_code'] = df['Type'].map(type_map)
    
    return df

df = load_and_prepare()

# RFE 选中的 8 个特征
FEATURES = [
    'Air temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]',
    'Tool wear [min]', 'Temperature_diff', 'Power_approx',
    'Speed_torque_ratio', 'Wear_stage_code'
]

X = df[FEATURES].values
y = df['Machine failure'].values

# ---------- 训练模型 ----------
from sklearn.model_selection import train_test_split
from xgboost import XGBClassifier

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

model = XGBClassifier(
    n_estimators=200, max_depth=5, learning_rate=0.1,
    scale_pos_weight=28, eval_metric='logloss', random_state=42
)
model.fit(X_train, y_train)

print(f"训练完成 | 测试集 F1: {model.score(X_test, y_test):.4f}")
print(f"特征数: {len(FEATURES)}")

训练完成 | 测试集 F1: 0.9711
特征数: 8

📌 注意：这里的 F1 用的是 model.score()（即准确率），不是严格的 F1 Score。后面部署时会用专门的评估函数。这个模型和第 5 篇的配置完全一致，可以直接复用。

---

二、业务理解：从 Notebook 到产线，差距有多大？

2.1 实验室模型 vs 产线系统

很多人以为训练好模型就万事大吉了。但在实际工厂里，模型只是一个组件，不是系统。两者的差距：

维度	Notebook 里的模型	产线上的系统
运行方式	手动执行，跑一次出一次结果	7×24 小时自动运行
数据来源	手动加载 CSV	传感器实时流式推送
调用方式	只有你能用	其他系统通过 API 调用
异常处理	报错了手动排查	自动报警 + 降级策略
模型更新	手动重训练 + 手动替换	自动检测漂移 + 自动重训练
部署环境	你的笔记本	服务器 / 边缘设备 / 容器

类比：Notebook 模型就像你在家里做饭——自己买菜、自己切、自己炒、自己吃。产线系统就像开餐厅——标准化流程、批量出餐、持续供应、食品安全监控。食材（数据）和菜谱（模型）是一样的，但工程复杂度完全不同。

2.2 部署全链路路线图

本篇按模型上线的 4 个阶段逐步推进，每个阶段解决一个核心问题：

阶段 1：模型序列化 → 把模型从 Python 对象变成可传输、可加载的文件
    ↓
阶段 2：API 服务化 → 让其他系统能通过 HTTP 调用模型
    ↓
阶段 3：容器化部署 → 打包成 Docker 镜像，一次构建到处运行
    ↓
阶段 4：监控与迭代 → 上线后持续监控效果，发现漂移自动重训练

每个阶段都有独立的代码和验证方法，你可以按需学习。

---

三、模型序列化与推理服务

3.1 为什么要序列化？

模型训练完后，它存在于 Python 进程的内存里——关掉 Jupyter 就没了。要让它持久运行，第一步是把模型保存到磁盘上，需要时再加载。

这就像你在 Notebook 里写了一个很好的函数，要给别人用时，不能让对方打开你的 Notebook——你需要把它打包成一个库（pip 包）或者服务（API），别人才能调用。

3.2 三种序列化方案对比

方案	格式	优点	缺点	适用场景
Joblib	.joblib	scikit-learn 官方推荐，保存完整 Python 对象	只能在 Python 里加载	快速原型、纯 Python 环境
Pickle	.pkl	Python 内置，任何 Python 对象都能序列化	安全风险（反序列化可执行任意代码）	内部系统、可信环境
ONNX	.onnx	跨语言（Python/Java/C++/Go），推理速度快	转换过程可能丢失部分操作	生产部署、跨平台

什么是 ONNX？

ONNX（Open Neural Network Exchange）是一个开放的模型交换格式。你可以把它理解为「模型的 PDF」——不管你用什么框架（PyTorch、TensorFlow、scikit-learn）训练的模型，转成 ONNX 后，任何支持 ONNX 的语言和平台都能加载和推理。

它的价值在于解耦：训练用 Python，推理可以用 C++（更快）、Java（企业系统）、甚至 JavaScript（浏览器端）。

3.3 实操：保存模型

import joblib
import os
import numpy as np

os.makedirs('models', exist_ok=True)

# ---------- 方案一：Joblib（最简单） ----------
joblib.dump(model, 'models/fault_xgb.joblib')
print("✅ Joblib 保存完成")

# ---------- 方案二：Pickle ----------
import pickle
with open('models/fault_xgb.pkl', 'wb') as f:
    pickle.dump(model, f)
print("✅ Pickle 保存完成")

# ---------- 方案三：ONNX（生产推荐） ----------
from skl2onnx import to_onnx
from skl2onnx.common.data_types import FloatTensorType

# 定义输入类型（8 个 float 特征）
initial_type = [('float_input', FloatTensorType([None, len(FEATURES)]))]

# 转换为 ONNX 格式
onnx_model = to_onnx(model, initial_types=initial_type, target_opset=15)

with open('models/fault_xgb.onnx', 'wb') as f:
    f.write(onnx_model.SerializeToString())
print("✅ ONNX 保存完成")

# ---------- 对比文件大小 ----------
for f in ['models/fault_xgb.joblib', 'models/fault_xgb.pkl', 'models/fault_xgb.onnx']:
    size_kb = os.path.getsize(f) / 1024
    print(f"  {os.path.basename(f):25s} {size_kb:>8.1f} KB")

✅ Joblib 保存完成
✅ Pickle 保存完成
✅ ONNX 保存完成
  fault_xgb.joblib             145.3 KB
  fault_xgb.pkl                145.1 KB
  fault_xgb.onnx               358.7 KB

📌 注意：ONNX 文件比 Joblib 大是正常的——ONNX 包含了完整的计算图描述和元数据，而 Joblib 只保存了模型参数。文件大小的差异在推理速度上会反转：ONNX Runtime 做了大量底层优化，推理通常比原生 scikit-learn 更快。

3.4 加载与验证

保存后必须验证：加载回来的模型和原模型预测结果一致吗？

# ---------- 验证 Joblib ----------
model_joblib = joblib.load('models/fault_xgb.joblib')

# ---------- 验证 ONNX ----------
import onnxruntime as ort

session = ort.InferenceSession('models/fault_xgb.onnx')
input_name = session.get_inputs()[0].name

# 用测试数据的前 5 条做验证
sample = X_test[:5].astype(np.float32)

pred_joblib = model_joblib.predict(sample)
pred_onnx = session.run(None, {input_name: sample})[0].argmax(axis=1)

print("验证前 5 条样本：")
print(f"  Joblib 预测: {pred_joblib}")
print(f"  ONNX  预测: {pred_onnx}")
print(f"  一致: {np.array_equal(pred_joblib, pred_onnx)}")

验证前 5 条样本：
  Joblib 预测: [0 0 0 0 0]
  ONNX  预测: [0 0 0 0 0]
  一致: True

---

四、API 服务化——让其他系统调用模型

4.1 为什么需要 API？

模型保存在磁盘上后，下一步是让其他系统能调用它。想象一下工厂里的场景：

• MES 系统（制造执行系统）想知道当前工件是否可能出故障 → 调模型
• SCADA 系统（数据采集与监控系统）想实时显示设备健康度 → 调模型
• 运维 App 想在手机上收到告警推送 → 调模型

这些系统可能用 Java、C#、甚至 JavaScript 开发，不可能都跑 Python。所以需要一个标准化的接口——HTTP API。任何语言都能发 HTTP 请求，这就是 API 服务化的价值。

什么是 FastAPI？

FastAPI 是 Python 最现代的 Web 框架，专为构建 API 设计。它的核心优势：

• 自动生成文档：写完代码，Swagger UI 就有了，前端工程师可以直接对着文档调试
• 类型校验：请求参数不对时自动返回 422 错误，不用手动写校验代码
• 异步支持：天然支持高并发请求

如果你用过 Flask，FastAPI 的用法类似但更现代（类型提示 + 自动文档）。

4.2 构建推理 API

创建一个独立的 Python 文件 predict_api.py：

"""
设备故障预警推理 API
启动命令: uvicorn predict_api:app --host 0.0.0.0 --port 8000
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import joblib
import numpy as np
import time

# ---------- 创建应用 ----------
app = FastAPI(
    title="设备故障预警 API",
    description="基于 XGBoost 的设备故障二分类预测服务",
    version="1.0.0"
)

# ---------- 定义请求/响应模型 ----------
# Pydantic 模型会自动做数据校验和生成文档

class SensorReading(BaseModel):
    """单次传感器读数"""
    air_temperature: float       # 环境温度 [K]
    process_temperature: float   # 加工温度 [K]
    rotational_speed: float      # 转速 [rpm]
    torque: float                # 扭矩 [Nm]
    tool_wear: float             # 刀具磨损时间 [min]
    product_type: Optional[str] = "M"  # 产品规格: L/M/H

class PredictionResponse(BaseModel):
    """预测结果"""
    failure_predicted: bool      # 是否预测故障
    failure_probability: float   # 故障概率
    health_score: float          # 健康度 (1 - 故障概率)
    inference_time_ms: float     # 推理耗时

# ---------- 启动时加载模型 ----------
FEATURES = [
    'Air temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]',
    'Tool wear [min]', 'Temperature_diff', 'Power_approx',
    'Speed_torque_ratio', 'Wear_stage_code'
]

@app.on_event("startup")
def load_model():
    """服务启动时加载模型到全局变量"""
    global model
    model = joblib.load('models/fault_xgb.joblib')
    print(f"✅ 模型加载完成")

# ---------- 特征工程函数 ----------
def build_features(reading: SensorReading) -> np.ndarray:
    """从原始传感器读数构造特征（和第 4 篇一致）"""
    temp_diff = reading.process_temperature - reading.air_temperature
    power = reading.rotational_speed * reading.torque
    speed_torque_ratio = reading.rotational_speed / (reading.torque + 1e-8)
    
    # 磨损阶段
    wear = reading.tool_wear
    if wear <= 50:
        wear_stage = 0
    elif wear <= 120:
        wear_stage = 1
    elif wear <= 180:
        wear_stage = 2
    else:
        wear_stage = 3
    
    features = np.array([[
        reading.air_temperature,
        reading.rotational_speed,
        reading.torque,
        reading.tool_wear,
        temp_diff,
        power,
        speed_torque_ratio,
        wear_stage
    ]])
    return features

# ---------- 推理接口 ----------
@app.post("/predict", response_model=PredictionResponse)
def predict(reading: SensorReading):
    """
    输入传感器读数，返回故障预测结果。
    
    - **failure_predicted**: True 表示预测为故障
    - **failure_probability**: 故障概率 (0~1)
    - **health_score**: 健康度 (0~1)，越高越好
    """
    start = time.time()
    
    # 构造特征
    features = build_features(reading)
    
    # 推理
    prob = model.predict_proba(features)[0][1]  # 故障概率
    pred = int(prob > 0.5)
    
    elapsed = (time.time() - start) * 1000
    
    return PredictionResponse(
        failure_predicted=bool(pred),
        failure_probability=round(float(prob), 4),
        health_score=round(float(1 - prob), 4),
        inference_time_ms=round(elapsed, 2)
    )

# ---------- 批量推理接口 ----------
@app.post("/predict/batch")
def predict_batch(readings: List[SensorReading]):
    """批量预测，一次传入多条传感器读数"""
    results = []
    for reading in readings:
        features = build_features(reading)
        prob = model.predict_proba(features)[0][1]
        results.append({
            "failure_predicted": bool(prob > 0.5),
            "failure_probability": round(float(prob), 4),
            "health_score": round(float(1 - prob), 4)
        })
    return {"predictions": results, "count": len(results)}

# ---------- 健康检查 ----------
@app.get("/health")
def health_check():
    """服务健康检查接口，供 Kubernetes/Docker 使用"""
    return {"status": "ok", "model_loaded": True}

4.3 启动并测试 API

启动服务：

# 在包含 predict_api.py 的目录下运行
uvicorn predict_api:app --host 0.0.0.0 --port 8000 --reload

💡 --reload 参数只在开发时使用（修改代码后自动重启），生产环境应去掉。

测试 API（在另一个终端或 Notebook 中）：

import httpx

# ---------- 测试单条预测 ----------
# 模拟一条正常运行数据
response = httpx.post("http://localhost:8000/predict", json={
    "air_temperature": 300.0,
    "process_temperature": 310.0,
    "rotational_speed": 1500,
    "torque": 40,
    "tool_wear": 50,
    "product_type": "M"
})

result = response.json()
print(f"状态码: {response.status_code}")
print(f"预测结果: {'故障 ⚠️' if result['failure_predicted'] else '正常 ✅'}")
print(f"故障概率: {result['failure_probability']}")
print(f"健康度:   {result['health_score']}")
print(f"推理耗时: {result['inference_time_ms']} ms")

# ---------- 测试一条高风险数据 ----------
response_risk = httpx.post("http://localhost:8000/predict", json={
    "air_temperature": 303.0,
    "process_temperature": 314.0,
    "rotational_speed": 2800,     # 高转速
    "torque": 70,                 # 高扭矩
    "tool_wear": 220,             # 严重磨损
    "product_type": "H"
})

result_risk = response_risk.json()
print(f"\n高风险样本:")
print(f"预测结果: {'故障 ⚠️' if result_risk['failure_predicted'] else '正常 ✅'}")
print(f"故障概率: {result_risk['failure_probability']}")

状态码: 200
预测结果: 正常 ✅
故障概率: 0.0048
健康度:   0.9952
推理耗时: 0.85 ms

高风险样本:
预测结果: 故障 ⚠️
故障概率: 0.9863

4.4 自动生成 API 文档

FastAPI 最大的优势之一是自动生成 API 文档。启动服务后，在浏览器打开：

http://localhost:8000/docs

你会看到一个 Swagger UI 页面，里面列出了所有接口（/predict、/predict/batch、/health），每个接口的请求参数、响应格式都写得清清楚楚。前端工程师或者其他系统的开发者可以直接在这个页面上测试接口，不需要你额外写文档。

📌 为什么自动文档很重要？ 在团队协作中，API 文档是前后端沟通的桥梁。手动写文档容易过时（代码改了文档没改），而 FastAPI 的文档是从代码自动生成的——代码就是文档，永远不会不一致。

---

五、容器化部署——Docker

5.1 为什么需要容器化？

API 跑在你的电脑上没问题，但部署到服务器时可能遇到：

• 服务器上 Python 版本不对
• 服务器上缺少某个依赖包
• 你的电脑能跑，别人的跑不了
• 想部署 3 台服务器做负载均衡，每台都要手动配置一遍

Docker 解决的正是这个问题：把代码、模型、Python 环境、所有依赖打包成一个镜像，任何装了 Docker 的机器都能一键运行，环境完全一致。

什么是 Docker？

你可以把 Docker 理解为「轻量级虚拟机」——它把你的应用和运行环境打包成一个容器（Container），和宿主系统隔离。但和传统虚拟机不同，Docker 容器共享宿主机内核，启动只需要几秒（虚拟机要几分钟）。

核心概念只有三个：

• 镜像（Image）：打包好的应用 + 环境，类似于一个安装程序
• 容器（Container）：镜像运行起来的实例，类似于安装后的程序
• Dockerfile：构建镜像的配方文件，告诉 Docker 怎么一步步构建

5.2 编写 Dockerfile

在项目根目录创建 Dockerfile：

# ---------- 基础镜像 ----------
# 使用 Python 3.11 slim 版本（约 150MB，比完整版小很多）
FROM python:3.11-slim

# ---------- 工作目录 ----------
WORKDIR /app

# ---------- 安装依赖 ----------
# 先复制依赖文件，利用 Docker 缓存（依赖不变时不会重新安装）
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# ---------- 复制代码和模型 ----------
COPY predict_api.py .
COPY models/ models/

# ---------- 暴露端口 ----------
EXPOSE 8000

# ---------- 启动命令 ----------
# 生产环境用 4 个 worker 进程，提高并发能力
CMD ["uvicorn", "predict_api:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

配套的 requirements.txt：

fastapi==0.111.*
uvicorn[standard]==0.30.*
joblib==1.4.*
numpy==1.26.*
scikit-learn==1.5.*
xgboost==2.1.*
pydantic==2.*

📌 生产细节：--workers 4 表示启动 4 个 worker 进程。每个 worker 独立加载模型，可以并行处理请求。worker 数量一般设为 CPU 核数的 1-2 倍。如果内存不够（每个 worker 都加载一份模型），可以减少 worker 数或换用 ONNX Runtime（内存占用更小）。

5.3 构建并运行

# 构建镜像（-t 指定名称和版本标签）
docker build -t fault-predictor:v1.0 .

# 运行容器
# -d: 后台运行
# -p: 端口映射（宿主机 8000 → 容器 8000）
# --name: 容器名称
docker run -d -p 8000:8000 --name fault-predictor fault-predictor:v1.0

# 查看运行状态
docker ps

CONTAINER ID   IMAGE                    STATUS        PORTS                    NAMES
a1b2c3d4e5f6   fault-predictor:v1.0     Up 5 seconds  0.0.0.0:8000->8000/tcp   fault-predictor

验证容器中的 API：

# 和之前的测试代码完全一样——API 接口不变，只是运行环境从本地换成了容器
response = httpx.post("http://localhost:8000/predict", json={
    "air_temperature": 303.0,
    "process_temperature": 314.0,
    "rotational_speed": 2800,
    "torque": 70,
    "tool_wear": 220,
    "product_type": "H"
})
print(response.json())

{'failure_predicted': True, 'failure_probability': 0.9863, 'health_score': 0.0137, 'inference_time_ms': 0.92}

5.4 Docker 常用运维命令

# 查看容器日志
docker logs fault-predictor

# 进入容器排查问题（类似 SSH 登录）
docker exec -it fault-predictor bash

# 停止 / 启动 / 删除容器
docker stop fault-predictor
docker start fault-predictor
docker rm fault-predictor

# 更新模型（构建新版本镜像 → 停掉旧容器 → 启动新容器）
docker build -t fault-predictor:v1.1 .
docker stop fault-predictor && docker rm fault-predictor
docker run -d -p 8000:8000 --name fault-predictor fault-predictor:v1.1

📌 生产建议：在实际项目中，镜像版本标签不要用 latest，而是用语义化版本（如 v1.0、v1.1）或 Git commit hash。这样出问题时可以快速回滚到上一个版本。

---

六、实时数据管道

6.1 从「被动查询」到「主动推送」

到目前为止，API 是「被动」的——需要别的系统主动发 HTTP 请求才能获取预测结果。但在工厂里，传感器数据是持续不断地产生的：

• 温度传感器每秒上报一次
• 振动传感器每秒上报几百次
• 设备 PLC（可编程逻辑控制器）每隔几百毫秒上报一次状态

这些数据不可能每次都手动发 HTTP 请求。需要一套数据管道，让传感器数据自动流入、自动处理、自动触发预测。

6.2 MQTT vs Kafka：选哪个？

维度	MQTT	Kafka
定位	轻量级物联网消息协议	企业级分布式消息系统
消息量	适合中小规模（每秒几千条）	适合超大规模（每秒百万条）
部署复杂度	低，一个 Mosquitto 就行	高，需要 ZooKeeper/KRaft 集群
适用场景	设备传感器直连	多系统对接、数据湖、日志聚合
学习成本	10 分钟上手	需要理解 Topic/Partition/Consumer Group

类比：MQTT 像快递驿站——设备把数据（包裹）送到驿站，订阅者来取。Kafka 像物流分拣中心——海量包裹按目的地分拣，多个快递员同时取件配送。小型工厂用 MQTT 足够，大型集团用 Kafka。

本篇用 MQTT 做演示（轻量、好理解），Kafka 的原理类似，只是配置更复杂。

6.3 MQTT 基础概念

MQTT 只有三个角色：

1. 发布者（Publisher）：传感器 / PLC 把数据发到某个 Topic（主题）
2. 代理（Broker）：消息中转站，收到消息后转发给订阅者
3. 订阅者（Subscriber）：我们的推理服务，订阅特定 Topic，收到数据后触发预测

Topic 是一个字符串路径，类似于文件系统的目录结构：

factory/line1/cnc01/temperature   → 1号产线 1号CNC 的温度数据
factory/line1/cnc01/vibration     → 1号产线 1号CNC 的振动数据
factory/line2/cnc03/temperature   → 2号产线 3号CNC 的温度数据

6.4 模拟传感器数据发布

先写一个模拟传感器，每隔 2 秒发送一条设备运行数据：

"""
模拟传感器：定时向 MQTT Broker 发送设备运行数据
"""
import paho.mqtt.client as mqtt
import json
import time
import random

# ---------- 连接 MQTT Broker ----------
# 使用公共测试 Broker（仅用于开发测试，生产环境应自建）
broker = "broker.emqx.io"
port = 1883
topic = "factory/line1/cnc01/sensors"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sensor_simulator")
client.connect(broker, port)
client.loop_start()
print(f"✅ 已连接 MQTT Broker: {broker}")

# ---------- 模拟正常 → 逐渐恶化的运行数据 ----------
print("\n开始模拟传感器数据发布...")

for i in range(20):
    # 模拟设备从正常逐渐恶化的过程
    wear = i * 12  # 刀具磨损逐渐增加
    temp_rise = i * 0.3  # 温度逐渐升高
    speed = 1500 + random.gauss(0, 30)  # 转速有正常波动
    
    reading = {
        "timestamp": time.time(),
        "device_id": "CNC-01",
        "air_temperature": round(300.0 + temp_rise, 1),
        "process_temperature": round(310.0 + temp_rise * 1.5, 1),
        "rotational_speed": round(speed, 0),
        "torque": round(40 + i * 1.5 + random.gauss(0, 2), 1),
        "tool_wear": wear,
        "product_type": "M"
    }
    
    # 发布到 Topic
    client.publish(topic, json.dumps(reading))
    
    status = "🟢 正常" if i < 12 else "🟡 注意" if i < 17 else "🔴 高风险"
    print(f"  [{i+1:>2d}/20] {status} | 温度={reading['process_temperature']:.1f}K  "
          f"转速={reading['rotational_speed']:.0f}rpm  扭矩={reading['torque']:.1f}Nm  "
          f"磨损={reading['tool_wear']}min")
    
    time.sleep(2)

client.loop_stop()
client.disconnect()
print("\n✅ 模拟结束")

6.5 订阅并实时推理

推理服务作为 MQTT 订阅者，收到数据后自动调用模型预测：

"""
实时推理服务：订阅 MQTT Topic，收到数据后自动预测
"""
import paho.mqtt.client as mqtt
import json
import joblib
import numpy as np

# ---------- 加载模型 ----------
model = joblib.load('models/fault_xgb.joblib')
print("✅ 模型加载完成，等待传感器数据...")

# ---------- 特征构造 ----------
def build_features(data):
    temp_diff = data['process_temperature'] - data['air_temperature']
    power = data['rotational_speed'] * data['torque']
    speed_torque_ratio = data['rotational_speed'] / (data['torque'] + 1e-8)
    wear = data['tool_wear']
    wear_stage = 0 if wear <= 50 else 1 if wear <= 120 else 2 if wear <= 180 else 3
    return np.array([[
        data['air_temperature'], data['rotational_speed'], data['torque'],
        data['tool_wear'], temp_diff, power, speed_torque_ratio, wear_stage
    ]])

# ---------- MQTT 回调函数 ----------
alert_count = 0  # 连续异常计数（和第 5 篇预警策略一致）

def on_message(client, userdata, msg):
    global alert_count
    
    data = json.loads(msg.payload.decode())
    features = build_features(data)
    prob = model.predict_proba(features)[0][1]
    
    # 三级告警逻辑
    if prob > 0.8:
        alert_count += 1
        level = "🔴 紧急" if alert_count >= 3 else "🟠 警告"
        action = " → 立即停机！" if alert_count >= 3 else ""
        print(f"  {level} | P(故障)={prob:.4f} | 连续异常={alert_count}次{action}")
    elif prob > 0.3:
        alert_count = max(0, alert_count - 1)  # 偶尔正常则递减
        print(f"  🟡 注意 | P(故障)={prob:.4f} | 建议关注")
    else:
        alert_count = 0
        print(f"  🟢 正常 | P(故障)={prob:.4f}")

# ---------- 连接并订阅 ----------
broker = "broker.emqx.io"
topic = "factory/line1/cnc01/sensors"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="inference_service")
client.on_message = on_message
client.connect(broker, 1883)
client.subscribe(topic)

print(f"✅ 已订阅 Topic: {topic}")
print("等待数据中...\n")

# 阻塞运行（实际部署时用 systemd 或 Docker 管理进程）
client.loop_forever()

📌 生产注意：上面用的 broker.emqx.io 是公共测试 Broker，任何人都能发消息，不适合生产环境。实际工厂应自建 MQTT Broker（推荐 EMQX 或 Mosquitto），并配置 TLS 加密和认证。

---

七、模型监控与维护

7.1 为什么上线后还要监控？

模型部署不是终点，而是新的起点。在实验室里，模型在测试集上效果好，不代表上线后一直好。原因是数据分布会变化：

• 季节变化：夏天气温 35°C，冬天 5°C，传感器读数的基准值不同
• 设备老化：运行一年后，同工况下的振动、温度特征会漂移
• 工艺变更：换了新材料、新产品，旧的故障模式可能消失，新的可能出现
• 传感器退化：传感器本身也会老化，读数精度下降

这种现象叫模型漂移（Model Drift）——模型没变，但世界变了，模型的预测效果就下降了。

类比：你训练了一个识别猫狗的模型，效果 95%。但如果用户开始上传仓鼠的照片，模型从来没见过仓鼠，预测效果就会下降。这就是数据漂移。

7.2 监控什么？

监控维度	具体指标	怎么监控	阈值示例
预测分布	故障预测占比是否异常波动	每小时统计一次	正常时故障率 3-5%，突然飙到 30% → 异常
推理延迟	单次推理耗时	记录每次请求的耗时	P99 > 100ms → 需要优化
特征分布	输入特征的均值/方差	对比训练集统计量	特征均值偏移 > 2σ → 可能漂移
模型准确率	预测 vs 实际结果	延迟反馈（维修记录回来后对比）	准确率下降 > 5% → 触发重训练

7.3 实操：监控代码

"""
模型监控：记录预测分布和推理延迟，超阈值时告警
"""
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import numpy as np
import joblib

# ---------- Prometheus 指标定义 ----------
# Prometheus 是最流行的开源监控系统，指标可以被 Grafana 可视化

# 计数器：预测总数和故障预测数
predictions_total = Counter('model_predictions_total', 'Total predictions', ['result'])

# 直方图：推理延迟分布
inference_latency = Histogram('model_inference_seconds', 'Inference latency')

# 仪表盘：故障预测比例
fault_rate = Gauge('model_fault_rate', 'Current fault prediction rate')

# 仪表盘：特征漂移检测
feature_mean_gauge = Gauge('model_feature_mean', 'Feature mean value', ['feature'])

# ---------- 启动监控端口 ----------
start_http_server(9090)
print("✅ 监控服务已启动: http://localhost:9090/metrics")

# ---------- 加载模型和训练集统计量 ----------
model = joblib.load('models/fault_xgb.joblib')

# 训练集的特征均值和标准差（部署时需要保存这些统计量）
train_means = np.array([300.0, 1537, 39.9, 107.6, 10.3, 61383, 43.5, 1.3])
train_stds = np.array([2.0, 161, 9.9, 63.6, 1.8, 19530, 15.2, 1.1])

# ---------- 模拟在线监控 ----------
print("\n开始模拟在线监控（每 5 条报告一次）...")

recent_preds = []  # 滑动窗口

for i in range(100):
    # 模拟输入数据（前半段正常，后半段发生漂移）
    if i < 60:
        sample = np.random.normal(train_means, train_stds * 0.5, size=(1, 8))
    else:
        # 模拟温度漂移（夏天来了，整体温度升高）
        drifted_means = train_means.copy()
        drifted_means[0] += 5   # 环境温度升高 5K
        drifted_means[4] += 3   # 温差变化
        sample = np.random.normal(drifted_means, train_stds * 0.5, size=(1, 8))
    
    # 推理并记录延迟
    start = time.time()
    prob = model.predict_proba(sample)[0][1]
    pred = int(prob > 0.5)
    elapsed = time.time() - start
    
    # 更新监控指标
    result = 'fault' if pred else 'normal'
    predictions_total.labels(result=result).inc()
    inference_latency.observe(elapsed)
    
    recent_preds.append(pred)
    if len(recent_preds) > 20:
        recent_preds.pop(0)
    
    # 每 5 条检查一次
    if (i + 1) % 5 == 0:
        current_fault_rate = sum(recent_preds) / len(recent_preds)
        fault_rate.set(current_fault_rate)
        
        # 特征漂移检测（简化版：对比特征均值）
        drift_detected = False
        for j, name in enumerate(['Air temp', 'Speed', 'Torque', 'Wear',
                                   'Temp diff', 'Power', 'S/T ratio', 'Wear stage']):
            feature_mean_gauge.labels(feature=name).set(sample[0][j])
            # 如果特征值偏离训练集均值超过 2 个标准差
            if abs(sample[0][j] - train_means[j]) > 2 * train_stds[j]:
                drift_detected = True
        
        status = "🟢" if current_fault_rate < 0.1 else "🟡" if current_fault_rate < 0.3 else "🔴"
        drift_mark = " ⚠️ 漂移" if drift_detected else ""
        print(f"  [{i+1:>3d}/100] {status} 故障率={current_fault_rate:.1%}  "
              f"延迟={elapsed*1000:.1f}ms{drift_mark}")

📌 生产建议：Prometheus 只负责采集指标，可视化用 Grafana（开源仪表盘工具）。Grafana 连接 Prometheus 数据源后，可以拖拽创建折线图、仪表盘，并设置告警规则（如故障率 > 20% 时发邮件通知）。这套 Prometheus + Grafana 组合是工业界最主流的监控方案。

7.4 自动重训练策略

当检测到模型漂移时，需要用新数据重新训练模型。自动重训练的触发条件和流程：

触发条件	具体规则	说明
时间触发	每月 1 号自动重训练	最简单，不管模型好不好都重训
性能触发	准确率连续 3 天下降 > 5%	需要延迟反馈（维修记录）
数据触发	特征均值偏移 > 2σ	不需要真实标签，最快发现
手动触发	运维人员手动点击「重训练」	换工艺、换材料后主动触发

"""
自动重训练触发器（简化版，生产环境需要配合调度系统）
"""
import joblib
from xgboost import XGBClassifier
from datetime import datetime
import os

def check_and_retrain(features_recent, labels_recent, model_path='models/fault_xgb.joblib'):
    """
    检查模型是否需要重训练，如果需要则自动执行。
    
    实际生产中，features_recent 来自最近 N 天的预测记录，
    labels_recent 来自维修工单反馈（延迟获得的真实标签）。
    """
    # 加载当前模型
    model = joblib.load(model_path)
    
    # 评估当前模型在新数据上的表现
    accuracy = model.score(features_recent, labels_recent)
    
    threshold = 0.90  # 准确率低于 90% 触发重训练
    
    if accuracy < threshold:
        print(f"⚠️ 模型准确率 {accuracy:.4f} < {threshold}，触发重训练...")
        
        # 用新数据微调模型
        model.fit(features_recent, labels_recent)
        
        # 保存新版本（带时间戳）
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        new_path = f'models/fault_xgb_{timestamp}.joblib'
        joblib.dump(model, new_path)
        
        # 同时更新 latest 软链接（API 服务读这个路径）
        joblib.dump(model, model_path)
        
        print(f"✅ 重训练完成 | 新准确率: {model.score(features_recent, labels_recent):.4f}")
        print(f"   新模型保存至: {new_path}")
        return True
    else:
        print(f"✅ 模型状态良好 | 准确率: {accuracy:.4f}，无需重训练")
        return False

# 模拟：用一些数据测试
print("检查模型状态...")
# check_and_retrain(X_test, y_test)  # 实际部署时替换为最近的数据

---

八、效果验证与迭代

8.1 上线不是终点：为什么需要 A/B 测试？

模型部署到产线后，最大的风险不是技术bug，而是**「这个模型真的比人工经验好吗？」**

直接全面切换到 AI 决策是危险的——万一模型有隐藏的偏差，可能导致批量误判。正确的做法是逐步验证：

阶段	方法	持续时间	风险
阶段一	影子模式	1-2 周	零风险
阶段二	A/B 测试	2-4 周	可控风险
阶段三	灰度发布	1-2 周	低风险
阶段四	全量切换	持续	正常运行

8.2 影子模式：零风险验证

影子模式的思路很简单：让 AI 模型和人工同时做判断，但只有人工的判断被执行。AI 的预测只记录、不执行，事后对比两者的准确率。

"""
影子模式：同时记录 AI 预测和人工判断，事后对比
"""
import pandas as pd
import time

# 影子日志：记录每次 AI 预测和人工判断
shadow_log = []

def shadow_predict(features, human_decision):
    """
    记录 AI 预测结果和人工判断，用于事后对比。
    
    参数:
        features: 传感器特征
        human_decision: 人工判断结果 (0=正常, 1=故障)
    """
    # AI 预测
    prob = model.predict_proba(features)[0][1]
    ai_decision = int(prob > 0.5)
    
    record = {
        'timestamp': time.time(),
        'ai_prediction': ai_decision,
        'ai_probability': round(prob, 4),
        'human_decision': human_decision,
        'agree': ai_decision == human_decision,
    }
    shadow_log.append(record)
    return record

# 模拟 100 次人机对比
import numpy as np

print("影子模式运行中...")
print(f"{'#':>4} | {'AI预测':>8} | {'人工判断':>8} | {'一致':>4} | {'AI概率':>8}")
print("-" * 55)

agree_count = 0
for i in range(100):
    # 模拟传感器数据
    sample = X_test[i:i+1]
    
    # 模拟人工判断（假设人工准确率约 80%）
    true_label = y_test[i]
    human_correct = np.random.random() < 0.80
    human_decision = true_label if human_correct else 1 - true_label
    
    record = shadow_predict(sample, human_decision)
    agree_count += record['agree']
    
    if (i + 1) % 20 == 0:
        ai_sym = "故障" if record['ai_prediction'] else "正常"
        hu_sym = "故障" if record['human_decision'] else "正常"
        mark = "✅" if record['agree'] else "❌"
        print(f"{i+1:>4} | {ai_sym:>8} | {hu_sym:>8} | {mark:>4} | {record['ai_probability']:>8.4f}")

print(f"\n影子模式报告：人机一致率 = {agree_count}/100 = {agree_count}%")
print(f"{'=' * 40}")
print("分析：")
if agree_count >= 85:
    print("  ✅ 人机一致率 > 85%，AI 判断和人工经验高度吻合，可以进入 A/B 测试")
elif agree_count >= 70:
    print("  🟡 人机一致率 70-85%，需要分析不一致的案例，排查 AI 或人工的偏差")
else:
    print("  🔴 人机一致率 < 70%，不建议上线，需要排查模型或人工标准的准确性")

8.3 A/B 测试：用数据说话

影子模式验证 AI「靠谱」后，进入 A/B 测试——把产线分成两组，一组用 AI 辅助决策，一组继续人工决策，运行一段时间后对比两组的实际效果。

对比项	A 组（AI 辅助）	B 组（人工）	怎么看
故障漏报率	AI 预测为正常但实际故障的比例	人工判断为正常但实际故障的比例	越低越好
误报率	AI 预测为故障但实际正常的比例	人工判断为故障但实际正常的比例	越低越好（但不能太低）
平均响应时间	从告警到安排维修的时间	从发现问题到安排维修的时间	越短越好
非计划停机次数	统计周期内的停机次数	统计周期内的停机次数	越少越好

📌 关键原则：A/B 测试的分组要随机，且两组的设备型号、工况、操作人员水平要尽量一致。否则对比没有意义——就像比较两个班的考试成绩，如果一个班全是尖子生，另一个全是后进生，成绩差异不能说明老师水平。

---

九、综合实战：完整的故障预警系统部署方案

前面 8 节把每个模块单独讲清楚了。现在把它们组装成一个完整的系统架构：

9.1 系统架构图

┌─────────────────────────────────────────────────────────┐
│                    设备层（传感器 / PLC）                    │
│   温度传感器 → 振动传感器 → 转速传感器 → 磨损计数器         │
└───────────────────────┬─────────────────────────────────┘
                        │ MQTT 发布
                        ▼
┌─────────────────────────────────────────────────────────┐
│                  数据层（MQTT Broker）                     │
│   Topic: factory/line1/cnc01/sensors                    │
│   Topic: factory/line1/cnc02/sensors                    │
│   Topic: factory/line2/cnc01/sensors                    │
└───────────────────────┬─────────────────────────────────┘
                        │ MQTT 订阅
                        ▼
┌─────────────────────────────────────────────────────────┐
│                推理层（FastAPI + ONNX）                    │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐             │
│   │ 特征计算  │→│ 模型推理  │→│ 告警判断  │             │
│   └──────────┘  └──────────┘  └──────────┘             │
└────────────┬──────────────────────┬────────────────────┘
             │                      │
             ▼                      ▼
┌────────────────────┐  ┌──────────────────────────────┐
│  监控层 Prometheus  │  │  应用层（MES / SCADA / App）  │
│  - 预测分布         │  │  - 实时仪表盘                │
│  - 推理延迟         │  │  - 告警推送（邮件/短信/App） │
│  - 特征漂移         │  │  - 维修工单自动生成          │
└────────┬───────────┘  └──────────────────────────────┘
         ▼
┌────────────────────┐
│  Grafana 仪表盘     │
│  - 模型健康度       │
│  - 设备故障趋势     │
│  - 告警统计         │
└────────────────────┘

9.2 完整部署脚本

# ========== 一键部署脚本 deploy.sh ==========

# 1. 构建推理服务镜像
echo "📦 构建推理服务镜像..."
docker build -t fault-predictor:v1.0 .

# 2. 启动 MQTT Broker（使用 EMQX）
echo "🔌 启动 MQTT Broker..."
docker run -d --name mqtt-broker \
    -p 1883:1883 -p 8083:8083 \
    emqx/emqx:5.6

# 3. 启动推理服务
echo "🚀 启动推理服务..."
docker run -d --name fault-predictor \
    -p 8000:8000 \
    --restart unless-stopped \
    fault-predictor:v1.0

# 4. 启动 Prometheus（监控）
echo "📊 启动 Prometheus..."
docker run -d --name prometheus \
    -p 9090:9090 \
    -v $(pwd)/prometheus.yml:/etc/prometheus/prometheus.yml \
    prom/prometheus:v2.51.0

# 5. 启动 Grafana（可视化）
echo "📈 启动 Grafana..."
docker run -d --name grafana \
    -p 3000:3000 \
    grafana/grafana:11.0.0

echo "✅ 全部服务启动完成！"
echo ""
echo "服务地址："
echo "  推理 API:   http://localhost:8000/docs"
echo "  MQTT:       localhost:1883"
echo "  Prometheus: http://localhost:9090"
echo "  Grafana:    http://localhost:3000 (默认账号 admin/admin)"

9.3 生产环境检查清单

模型上线前，逐项检查：

检查项	要求	状态
模型精度	测试集 F1 > 0.5（第 5 篇基准）	☐
推理延迟	单次 < 10ms	☐
API 健康检查	/health 接口返回 200	☐
异常输入处理	传入非法值（如温度 = -999）不崩溃	☐
容器自动重启	Docker --restart unless-stopped	☐
日志记录	每次预测记录时间戳 + 输入 + 输出	☐
监控告警	Prometheus 指标正常采集	☐
模型回滚	旧版本镜像保留，可一键回退	☐
TLS 加密	MQTT 和 API 均启用 HTTPS/WSS	☐
压力测试	模拟 100 并发请求，无崩溃	☐

📌 最重要的检查：异常输入处理。传感器故障时可能发送 NaN、Inf、负数温度等非法值。推理服务必须对这些输入做防御性处理（返回错误码而不是崩溃），否则一个坏传感器就能让整个服务挂掉。

---

总结与回顾

要点	总结
模型序列化	Joblib（最快上手）、ONNX（生产推荐，跨语言推理更快）
API 服务化	FastAPI 封装推理接口，自动生成 Swagger 文档
容器化	Docker 打包镜像，一次构建到处运行
数据管道	MQTT 轻量级实时传输，Kafka 适合大规模
监控	Prometheus 采集指标 + Grafana 可视化 + 漂移检测
上线策略	影子模式 → A/B 测试 → 灰度发布 → 全量切换
自动重训练	性能下降 > 5% 或特征偏移 > 2σ 时触发

核心认知：

1. 模型只是系统的一个组件——训练好的模型需要 API 服务化、容器化部署、实时监控，才能变成真正可用的产品
2. 部署不是终点，是新的起点——数据分布会变，模型会漂移，必须有自动重训练和监控机制
3. 逐步验证比一步到位更安全——影子模式 → A/B 测试 → 灰度发布，每一步都在降低风险

---

下篇预告

第 10 篇（高阶拓展）：LLM 辅助数据分析 —— 前 9 篇覆盖了从数据处理到模型部署的完整链路。接下来进入高阶拓展：用大语言模型（LLM）自动化 EDA 数据探索、智能报告生成、自然语言查询数据——让不会写代码的运维人员也能分析设备数据。

---

本文为「从零到落地：机器学习分析数据实战系列」第 9 篇，完整系列持续更新中。