1. 项目概述：当模型走出Jupyter，真正开始呼吸真实世界的空气

“From Notebook to Production: Running ML in the Real World (Part 4)”——这个标题本身就像一句暗号，专为那些在Jupyter里调通了模型、画出了漂亮ROC曲线、却在把代码扔进生产环境时突然窒息的工程师准备的。我带过十几支AI落地团队，几乎每支队伍都卡在Part 3和Part 4之间：Part 3是“模型能跑”，Part 4是“模型敢用”。这里的“敢用”，不是指技术上能启动服务，而是指它能在凌晨三点服务器负载飙到98%时稳住推理延迟，在用户上传一张模糊证件照时给出可解释的置信度，在数据分布悄悄漂移三个月后自动触发告警，而不是默默输出一堆错误结果还假装一切正常。这期讲的，就是那个被无数教程跳过的临门一脚： 如何让一个在本地笔记本里闪闪发光的ML模型，变成公司API网关背后那个沉默、可靠、可审计、可回滚、出问题能5分钟定位根因的生产级服务 。它不讲模型结构优化，不讲超参搜索，只聚焦一件事： 从 model.predict() 到 curl -X POST https://api.company.com/v1/credit-score 之间，那条布满坑洞、但必须走通的路 。适合刚完成第一个端到端项目的数据科学家、想接手模型部署的后端工程师、以及被业务方天天追问“模型什么时候上线”的AI平台负责人。你不需要精通Kubernetes，但得知道Docker镜像不是魔法；你不必手写gRPC协议，但得明白为什么不能直接把 pickle 文件挂到Nginx后面。

2. 整体设计思路：为什么我们坚决不用“一键部署”工具

2.1 核心矛盾：研究范式与工程范式的天然撕裂

在Jupyter里，我们追求的是 迭代速度 ：改一行代码， Shift+Enter ，看结果；换一个特征， df['new_feat'] = ... ，再跑一遍。这种模式依赖三个隐形假设：数据是静态的、计算资源是无限的、执行环境是完全受控的。而生产环境恰恰击碎这三者。数据每秒都在流入，GPU显存会被其他任务抢占，线上服务要和支付系统、风控引擎共享同一套K8s集群。所以，Part 4的设计起点不是“怎么把Notebook打包”，而是 主动切割开研究与生产的边界 。我们采用“三段式隔离架构”：

  • Stage 1：Notebook沙盒 ——仅用于探索性分析、特征工程验证、模型原型训练。所有输出（数据快照、模型权重、特征字典）必须打上不可变哈希标签（如 sha256(data_v20240517.csv) ），禁止任何硬编码路径或环境变量。
  • Stage 2：Pipeline编译器 ——将Notebook中验证过的逻辑，用 scikit-learn 的 Pipeline 或 TensorFlow Transform 重写为可序列化的、无副作用的函数链。关键点： 所有数据预处理必须固化为独立模块，与模型权重解耦 。比如，文本清洗不能写在模型 forward() 里，而要封装成 TextNormalizer.transform() ，这样当NLP团队更新停用词表时，只需替换预处理器，无需重训整个BERT。
  • Stage 3：服务运行时 ——基于容器的轻量级服务框架，核心只做三件事：接收请求、调用Stage 2编译好的Pipeline、返回结构化响应。它不碰数据源，不连数据库，不写日志到磁盘（日志全走stdout/stderr由K8s收集），彻底剥离业务逻辑。

这个设计不是为了炫技，而是解决一个血泪教训：去年帮一家银行部署反欺诈模型，他们最初用 joblib.dump(model, 'prod_model.pkl') ，结果某次特征工程脚本更新后忘了同步 pkl 文件里的预处理代码，导致线上服务对新用户返回全零预测。排查花了6小时——因为 pkl 文件里混着模型和预处理逻辑，根本分不清是模型坏了还是数据喂错了。三段式之后，预处理器版本号和模型版本号独立管理，发布时自动生成依赖矩阵图，故障定位时间从小时级降到分钟级。

2.2 为什么拒绝“Notebook即服务”类方案

