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，最后三步必须做：
固化数据快照：
# 在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")导出预处理器：
# 不要用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)生成inference_contract.yaml：
手动编写（或用脚本生成），确保和后续Pipeline代码100%一致。这是契约，不是文档。
4.2 Stage 2：Pipeline编译器的代码实现
创建pipeline/目录，结构如下：
pipeline/ ├── __init__.py ├── preprocessor.py # 实现Preprocessor接口 ├── model.py # ONNXModel或TorchScriptModel └── main.py # 组装Pipelinepreprocessor.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固化，此处只存元数据 passmodel.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 response4.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 上线后第一周：必须做的五件事
- 开启全量采样日志：在Grafana里设置告警：“过去1小时，
input_size_bytesP99 > 1MB”，这能发现上游是否在传冗余字段。 - 运行基线对比：用相同1000条样本，对比新旧模型输出，计算KS距离。如果
KS>0.15，说明数据分布已漂移，需触发数据重采样。 - 压力测试：用
locust模拟1000 QPS，观察P99
)
