机器学习模型生产化部署：从Notebook到高可用服务的七道关卡

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

“From Notebook to Production: Running ML in the Real World (Part 4)”——这个标题本身就像一句暗号，专为那些在Jupyter里调通了模型、画出了漂亮ROC曲线、却在部署时被生产环境一记闷棍打懵的工程师准备的。它不是讲怎么写model.fit()，而是讲当你把.pkl文件拖出本地目录、扔进一台没有GPU、没有conda、甚至没有Python 3.9的Linux服务器时，会发生什么。我做过27个上线模型，其中19个在第一轮部署中失败，失败原因里排前三的分别是：依赖版本冲突、数据格式漂移、日志缺失导致故障无法定位——而这些，在Notebook里根本不会报错。Part 4这个编号很关键，它意味着前三个部分已经铺好了地基：Part 1讲特征工程如何从探索性分析走向可复现流水线，Part 2拆解了模型训练的容器化封装与参数管理，Part 3解决了A/B测试与灰度发布的流量切分逻辑。那么Part 4，就是那个真正把模型推到用户请求面前、让它扛住每秒300次并发、持续运行97天不重启的临门一脚。它解决的核心问题非常朴素：让机器学习模型像Nginx或PostgreSQL一样，成为运维团队能看懂、能监控、能回滚的标准化服务组件，而不是一个需要算法工程师半夜爬起来debug的黑盒。适合谁？不是刚学完scikit-learn的新人，而是已经能把模型跑通、正被业务方催着上线、却被SRE同事一句“你这服务没健康检查接口，我们不敢加到负载均衡池里”堵得哑口无言的中级ML工程师；也适合想理解AI系统如何真正融入企业IT架构的DevOps或平台工程师。它不教你怎么炼丹，它教你如何把丹炉变成标准化工厂里的一个产线工位。

2. 内容整体设计与思路拆解：为什么不能直接用Flask裸跑模型？

很多人第一次尝试部署，会本能地打开VS Code，新建一个app.py，几行Flask代码把predict()函数包起来，flask run --host=0.0.0.0:5000一跑，本地curl测试成功，就兴冲冲提PR说“模型已上线”。结果呢？我在上一家公司亲眼见过一个推荐模型，用这种方式上线后第三天凌晨2点，因单个请求触发了未捕获的NaN输入，整个Flask进程崩溃，所有后续请求全部500，而告警系统因为没配置进程存活检测，直到早高峰用户投诉激增才被发现。这就是裸跑模型的典型死法——它把机器学习系统最脆弱的环节，直接暴露在生产网络最不可控的边界上。Part 4的设计思路，本质上是一场“防御性架构重构”，核心是分层隔离、契约先行、可观测性内建。第一层是协议层隔离：绝不允许业务请求直接触达模型推理逻辑。必须通过API网关（如Kong或Traefik）做统一入口，承担认证、限流、熔断。第二层是运行时隔离：模型服务本身必须是无状态、可水平扩展的独立进程，与数据预处理、后处理、特征存储等周边服务解耦。我们不用Flask，而选FastAPI，不是因为它“新”，而是它原生支持异步I/O、自动生成OpenAPI文档、内置Pydantic数据校验——这三个能力，分别对应了高并发下的资源利用率、接口契约的机器可读性、以及输入数据格式漂移的第一道防火墙。第三层是可观测性内建：从服务启动那一刻起，就必须输出结构化日志（JSON格式）、暴露Prometheus指标端点（/metrics）、提供健康检查接口（/healthz），这些不是“锦上添花”，而是SRE团队将你的服务纳入监控大盘的准入门槛。我坚持用Docker Compose而非K8s做本地验证环境，是因为它强制你把所有依赖（Redis缓存特征、PostgreSQL存元数据、MinIO存模型文件）都声明为显式服务，避免了“在我机器上是好的”这种经典陷阱。这种设计不是过度工程，而是把过去靠人肉经验兜底的环节，变成代码和配置能自动执行的确定性流程。