市面上不少工具鼓吹“把.ipynb文件拖进平台，自动转成API”，这类方案在Part 4里是危险的。原因有三：

  • 状态污染不可控 ：Notebook单元格执行顺序依赖人工记忆。A单元格定义了全局变量 SCALER ，B单元格用了它，但C单元格又 del SCALER 。服务化后，多线程并发请求可能触发 NameError ，而你在本地永远复现不了——因为Jupyter是单线程顺序执行的。
  • 资源泄漏隐蔽 ： plt.figure() 创建的图形对象、 cv2.VideoCapture() 打开的摄像头句柄，在Notebook里按 Kernel → Restart 就清掉了。但在常驻服务里，这些句柄会越积越多，直到OOM Kill。我们实测过，一个没清理 matplotlib 后端的模型服务，运行72小时后内存占用从200MB涨到3.2GB。
  • 依赖地狱升级 ：Notebook里 !pip install xgboost==1.7.5 看似简单，但生产环境要求所有Python包通过 requirements.txt 锁定版本，并经过安全扫描。自动转换工具往往忽略 !pip 命令，导致线上环境装的是 xgboost==2.0.0 ，而新版本里 predict_proba() 返回格式变了，下游业务直接报错。

所以我们的选型原则很粗暴： 任何需要解析.ipynb文件AST树的工具，一律不进生产链路 。取而代之的是“契约先行”——在Stage 1结束时，强制产出一份 inference_contract.yaml ，明确声明：

input_schema:
  - name: "user_id"
    type: "string"
    required: true
  - name: "transaction_amount"
    type: "float"
    min: 0.01
    max: 1000000.0
output_schema:
  - name: "risk_score"
    type: "float"
    min: 0.0
    max: 1.0
  - name: "explanation"
    type: "object"  # SHAP值结构

这个YAML文件就是Stage 2和Stage 3之间的唯一接口。模型开发者按它写测试，服务开发者按它写路由，QA按它写压力脚本。没有歧义，没有猜测，没有“我以为你传了字符串”。

2.3 架构选型背后的成本权衡

很多人问：“为什么不用SageMaker或Vertex AI？它们不是标榜‘一键部署’吗？”——答案是： 它们解决的是‘部署’问题，而Part 4要解决的是‘可信交付’问题 。大厂托管服务确实省去了K8s运维，但带来了新枷锁：

  • 调试黑盒化 ：SageMaker的 invoke_endpoint() 返回500错误，你只能看到CloudWatch里一行 ModelError ，无法进入容器 strace 看是哪个Python线程卡死，也无法 tcpdump 抓包确认是网络超时还是模型hang住。
  • 灰度发布能力弱 ：想给5%的流量切到新模型，同时保留旧模型供对比，SageMaker需要配复杂的A/B测试Endpoint，而我们用Istio的 VirtualService 规则，三行YAML搞定： 
    http:
- route:
  - destination:
      host: model-v1
    weight: 95
  - destination:
      host: model-v2
    weight: 5
  • 合规审计缺失 ：金融客户要求所有模型输入/输出留存6个月以备监管检查。SageMaker的 DataCaptureConfig 只存S3，但你需要自己写Lambda去解析Parquet、加水印、做加密。而我们的服务在Stage 3里内置了 AuditLogger 中间件，收到请求立刻生成 audit_id = sha256(f"{timestamp}_{request_id}_{input_hash}") ，所有日志带此ID，审计时 grep audit_id /var/log/app.log 就能拉出完整链路。

所以最终技术栈是“务实混合体”：用Docker做环境隔离（避免 conda env export 的不确定性），用FastAPI做服务框架（比Flask更严格的Pydantic校验，自动生OpenAPI文档），用Prometheus+Grafana做监控（比CloudWatch便宜83%，且指标可自定义）。这不是技术洁癖，而是算过一笔账：一个中型项目，用托管服务年成本约$42,000，自建栈$18,000，省下的钱够雇半个专职MLOps工程师，专门盯模型漂移和性能退化。

3. 核心细节解析：从模型文件到可观察服务的七道工序

3.1 模型序列化：Pickle是毒药，ONNX是解药？

