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里我们追求的是迭代速度改一行代码ShiftEnter看结果换一个特征df[new_feat] ...再跑一遍。这种模式依赖三个隐形假设数据是静态的、计算资源是无限的、执行环境是完全受控的。而生产环境恰恰击碎这三者。数据每秒都在流入GPU显存会被其他任务抢占线上服务要和支付系统、风控引擎共享同一套K8s集群。所以Part 4的设计起点不是“怎么把Notebook打包”而是主动切割开研究与生产的边界。我们采用“三段式隔离架构”Stage 1Notebook沙盒——仅用于探索性分析、特征工程验证、模型原型训练。所有输出数据快照、模型权重、特征字典必须打上不可变哈希标签如sha256(data_v20240517.csv)禁止任何硬编码路径或环境变量。Stage 2Pipeline编译器——将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单元格定义了全局变量SCALERB单元格用了它但C单元格又del SCALER。服务化后多线程并发请求可能触发NameError而你在本地永远复现不了——因为Jupyter是单线程顺序执行的。资源泄漏隐蔽plt.figure()创建的图形对象、cv2.VideoCapture()打开的摄像头句柄在Notebook里按Kernel → Restart就清掉了。但在常驻服务里这些句柄会越积越多直到OOM Kill。我们实测过一个没清理matplotlib后端的模型服务运行72小时后内存占用从200MB涨到3.2GB。依赖地狱升级Notebook里!pip install xgboost1.7.5看似简单但生产环境要求所有Python包通过requirements.txt锁定版本并经过安全扫描。自动转换工具往往忽略!pip命令导致线上环境装的是xgboost2.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文档用PrometheusGrafana做监控比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-learn1.2.2训练的模型在scikit-learn1.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/TensorFlowPyTorch用torch.jit.script()转TorchScriptTensorFlow用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的离线数据集上计算并固化不能在线上实时计算。例如标准化用的mean24.7, std8.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(..., exampleU123456) transaction_amount: float Field(..., ge0.01, le1000000.0) # 自动校验小于0.01或大于1e6直接422错误不进业务逻辑 class InferenceResponse(BaseModel): risk_score: float Field(..., ge0.0, le1.0) explanation: dict Field(default_factorydict) app.post(/v1/predict, response_modelInferenceResponse) 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(ge0.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 --frombuilder /wheels /wheels COPY --frombuilder /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 --interval30s --timeout3s --start-period5s --retries3 \ 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 Server85%持续5分钟资源不足需扩容服务健康HTTP 5xx比率、P99延迟、QPSFastAPI的prometheus-fastapi-instrumentator5xx1% or P99500ms服务异常影响用户体验模型性能预测耗时分布、特征缺失率、输入数据量突增自研ModelMetricsMiddleware耗时P99翻倍 or 缺失率5%模型或数据管道问题业务效果风险分分布偏移KS检验、bad rate vs baseline每日离线计算实时采样KS0.2 or bad_rate_delta10%模型失效需紧急干预关键创新点在模型性能维度。我们不只看“服务是否活着”更要看“模型是否在正确工作”。ModelMetricsMiddleware会在每次请求后记录input_size_bytes: 请求体大小检测异常大请求feature_null_ratio: 各字段空值率如id_number为空达80%说明上游ETL崩了prediction_latency_ms: 从收到请求到返回的毫秒数区分CPU/GPU瓶颈这些指标全推到PrometheusGrafana看板上运营人员能一眼看出“今天10:23分user_id字段空值率从0.1%飙升到73%同时risk_score平均值从0.32降到0.08——肯定是用户注册接口挂了不是模型问题。”实操心得不要用logging.info()打监控日志它写磁盘慢高并发下IO阻塞。我们用aiologger异步写stdoutK8s的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的?wait60s。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.25秒生效。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 modelregistry.company.com/model:v${CI_COMMIT_TAG} kubectl scale deployment/model-v2 --replicas2 # 假设总replicas20 sleep 300 # 3. 检查新版本指标P99延迟500ms 5xx0.1% if ! check_metrics model-v2 latency_p99500 error_rate0.001; then echo New version failed metrics check! exit 1 fi # 4. 全量切流 kubectl scale deployment/model-v2 --replicas20这个流程的核心是数据驱动决策不靠人眼盯Dashboard而用check_metrics脚本调Prometheus API查指标。如果新版本P99延迟超标流水线自动失败运维收到企业微信告警“model-v2发布中断原因latency_p99623ms 500ms阈值”。此时回滚只需kubectl rollout undo deployment/model-v230秒内流量切回v2.2。注意蓝绿发布要求服务无状态。所有状态如用户session必须外置到Redis。我们曾有个服务把临时特征缓存到内存字典里蓝绿切换时新Pod没缓存导致首请求延迟飙高被误判为性能问题。后来强制要求任何状态要么是只读的模型权重要么是外置的Redis/DB绝不允许进程内可变状态。4. 实操过程从本地Notebook到K8s集群的完整流水线4.1 Stage 1Notebook沙盒的收尾动作假设你已完成一个信用卡违约预测Notebook最后三步必须做固化数据快照# 在Notebook末尾执行 from datetime import datetime import hashlib import pandas as pd # 保存训练数据快照采样10万行避免太大 train_sample train_df.sample(n100000, random_state42) timestamp datetime.now().strftime(%Y%m%d_%H%M%S) filename fdata_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 2Pipeline编译器的代码实现创建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加载ONNXimport 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要求输入是dictkey为模型输入名 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.pyfrom 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(titleCredit Risk API, version2.3) app.post(/v1/predict, response_modelInferenceResponse) 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_scorerisk_score, inference_time_msround(latency_ms, 2) ) except Exception as e: # 记录详细错误但不暴露给用户 print(fPrediction error: {e}) raise HTTPException(status_code500, detailInternal 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(fLatency: {latency:.3f}s) # 真实代码用client return response4.4 构建与部署CI/CD流水线实录GitLab CI.gitlab-ci.ymlstages: - test - build - deploy test: stage: test image: python:3.9 script: - pip install pytest pytest-cov - pytest tests/ --covpipeline --cov-reporthtml 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 modelregistry.company.com/model:${CI_COMMIT_TAG} - kubectl rollout status deployment/model-v2 --timeout120s部署后验证# 1. 检查Pod状态 kubectl get pods -l appmodel-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距离。如果KS0.15说明数据分布已漂移需触发数据重采样。压力测试用locust模拟1000 QPS观察P99