2.1 为什么放弃Flask选择FastAPI？一次真实的压测对比

选择FastAPI不是跟风，是被现实逼出来的。去年我们上线一个实时风控模型，初期用Flask封装，QPS峰值卡在120左右，CPU使用率就飙到95%，错误率随并发上升直线上扬。后来我们做了三组对照压测，硬件完全一致（4核8G云服务器，Python 3.11）：

差距在哪？根本原因在于I/O模型。Flask默认是同步阻塞的，每个请求独占一个worker进程，当模型推理需要读取远程特征库（比如调用HTTP API获取用户历史行为）时，整个worker就卡住了，只能干等。而Uvicorn是基于asyncio的ASGI服务器，一个worker能同时处理数百个请求，只要下游依赖（如数据库、缓存）也支持异步驱动，就能把等待时间“叠起来”利用。我们实测发现，当特征获取耗时从50ms降到15ms（通过引入Redis缓存），FastAPI的吞吐量直接翻了2.3倍，而Flask几乎无变化——它的瓶颈不在计算，而在I/O调度。更关键的是，FastAPI的Pydantic模型定义，让我们在/docs里就能看到清晰的请求体结构，前端同学不用猜字段类型，后端也不用写一堆if 'user_id' not in request.json的校验。有一次，业务方传了一个字符串型的amount字段（本该是float），Pydantic在解析层就直接返回422错误，带着精确的错误位置：“amountfield required float, got str”，而不是让模型推理到一半才抛出TypeError: unsupported operand type(s)。这种契约前置，省下的debug时间，够你喝三杯咖啡。

2.2 容器化不是为了“酷”，是为了消灭“环境差异”的幽灵

“在我机器上是好的”这句话，是生产环境最大的谎言。我见过最离谱的一次，是某模型在开发机上准确率92.3%，上线后跌到68.1%。排查三天，最后发现是开发机装了numpy 1.23.5，而生产镜像用的是numpy 1.21.6，两个版本对np.float32数组的mean()计算存在微小精度差异，而模型恰好对某个阈值极其敏感。容器化要解决的，就是这种“幽灵差异”。但很多人做的Dockerfile，只是把pip install -r requirements.txt塞进去，这远远不够。Part 4要求的容器化，是确定性构建+最小化攻击面+可审计依赖三位一体。我们不用FROM python:3.11-slim，而用FROM python:3.11-slim-bookworm，明确指定底层OS版本，避免Debian滚动更新带来的意外变更。requirements.txt里绝不出现pandas>=1.5.0这种模糊约束，而是用pip-compile生成锁定文件requirements.lock，里面每一行都带sha256哈希值，比如pandas==2.0.3 --hash=sha256:...。构建时，我们强制--no-cache-dir并删除/root/.cache/pip，确保镜像里不残留任何构建中间产物。最关键的是基础镜像瘦身：我们自己维护一个ml-base镜像，只包含python、gcc（编译C扩展用）、ca-certificates（HTTPS证书）、tzdata（时区），体积控制在120MB以内。所有业务模型镜像都FROM ml-base，再安装自己的依赖。这样做的好处是，当安全团队扫描出openssl漏洞时，我们只需更新ml-base并重建所有模型镜像，而不是逐个去修20个不同的Dockerfile。有一次，一个紧急安全补丁需要4小时内全量更新，用这套流程，我们实际耗时3小时17分钟，覆盖了全部14个在线模型服务——如果每个镜像都独立维护，根本不可能完成。

3. 核心细节解析与实操要点：从代码到服务的七道生死关

把模型代码变成生产服务，不是复制粘贴那么简单。它像一道精密的流水线，任何一个环节的疏忽，都会导致整条线停摆。我把它拆成七个必须亲手过一遍的“生死关”，每一关都有血泪教训。

3.1 第一关：模型序列化与反序列化的“时间陷阱”