模型保存方式，是Part 4的第一道生死线。新手常犯的错是 joblib.dump(model, 'model.pkl') ，然后在服务里 joblib.load('model.pkl') 。这看似简单，实则埋雷：

  • Python版本锁死 ： pkl 文件包含Python对象的二进制表示， scikit-learn==1.2.2 训练的模型，在 scikit-learn==1.3.0 里 load() 可能失败。我们遇到过客户升级Python 3.9→3.10后，所有 pkl 模型加载报 ModuleNotFoundError: No module named 'sklearn.ensemble._forest' 。
  • 反序列化RCE风险 ： pickle 可执行任意代码。如果攻击者篡改了 model.pkl 文件，服务 load() 时就会执行恶意payload。虽然生产环境不该让外部控制模型文件，但防御纵深原则要求： 序列化格式必须是纯数据，不含代码 。

所以我们的标准流程是“双轨制”：

  • 算法层模型 （XGBoost/LightGBM/Sklearn）：导出为ONNX格式。用 onnxmltools.convert_sklearn() 或 hummingbird.ml.convert() ，转换后用 onnxruntime.InferenceSession 加载。ONNX是开放标准，跨语言、跨平台、无Python依赖，且 onnxruntime 支持CPU/GPU/ARM多种后端。实测显示，ONNX版XGBoost比原生 xgboost.Booster.predict() 快12%，因为ORT做了图优化。
  • 深度学习模型 （PyTorch/TensorFlow）：PyTorch用 torch.jit.script() 转TorchScript，TensorFlow用 tf.saved_model.save() 。 绝不使用 torch.save() 的 .pt 格式 ——它和 pkl 一样有版本兼容问题。TorchScript的好处是：模型和推理逻辑一起编译， model.forward() 被固化为图节点，避免Python解释器开销。我们压测过，TorchScript版ResNet50在T4 GPU上P99延迟比Eager模式低47ms。

提示：ONNX转换不是无损的。某些Sklearn模型（如 IsolationForest ）的 decision_function() 在ONNX里不支持。此时我们采用“降级策略”：用 skl2onnx 转换主干，对不支持的方法，单独写一个 FallbackPredictor 类，在ONNX加载失败时自动切换回原生Sklearn，但会记录 fallback_count 指标告警。这是工程妥协，不是技术倒退。

3.2 特征预处理器：为什么它比模型还重要

在Part 4里，我常说一句话：“ 模型是大脑，预处理器是感官和运动神经 ”。一个准确率99%的模型，如果预处理器把用户年龄字段当成字符串处理（ '35' →ASCII码求和），结果就是灾难。所以预处理器必须满足：

  • 可重现性 ：所有统计量（均值、标准差、词频、分位数）必须在Stage 1的离线数据集上计算并固化，不能在线上实时计算。例如，标准化用的 mean=24.7, std=8.3 要写死在 preprocessor.json 里，而不是每次启动服务时 df['age'].mean() 。
  • 原子性 ：每个预处理步骤必须是纯函数，无状态、无副作用。 TextNormalizer 不能维护内部词典缓存，而应把词典作为参数传入 transform(text, vocab_dict) 。
  • 可测试性 ：提供 preprocessor.test() 方法，输入一组已知样本，断言输出是否符合 inference_contract.yaml 定义的schema。

我们强制要求所有预处理器实现统一接口：

class Preprocessor(ABC):
    @abstractmethod
    def fit(self, X: pd.DataFrame) -> 'Preprocessor': ...
    @abstractmethod
    def transform(self, X: pd.DataFrame) -> pd.DataFrame: ...
    @abstractmethod
    def save(self, path: str) -> None: ...  # 序列化为JSON/Joblib
    @abstractmethod
    def load(self, path: str) -> 'Preprocessor': ...  # 反序列化

这样，Stage 2的Pipeline编译器就能用 Pipeline([("scaler", StandardScaler()), ("model", ONNXModel())]) 统一管理，无论里面是Sklearn还是ONNX模型。

注意：不要用 sklearn.preprocessing.StandardScaler().fit_transform() 一步到位！必须拆成 fit() 和 transform() 两步。因为 fit_transform() 在训练时计算统计量并立即应用，但生产时你只有 transform() 阶段，没有原始训练数据。我们见过太多团队在 fit_transform() 后直接 joblib.dump() ，结果线上 transform() 时因未 fit() 报 NotFittedError 。

3.3 服务框架：FastAPI不是“更快的Flask”，而是契约执行器

选FastAPI而非Flask，核心原因是它的 类型驱动开发（TDD）能力 。在Part 4里，我们把 inference_contract.yaml 直接映射为Pydantic模型：

from pydantic import BaseModel, Field
from typing import List, Optional