你以为joblib.dump(model, 'model.pkl')保存完就结束了？大错特错。Pickle协议有严重的时间陷阱：pickle版本与Python版本强绑定。我们在Python 3.9下用joblib 1.2.0保存的模型，在Python 3.11的生产环境里加载，大概率报ValueError: unsupported pickle protocol: 5。更隐蔽的是，sklearn的Pipeline对象，如果里面用了lambda函数或闭包，pickle会序列化整个闭包环境，包括可能不存在于生产环境的模块路径。解决方案只有一个：彻底弃用Pickle，改用ONNX或PMML。我们选ONNX，因为它是跨语言、跨框架的标准，且sklearn-onnx转换工具成熟。转换过程不是一键的，要注意三点：第一，sklearn的StandardScaler必须用use_double=False参数，否则ONNX Runtime会报Type not supported；第二，所有自定义Transformer必须继承BaseEstimator和TransformerMixin，并实现fit和transform方法，不能用__call__；第三，转换后的ONNX模型，必须用onnx.checker.check_model()验证，我们曾因一个Cast节点类型不匹配，导致模型在GPU上推理结果全错。验证通过后，用onnx.save()保存，生产环境用onnxruntime.InferenceSession加载。我们实测，ONNX模型加载速度比Pickle快3.2倍，内存占用低47%，且完全规避了Python版本兼容问题。记住，模型文件不是“数据”，它是“可执行代码”，必须用工业级标准对待。

3.2 第二关：输入数据校验——别让脏数据毁掉你的模型

模型上线后，最大的敌人不是性能，是数据。我们有个电商点击率模型，上线一周后AUC从0.82掉到0.71。查日志发现，每天凌晨3点有大量400 Bad Request，错误信息是ValueError: Input contains NaN。追查源头，是上游订单系统在批量导入历史数据时，把user_age字段全设成了空字符串""，而我们的模型代码里只写了df['user_age'].fillna(0)，没处理字符串转数字的异常。这就是典型的“校验缺失”。Part 4强制要求：所有API入口，必须有两层校验。第一层是FastAPI的Pydantic模型，定义字段类型、范围、是否必填。比如：

class PredictionRequest(BaseModel): user_id: str = Field(..., min_length=5, max_length=32) item_id: str = Field(..., min_length=5, max_length=32) user_age: Optional[float] = Field(None, ge=0, le=120) # ge=greater than or equal, le=less than or equal

第二层是业务逻辑校验，在predict()函数内部，对经过Pydantic解析后的数据，做领域规则检查。比如：

def predict(request: PredictionRequest): if request.user_age is None: raise HTTPException(status_code=400, detail="user_age cannot be null for this model version") if request.user_age < 16: logger.warning(f"Underage user detected: {request.user_id}") # 触发特殊处理逻辑，如降权或拒绝

我们还专门建了一个data_quality模块，对每个请求的原始JSON做采样分析，统计user_age为空的比例、item_id长度分布等，一旦超过阈值（如空值率>0.5%），自动触发告警并暂停该接口的流量。这招帮我们提前发现了3次上游数据管道的故障，避免了模型效果劣化。

3.3 第三关：特征获取的“雪崩防护”

模型效果70%取决于特征。但特征获取往往是链路中最不稳定的环节。我们有个模型依赖实时用户行为特征，需要调用一个内部HTTP服务。某天该服务响应时间从50ms飙升到2s，我们的模型服务QPS瞬间归零——因为所有worker都在傻等那个HTTP请求。这就是“雪崩”。防护手段有三重：超时、熔断、降级。第一，HTTP客户端必须设硬超时：httpx.AsyncClient(timeout=Timeout(3.0, connect=1.0, read=2.0))，连接1秒，读取2秒，总超时3秒。第二，用tenacity库实现熔断：连续5次超时后，自动熔断30秒，在此期间所有请求直接返回预设的默认特征向量（如全0向量），并记录circuit_breaker_open指标。第三，降级策略：当熔断开启时，自动切换到Redis缓存的TTL为5分钟的特征快照，保证服务可用性。我们把这三重防护封装成一个FeatureFetcher类，所有模型服务统一调用，而不是每个工程师自己写requests.get()。上线后，该特征服务宕机2小时，我们的模型服务P99延迟仅从47ms升到89ms，错误率保持0%，业务方完全无感知——这才是生产级的韧性。

3.4 第四关：日志——你唯一的“事故现场勘查员”

生产环境没有IDE，没有断点，日志是你唯一能回溯真相的线索。但很多人的日志是这样的：“模型预测完成”，“发生错误”。这等于没记。Part 4的日志规范是：结构化、上下文完整、可追溯、可聚合。我们用structlog替代logging，每条日志都是JSON：

{ "event": "prediction_completed", "request_id": "req_abc123", "model_version": "v2.4.1", "input_hash": "sha256:...", "inference_time_ms": 23.4, "output_score": 0.872, "timestamp": "2024-05-20T08:15:22.123Z" }

request_id是关键，它贯穿整个请求生命周期：从API网关生成，透传给模型服务，再透传给特征服务、缓存服务。这样，当一个请求出问题时，运维同学只要搜request_id，就能把所有相关服务的日志串起来。input_hash是输入数据的SHA256摘要，用于快速定位“是不是同一个输入在不同环境表现不同”。我们还强制要求，所有异常日志必须包含exc_info=True，且logger.exception()必须放在except块的最末尾，确保堆栈完整。有一次，一个KeyError只在特定用户ID下触发，靠request_id和input_hash，我们15分钟就定位到是上游数据清洗脚本漏掉了某个国家的邮编格式转换——没有这些日志字段，这事得查两天。

3.5 第五关：健康检查与就绪探针——让K8s真正“懂”你的服务

K8s的livenessProbe和readinessProbe不是摆设。很多人配个curl http://localhost:8000/healthz就完事，结果服务明明卡死了，探针还能返回200。真正的健康检查必须反映服务的真实就绪状态。我们的/healthz接口，不只是检查进程存活，而是做三件事：第一，检查模型文件是否可读、是否在内存中加载成功；第二，检查Redis连接是否正常，执行PING命令；第三，检查一个轻量级的“自检”模型是否能正常推理（比如一个只有2个特征的LR模型）。只有这三项全通过，才返回200。/readyz则更严格，它还要检查特征缓存的命中率是否高于95%，如果低于阈值，说明特征服务可能有问题，就返回503，让K8s把流量从这个Pod摘除。我们还加了/metrics端点，暴露4个核心指标：model_load_success_total（模型加载成功次数）、prediction_request_total（总请求数）、prediction_latency_seconds（延迟直方图）、feature_cache_hit_ratio（特征缓存命中率）。这些指标被Prometheus定时抓取，Grafana里一张图就能看清服务水位。上周，feature_cache_hit_ratio突然从98%掉到65%，我们立刻知道是Redis集群扩容导致连接抖动，而不是模型本身的问题——这就是可观测性带来的决策效率。

3.6 第六关：配置管理——别让密码和密钥躺在代码里

config.py里写着DB_PASSWORD = "my_secret"？这是生产环境的自杀行为。Part 4要求：所有配置必须外部化、分环境、加密传输。我们用HashiCorp Vault做密钥管理。服务启动时，通过K8s Service Account Token，向Vault申请一个临时Token，用它拉取production/model-service路径下的密钥，包括数据库密码、API密钥、模型文件的S3访问密钥。这些密钥不落地，只存在于内存中。环境变量只用来传Vault地址和初始Token路径，绝不传业务密钥。对于非敏感配置，如模型路径、超时时间，我们用K8s ConfigMap，但必须配合envFrom方式注入，而不是valueFrom，因为后者会让配置项变成Pod的环境变量，容易被ps aux看到。我们还写了config_validator.py，在服务启动时，校验所有必需配置项是否都存在、类型是否正确（比如TIMEOUT_SECONDS必须是int），缺失或错误就直接sys.exit(1)，绝不带病上岗。有一次，一个新同学忘了在生产ConfigMap里加MODEL_VERSION，服务启动失败，K8s自动重启，但config_validator在第二次启动前就报错退出，避免了服务在无模型版本号的状态下“裸奔”。