class InferenceRequest(BaseModel):
    user_id: str = Field(..., example="U123456")
    transaction_amount: float = Field(..., ge=0.01, le=1000000.0)
    # 自动校验：小于0.01或大于1e6直接422错误，不进业务逻辑

class InferenceResponse(BaseModel):
    risk_score: float = Field(..., ge=0.0, le=1.0)
    explanation: dict = Field(default_factory=dict)

@app.post("/v1/predict", response_model=InferenceResponse)
def predict(request: InferenceRequest):
    # 这里request已经是校验后的干净数据
    result = pipeline.run(request.dict())
    return result

这个设计带来三个硬性收益：

  • 零成本API文档 ： /docs 自动生成Swagger UI，业务方不用读代码就知道怎么调用，减少50%的联调会议。
  • 前端强类型保障 ：用 pydantic 生成TypeScript接口，前端 fetch() 返回的数据结构和后端定义100%一致，避免 response.risk_score 写成 response.score 的低级错误。
  • 异常防御前置 ： Field(ge=0.01) 在请求解析阶段就拦截非法值，不会让脏数据流到模型推理层。我们线上服务92%的4xx错误来自这里，而不是模型崩溃——这正是我们想要的： 把错误扼杀在入口，而不是让它污染核心逻辑 。

3.4 容器化：Dockerfile里的每一行都是生产承诺

一个合格的生产镜像，Dockerfile必须回答三个问题： 它装了什么？为什么装？谁来验证？ 我们的模板长这样：

# 第一阶段：构建环境（多阶段构建，减小镜像体积）
FROM python:3.9-slim AS builder
COPY requirements.txt .
RUN pip wheel --no-cache-dir --no-deps --wheel-dir /wheels -r requirements.txt