3.7 第七关：模型版本与回滚——上线不是终点，而是监控的起点

上线不是发布按钮一按就完事。Part 4的最后一个环节，是建立可审计、可追溯、可秒级回滚的模型版本管理体系。我们不用Git标签管理模型，因为模型文件太大，Git不堪重负。我们用MinIO对象存储，路径规范为s3://models/{project_name}/{model_name}/v{version}/{timestamp}/，比如s3://models/recommender/click_model/v2.4.1/20240520T081522/。每次CI/CD流水线构建，都会生成一个manifest.json，记录模型哈希、训练数据版本、特征工程代码Commit ID、评估指标（AUC、F1等）。这个Manifest是模型的“出生证明”。上线时，我们不直接替换线上模型，而是用K8s的ConfigMap指向新的S3路径，然后滚动更新Deployment。回滚？只需把ConfigMap里的路径改回上一个版本，kubectl apply，30秒内全量生效。我们还强制要求，每个新版本上线后，必须自动触发一个“影子流量”任务：把1%的真实请求，同时发送给新旧两个版本，对比输出差异。如果新版本的output_score与旧版本的绝对差值超过0.1的概率大于5%，就自动触发告警，并暂停灰度。这套机制，让我们在过去一年里，实现了0次因模型更新导致的P0级事故。

4. 实操过程与核心环节实现：手把手搭建一个可上线的模型服务

现在，我们把前面所有原则，落地成一个可直接运行的实操流程。以一个简化的“用户流失预警”模型为例，目标是构建一个符合Part 4标准的FastAPI服务。整个过程分为5个阶段，每个阶段都有可验证的产出物。

4.1 阶段一：环境准备与基础镜像构建（15分钟）

首先，创建一个最小化、可复现的基础镜像。新建Dockerfile.base：

# 使用明确的OS版本，避免滚动更新风险 FROM python:3.11-slim-bookworm # 设置时区，避免日志时间错乱 ENV TZ=Asia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone # 安装编译依赖和CA证书 RUN apt-get update && apt-get install -y \ gcc \ libpq-dev \ && rm -rf /var/lib/apt/lists/* # 创建非root用户，提升安全性 RUN addgroup -g 1001 -f mlgroup && adduser -S mluser -u 1001 # 切换到非root用户 USER mluser # 设置工作目录 WORKDIR /app

构建并推送：

docker build -f Dockerfile.base -t your-registry/ml-base:3.11-202405 . docker push your-registry/ml-base:3.11-202405

验证：docker run --rm -it your-registry/ml-base:3.11-202405 python --version应输出Python 3.11.x。这一步看似简单，但它锁定了Python解释器、OS内核、时区三大基石，是后续所有确定性的前提。

4.2 阶段二：模型转换与ONNX验证（20分钟）

假设你有一个训练好的sklearn.ensemble.RandomForestClassifier模型，保存在model.pkl。先安装转换工具：

pip install scikit-learn onnx sklearn-onnx onnxruntime

转换脚本convert_to_onnx.py：

from sklearn.ensemble import RandomForestClassifier from sklearn.preprocessing import StandardScaler from sklearn.pipeline import Pipeline import numpy as np from skl2onnx import convert_sklearn from skl2onnx.common.data_types import FloatTensorType import joblib import onnx # 加载原始模型 model = joblib.load("model.pkl") # 构建一个与训练时完全一致的Pipeline（注意：必须用相同参数） pipeline = Pipeline([ ('scaler', StandardScaler(use_double=False)), # 关键！use_double=False ('classifier', model) ]) # 定义输入类型：假设模型有10个float特征 initial_type = [('float_input', FloatTensorType([None, 10]))] # 转换 onnx_model = convert_sklearn(pipeline, initial_types=initial_type) # 验证 onnx.checker.check_model(onnx_model) # 保存 onnx.save(onnx_model, "model.onnx") print("ONNX conversion successful!")

运行后，得到model.onnx。用onnxruntime验证：

import onnxruntime as ort import numpy as np sess = ort.InferenceSession("model.onnx") # 用一个随机输入测试 dummy_input = np.random.rand(1, 10).astype(np.float32) result = sess.run(None, {"float_input": dummy_input}) print("ONNX inference OK:", result[0].shape) # 应输出 (1, 2)

这一步确保了模型能在生产环境的ONNX Runtime上正确加载和推理，是脱离Python版本束缚的关键。

4.3 阶段三：FastAPI服务骨架与核心逻辑（30分钟）

创建项目结构：

churn-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── models.py # Pydantic数据模型 │ ├── inference.py # ONNX推理核心 │ └── health.py # 健康检查 ├── Dockerfile ├── requirements.lock └── pyproject.toml

app/models.py定义输入输出：

from pydantic import BaseModel, Field from typing import Optional, List class ChurnPredictionRequest(BaseModel): user_id: str = Field(..., description="Unique user identifier") tenure_months: float = Field(..., ge=0, le=240, description="User tenure in months") monthly_charges: float = Field(..., ge=0, le=200, description="Monthly charge amount") total_charges: float = Field(..., ge=0, le=10000, description="Total charges to date") # ... 其他8个特征字段 class ChurnPredictionResponse(BaseModel): user_id: str churn_probability: float = Field(..., ge=0, le=1, description="Predicted probability of churn") is_churn: bool = Field(..., description="Binary prediction: True if probability >= 0.5") model_version: str = Field(..., description="Version of the serving model")

app/inference.py封装ONNX推理：

import onnxruntime as ort import numpy as np from pathlib import Path import logging logger = logging.getLogger(__name__) class ONNXModel: def __init__(self, model_path: str): self.model_path = Path(model_path) self.session = None self.input_name = None self.output_name = None self._load_model() def _load_model(self): try: self.session = ort.InferenceSession(str(self.model_path)) self.input_name = self.session.get_inputs()[0].name self.output_name = self.session.get_outputs()[0].name logger.info(f"ONNX model loaded successfully from {self.model_path}") except Exception as e: logger.error(f"Failed to load ONNX model: {e}") raise def predict(self, input_data: np.ndarray) -> np.ndarray: # 输入必须是float32，且维度正确 if input_data.dtype != np.float32: input_data = input_data.astype(np.float32) if len(input_data.shape) == 1: input_data = input_data.reshape(1, -1) try: result = self.session.run( [self.output_name], {self.input_name: input_data} ) return result[0] except Exception as e: logger.error(f"ONNX inference failed: {e}") raise # 全局单例，避免重复加载 model_instance = ONNXModel("/app/model.onnx")

app/main.py是FastAPI主程序：

from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from app.models import ChurnPredictionRequest, ChurnPredictionResponse from app.inference import model_instance from app.health import router as health_router import time import logging logger = logging.getLogger(__name__) app = FastAPI(title="Churn Prediction Service", version="1.0.0") # 添加CORS中间件（生产环境应限制origins） app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 挂载健康检查路由 app.include_router(health_router, prefix="/health") @app.post("/predict", response_model=ChurnPredictionResponse) async def predict(request: ChurnPredictionRequest): start_time = time.time() try: # 将Pydantic模型转换为numpy数组（按特征顺序） # 这里简化，实际应有完整的特征工程映射 input_array = np.array([ request.tenure_months, request.monthly_charges, request.total_charges, # ... 其他7个特征 ], dtype=np.float32).reshape(1, -1) # ONNX推理 raw_output = model_instance.predict(input_array) # raw_output shape: (1, 2), [prob_not_churn, prob_churn] churn_prob = float(raw_output[0][1]) # 取churn class概率 # 构建响应 response = ChurnPredictionResponse( user_id=request.user_id, churn_probability=churn_prob, is_churn=churn_prob >= 0.5, model_version="v1.0.0" # 从配置或环境变量读取 ) # 记录结构化日志 logger.info( "prediction_completed", extra={ "request_id": "req_temp", # 实际应从请求头提取 "user_id": request.user_id, "churn_probability": churn_prob, "inference_time_ms": (time.time() - start_time) * 1000, "model_version": "v1.0.0" } ) return response except Exception as e: logger.error(f"Prediction failed for user {request.user_id}: {e}", exc_info=True) raise HTTPException(status_code=500, detail="Internal server error")