# 第二阶段：运行环境（最小化基础镜像）
FROM python:3.9-slim-buster
# 复制构建好的wheel，不装编译工具链
COPY --from=builder /wheels /wheels
COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages
# 强制安装，跳过依赖检查（确保wheel版本精确匹配）
RUN pip install --no-cache-dir --force-reinstall --no-deps /wheels/*.whl

# 复制应用代码和模型资产
COPY app/ /app/
COPY models/ /app/models/

# 关键：设置非root用户，禁用交互式shell
RUN addgroup -g 1001 -f mlgroup && adduser -S mluser -u 1001
USER mluser
WORKDIR /app

# 健康检查：确保服务能响应HTTP GET /health
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4"]

这个Dockerfile的每一行都有深意：

  • python:3.9-slim-buster 而非 latest ：避免某天 latest 升级到3.10导致兼容问题。我们用 slim-buster 是因为它比 alpine 更兼容glibc，很多科学计算包（如 numpy ）在Alpine上需重新编译。
  • 多阶段构建： builder 阶段装 gcc 等编译工具， runner 阶段只复制编译好的wheel，最终镜像从1.2GB降到320MB，拉取速度快4倍，攻击面小得多。
  • adduser -S mluser ：禁止root运行。曾有个团队用root跑服务，攻击者利用 pandas 漏洞拿到shell后，直接 rm -rf / 删了整个宿主机。
  • HEALTHCHECK ：不是摆设。K8s的 livenessProbe 依赖它，当模型OOM时， curl 超时触发重启，比等 kubectl get pods 发现 CrashLoopBackOff 快10分钟。

3.5 监控与可观测性：别等用户投诉才看日志

生产服务的监控，不是“加个Prometheus exporter”就完事。Part 4要求 四维监控体系 ：

维度 指标示例 采集方式 告警阈值 业务意义
基础设施 CPU使用率、内存RSS、GPU显存 K8s Metrics Server >85%持续5分钟 资源不足，需扩容
服务健康 HTTP 5xx比率、P99延迟、QPS FastAPI的 prometheus-fastapi-instrumentator 5xx>1% or P99>500ms 服务异常，影响用户体验
模型性能 预测耗时分布、特征缺失率、输入数据量突增 自研 ModelMetricsMiddleware 耗时P99翻倍 or 缺失率>5% 模型或数据管道问题
业务效果 风险分分布偏移（KS检验）、bad rate vs baseline 每日离线计算+实时采样 KS>0.2 or bad_rate_delta>+10% 模型失效，需紧急干预

关键创新点在 模型性能维度 。我们不只看“服务是否活着”，更要看“模型是否在正确工作”。 ModelMetricsMiddleware 会在每次请求后记录：

  • input_size_bytes : 请求体大小（检测异常大请求）
  • feature_null_ratio : 各字段空值率（如 id_number 为空达80%，说明上游ETL崩了）
  • prediction_latency_ms : 从收到请求到返回的毫秒数（区分CPU/GPU瓶颈）

这些指标全推到Prometheus，Grafana看板上，运营人员能一眼看出：“今天10:23分， user_id 字段空值率从0.1%飙升到73%，同时 risk_score 平均值从0.32降到0.08——肯定是用户注册接口挂了，不是模型问题。”

实操心得：不要用 logging.info() 打监控日志！它写磁盘慢，高并发下IO阻塞。我们用 aiologger 异步写stdout，K8s的 fluent-bit 实时采集，延迟<200ms。曾经一个服务因日志同步写磁盘，QPS从1200掉到300，排查三天才发现是 logging.basicConfig() 没配 handlers=[RotatingFileHandler(...)] 。

3.6 配置管理：环境变量不是万能钥匙

新手总爱把所有配置塞进环境变量： MODEL_PATH=/models/v2.onnx , DB_URL=... 。这在Part 4里是反模式，因为：

  • 安全性差 ： DB_URL 含密码， env | grep DB 就能泄露。
  • 可维护性差 ：10个服务共用一个 REDIS_URL ，改一个全崩。
  • 不可审计 ：环境变量修改无记录，出问题无法追溯谁在何时改了什么。

我们的方案是“三层配置”：

  • 硬编码默认值 ：在代码里写死最安全的默认值，如 DEFAULT_TIMEOUT_SECONDS = 30 。这是底线，即使配置中心宕机，服务也能用默认值苟活。
  • 配置中心 （Consul/Etcd）：存敏感配置，如 model_version: v2.3 , canary_weight: 0.05 。服务启动时 GET /v1/kv/config 拉取，变更时长连接通知（Consul的 ?wait=60s ）。
  • K8s ConfigMap ：存非敏感、需频繁更新的配置，如 feature_whitelist: ["age","income","city"] 。用 volumeMount 挂载为文件，服务内 watch 文件mtime变化，热重载。

这样， MODEL_PATH 不再是一个路径，而是一个 配置项 ：服务从Consul读 model.version ，拼出 https://minio.company.com/models/{version}/model.onnx ，再用 requests.get() 下载。好处是：模型更新不用发版，改个配置，服务自动拉新模型；回滚只需把 version 设回 v2.2 ，5秒生效。

3.7 发布与回滚：蓝绿发布不是噱头，是生存必需

“上线”在Part 4里不是 git push ，而是一套自动化流水线。我们用GitLab CI定义 deploy-prod 阶段：

deploy-prod:
  stage: deploy
  image: alpine:latest
  script:
    - apk add curl
    - |
      # 1. 验证新模型SHA256与Stage 1报告一致
      NEW_SHA=$(curl -s https://minio.company.com/models/${CI_COMMIT_TAG}/SHA256)
      EXPECTED_SHA=$(cat reports/model_sha256.txt)
      if [ "$NEW_SHA" != "$EXPECTED_SHA" ]; then
        echo "Model hash mismatch!" && exit 1
      fi
    - |
      # 2. 执行蓝绿切换：先扩新版本到10%，观察5分钟
      kubectl set image deployment/model-v2 model=registry.company.com/model:v${CI_COMMIT_TAG}
      kubectl scale deployment/model-v2 --replicas=2  # 假设总replicas=20
      sleep 300
      # 3. 检查新版本指标：P99延迟<500ms & 5xx<0.1%
      if ! check_metrics "model-v2" "latency_p99<500" "error_rate<0.001"; then
        echo "New version failed metrics check!" && exit 1
      fi
      # 4. 全量切流
      kubectl scale deployment/model-v2 --replicas=20

这个流程的核心是 数据驱动决策 ：不靠人眼盯Dashboard，而用 check_metrics 脚本调Prometheus API查指标。如果新版本P99延迟超标，流水线自动失败，运维收到企业微信告警：“model-v2发布中断，原因：latency_p99=623ms > 500ms阈值”。此时，回滚只需 kubectl rollout undo deployment/model-v2 ，30秒内流量切回v2.2。

注意：蓝绿发布要求服务无状态。所有状态（如用户session）必须外置到Redis。我们曾有个服务把临时特征缓存到内存字典里，蓝绿切换时新Pod没缓存，导致首请求延迟飙高，被误判为性能问题。后来强制要求： 任何状态，要么是只读的（模型权重），要么是外置的（Redis/DB），绝不允许进程内可变状态 。

4. 实操过程：从本地Notebook到K8s集群的完整流水线

4.1 Stage 1：Notebook沙盒的收尾动作

假设你已完成一个信用卡违约预测Notebook，最后三步必须做：

  1. 固化数据快照 ：

    # 在Notebook末尾执行
from datetime import datetime
import hashlib
import pandas as pd

# 保存训练数据快照（采样10万行，避免太大）
train_sample = train_df.sample(n=100000, random_state=42)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"data_snapshot_{timestamp}.parquet"
train_sample.to_parquet(filename)

# 计算SHA256，写入REPORT.md
with open(filename, "rb") as f:
    sha = hashlib.sha256(f.read()).hexdigest()
with open("REPORT.md", "a") as f:
    f.write(f"\n## Data Snapshot\n- File: {filename}\n- SHA256: {sha}\n")

  2. 导出预处理器 ：

    # 不要用pickle！用JSON序列化统计量
import json
preprocessor_stats = {
    "age": {"mean": 42.3, "std": 12.7},
    "income": {"min": 2000.0, "max": 50000.0},
    "city": {"top_k": ["Beijing", "Shanghai", "Guangzhou"]}
}
with open("preprocessor_stats.json", "w") as f:
    json.dump(preprocessor_stats, f)

  3. 生成inference_contract.yaml ：
    手动编写（或用脚本生成），确保和后续Pipeline代码100%一致。这是契约，不是文档。

4.2 Stage 2：Pipeline编译器的代码实现

创建 pipeline/ 目录，结构如下：

pipeline/
├── __init__.py
├── preprocessor.py     # 实现Preprocessor接口
├── model.py            # ONNXModel或TorchScriptModel
└── main.py             # 组装Pipeline

preprocessor.py 关键代码：

import json
import numpy as np
import pandas as pd

class StandardScalerPreprocessor:
    def __init__(self, stats_path: str):
        with open(stats_path) as f:
            self.stats = json.load(f)
    
    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        X_out = X.copy()
        for col, stat in self.stats.items():
            if col in X_out.columns:
                # 处理缺失值：用中位数填充（比均值鲁棒）
                fill_val = stat.get("median", stat["mean"])
                X_out[col] = X_out[col].fillna(fill_val)
                # 标准化
                X_out[col] = (X_out[col] - stat["mean"]) / stat["std"]
        return X_out
    
    def save(self, path: str):
        # 已在Stage 1固化，此处只存元数据
        pass

model.py 加载ONNX：

import onnxruntime as ort
import numpy as np

class ONNXModel:
    def __init__(self, model_path: str):
        # 开启优化：启用内存复用和图优化
        sess_options = ort.SessionOptions()
        sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_EXTENDED
        sess_options.intra_op_num_threads = 1  # 避免线程竞争
        self.session = ort.InferenceSession(model_path, sess_options)
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        # ONNX要求输入是dict，key为模型输入名
        input_name = self.session.get_inputs()[0].name
        result = self.session.run(None, {input_name: X.astype(np.float32)})
        return result[0]  # 返回概率数组

main.py 组装：

from sklearn.pipeline import Pipeline
from pipeline.preprocessor import StandardScalerPreprocessor
from pipeline.model import ONNXModel

# 加载预处理器和模型
preprocessor = StandardScalerPreprocessor("preprocessor_stats.json")
model = ONNXModel("models/v2.3/model.onnx")

# 创建Pipeline（注意：顺序不能错！）
pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("model", model)
])

# 保存Pipeline（只存预处理器stats和模型路径，不存二进制）
import joblib
joblib.dump({
    "preprocessor_stats": "preprocessor_stats.json",
    "model_path": "models/v2.3/model.onnx"
}, "pipeline_v2.3.joblib")

4.3 Stage 3：服务代码的骨架搭建

app/main.py ：

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import Dict, Any
import joblib
import numpy as np
import time

# 全局加载Pipeline（启动时一次，非每次请求）
pipeline = joblib.load("pipeline_v2.3.joblib")

class InferenceRequest(BaseModel):
    user_id: str
    transaction_amount: float
    age: float
    income: float

class InferenceResponse(BaseModel):
    risk_score: float
    inference_time_ms: float

app = FastAPI(title="Credit Risk API", version="2.3")

@app.post("/v1/predict", response_model=InferenceResponse)
async def predict(request: InferenceRequest):
    start_time = time.time()
    try:
        # 转为DataFrame（按contract定义的字段）
        input_df = pd.DataFrame([request.dict()])
        
        # Pipeline执行
        pred_proba = pipeline.predict(input_df)
        risk_score = float(pred_proba[0][1])  # 二分类，取正类概率
        
        latency_ms = (time.time() - start_time) * 1000
        return InferenceResponse(
            risk_score=risk_score,
            inference_time_ms=round(latency_ms, 2)
        )
    except Exception as e:
        # 记录详细错误，但不暴露给用户
        print(f"Prediction error: {e}")
        raise HTTPException(status_code=500, detail="Internal server error")

app/middleware.py 添加监控：

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import time

class ModelMetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        
        # 记录指标到Prometheus
        if request.url.path == "/v1/predict":
            latency = time.time() - start_time
            # 这里调用prometheus_client.Counter().inc()等
            print(f"Latency: {latency:.3f}s")  # 真实代码用client
            
        return response

4.4 构建与部署：CI/CD流水线实录

GitLab CI .gitlab-ci.yml ：

stages:
  - test
  - build
  - deploy

test:
  stage: test
  image: python:3.9
  script:
    - pip install pytest pytest-cov
    - pytest tests/ --cov=pipeline --cov-report=html

build:
  stage: build
  image: docker:20.10.16
  services:
    - docker:20.10.16-dind
  script:
    - docker build -t registry.company.com/model:${CI_COMMIT_TAG} .
    - docker push registry.company.com/model:${CI_COMMIT_TAG}

deploy-prod:
  stage: deploy
  image: bitnami/kubectl:1.25
  before_script:
    - mkdir -p ~/.kube
    - echo "$KUBE_CONFIG" | base64 -d > ~/.kube/config
  script:
    - kubectl set image deployment/model-v2 model=registry.company.com/model:${CI_COMMIT_TAG}
    - kubectl rollout status deployment/model-v2 --timeout=120s

部署后验证：

# 1. 检查Pod状态
kubectl get pods -l app=model-v2

# 2. 调用健康检查
curl http://model-v2.prod.svc.cluster.local:8000/health

# 3. 发送测试请求（用contract定义的样例）
curl -X POST http://model-v2.prod.svc.cluster.local:8000/v1/predict \
  -H "Content-Type: application/json" \
  -d '{"user_id":"U123","transaction_amount":5000.0,"age":35.0,"income":15000.0}'
# 返回: {"risk_score":0.82,"inference_time_ms":12.45}

4.5 上线后第一周：必须做的五件事

  1. 开启全量采样日志 ：在Grafana里设置告警：“过去1小时， input_size_bytes P99 > 1MB”，这能发现上游是否在传冗余字段。
  2. 运行基线对比 ：用相同1000条样本，对比新旧模型输出，计算KS距离。如果 KS>0.15 ，说明数据分布已漂移，需触发数据重采样。
  3. 压力测试 ：用 locust 模拟1000 QPS，观察P99
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

Manim 数学动画完全指南：五分钟从零到成片

Manim 是用 Python 驱动的数学动画引擎，把抽象的公式与曲线转成逐帧可控的动态可视化。适合想制作 Python 数学可视化视频的编程新手、数学学生和科普创作者。## 数学为什么需要"动"起来你试过用一张静态图讲清导数的概念吗？切线、割线、极限位置都能画出来，但"割线逼近切线"发生的那一下，观众只能靠脑补。课本插图大多是"结论截图"：只留下最终状态，过程被压缩成一句"易见"。函

avatar 码道开发者社区
cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar 码道开发者社区
cover

GLM-5.3-Flash深度解析：3200亿参数只卖Opus四十分之一，国产算力首次跑通前沿模型

avatar 码道开发者社区