这个骨架已经包含了输入校验、ONNX推理、结构化日志、错误处理四大核心要素。

4.4 阶段四：Docker化与K8s部署清单（25分钟）

Dockerfile基于我们之前构建的ml-base：

FROM your-registry/ml-base:3.11-202405 # 复制ONNX模型（生产环境应从S3下载，此处为演示） COPY model.onnx /app/model.onnx # 复制应用代码 COPY app/ /app/ # 安装应用依赖（使用锁定文件） COPY requirements.lock /app/requirements.lock RUN pip install --no-cache-dir -r /app/requirements.lock # 设置非root用户 USER mluser # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4"]

requirements.lock内容示例（精简）：

fastapi==0.110.0 onnxruntime==1.17.1 pydantic==2.7.1 structlog==24.1.0 uvicorn==0.29.0

K8s部署清单k8s/deployment.yaml：

apiVersion: apps/v1 kind: Deployment metadata: name: churn-service spec: replicas: 3 selector: matchLabels: app: churn-service template: metadata: labels: app: churn-service spec: serviceAccountName: ml-service-account # 需提前创建，有Vault访问权限 containers: - name: churn-service image: your-registry/churn-service:v1.0.0 ports: - containerPort: 8000 livenessProbe: httpGet: path: /healthz port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /readyz port: 8000 initialDelaySeconds: 20 periodSeconds: 5 resources: requests: memory: "512Mi" cpu: "250m" limits: memory: "1Gi" cpu: "500m" env: - name: VAULT_ADDR value: "https://vault.internal" - name: VAULT_ROLE value: "ml-service-role" --- apiVersion: v1 kind: Service metadata: name: churn-service spec: selector: app: churn-service ports: - port: 80 targetPort: 8000 type: ClusterIP

部署命令：

kubectl apply -f k8s/deployment.yaml kubectl port-forward svc/churn-service 8000:80 # 本地测试

4.5 阶段五：本地验证与压测（20分钟）

本地验证三步走：

1. 功能验证：curl -X POST http://localhost:8000/predict -H "Content-Type: application/json" -d '{"user_id":"u123","tenure_months":12,"monthly_charges":50.0,"total_charges":600.0}'
2. 健康检查：curl http://localhost:8000/healthz应返回{"status":"ok"}
3. 文档查看：访问http://localhost:8000/docs，确认Swagger UI正常，且请求体结构与ChurnPredictionRequest一致。

压测用hey工具：

hey -n 1000 -c 100 -m POST -H "Content-Type: application/json" -d '{"user_id":"u123","tenure_months":12,"monthly_charges":50.0,"total_charges":600.0}' http://localhost:8000/predict

关注报告中的Requests/sec（应>300）、Latency distribution（P99应<100ms）、Error rate（应为0%）。如果失败，优先检查Docker日志：docker logs <container_id>，看是否有ONNX加载失败或依赖缺失错误。

5. 常见问题与排查技巧实录：那些让你半夜爬起来的坑

即使严格按照上述流程，上线路上依然布满地雷。我把过去踩过的、帮客户解决的、以及社区高频提问的典型问题，整理成