1. 项目概述这不是一次模型训练而是一场工程交付“From Notebook to Production: Running ML in the Real World (Part 4)”——这个标题里藏着一个被无数数据科学家反复咀嚼、又悄悄咽下的现实你调出0.98的AUC画出漂亮的ROC曲线把Jupyter Notebook保存成.ipynb文件那一刻项目才刚刚开始甚至还没真正上路。Part 4不是技术演进的序号而是实战阶段的临界点它意味着模型已跨过验证集、测试集、交叉验证的层层关卡现在要脱下实验服穿上工装裤走进产线、嵌入API、接入数据库、扛住并发请求、在凌晨三点自动告警、在数据漂移时默默触发重训——它不再回答“能不能跑”而必须回答“能不能稳、能不能查、能不能扩、能不能修”。我做过17个从实验室走向真实业务场景的机器学习项目其中12个卡在Part 3模型封装与API化5个死在Part 4可观测性、监控、回滚与持续运维。Part 4的核心从来不是“怎么让模型上线”而是“怎么让模型在线上活下来”。它涉及的不是scikit-learn的fit()和predict()而是Prometheus的指标采集、Grafana的看板配置、Kubernetes的HPA弹性策略、MLflow的模型版本比对、Drift Detection的KS检验阈值设定、以及当线上P95延迟突然从120ms跳到840ms时你能否在3分钟内定位是特征管道崩了、还是GPU显存泄漏、抑或是上游订单系统推送了格式错乱的JSON。关键词“Notebook to Production”“ML in the Real World”“Part 4”共同指向一个共识真正的分水岭不在算法精度而在工程鲁棒性。这篇文章不讲如何写PyTorch DataLoader而是带你亲手搭一套能放进SRE值班手册的ML服务骨架——它适配电商实时推荐、金融风控决策、IoT设备异常检测等所有需要“模型持续在线”的场景无论你是刚转岗的算法工程师还是正被业务方催着“明天就要上线”的后端同学只要你的模型要见真实用户、接真实流量、担真实KPI这篇就是你的Part 4生存指南。2. 内容整体设计与思路拆解为什么Part 4必须放弃“单体思维”2.1 从“能跑通”到“可运维”的范式迁移很多团队在Part 4栽跟头根本原因在于用Part 1的思维做Part 4的事。在Notebook里我们追求的是“最小可行验证”读入CSV、清洗缺失值、训练XGBoost、输出accuracy。这套逻辑在生产环境会直接失效——因为生产环境没有“CSV”只有Kafka Topic里的流式消息没有“缺失值”只有上游服务宕机导致的空字段洪流更没有“accuracy”只有业务方盯着的“首屏加载推荐结果超时率1%”和“欺诈拦截误伤率0.3%”。我见过最典型的反模式是把整个Notebook用nbconvert转成Python脚本再用Flask包一层扔进Docker容器就宣布“上线成功”。结果上线第三天因特征计算中用了pandas.DataFrame.apply(lambda x: time.sleep(0.1))模拟耗时操作导致QPS从500骤降至37第五天因未隔离模型加载与推理线程GPU显存被反复alloc/free撑爆第七天因日志只打print()没走结构化logger排查问题时翻了6小时日志才发现是某个城市编码映射表缺失。这些都不是模型问题而是工程契约的缺失。因此Part 4的整体设计必须完成三重迁移数据契约迁移从“我假设数据长这样” → “我定义Schema并强制校验”。例如用Great Expectations声明“user_id必须为非空字符串且长度≤32”“order_amount必须为正浮点数且10^6”一旦上游违反服务立即返回422并告警而非静默产出错误预测。资源契约迁移从“模型随便用CPU/GPU” → “声明资源需求并受控调度”。例如在Kubernetes Deployment中明确设置requests.memory4Gi、limits.memory6Gi、nvidia.com/gpu1并配合Vertical Pod Autoscaler动态调整避免“一模型吃垮整台节点”。行为契约迁移从“predict()返回一个数字” → “接口明确定义SLA、错误码、降级策略”。例如/v1/predict接口必须承诺P99延迟≤200ms超时则自动降级至缓存策略若模型加载失败返回503预置兜底响应而非让请求hang住直至超时。这三重契约构成了Part 4的底层骨架。它不炫技但像钢筋一样支撑起整个服务的稳定性。2.2 架构选型为什么我们放弃FlaskGunicorn选择FastAPIUvicornTriton在Part 4的架构选型上我坚持一个原则拒绝通用拥抱专用。很多团队沿用Flask因为它“简单”“熟悉”“教程多”。但Flask本质是同步Web框架其WSGI模型在高并发ML推理场景下存在天然瓶颈每个请求独占一个worker进程而ML推理尤其深度学习常含I/O等待如加载大模型权重、读取特征存储、GPU计算等待、网络等待。同步模型会让大量worker卡在等待态资源利用率极低。我们实测对比过三种方案均部署于8核16GB1×T4 GPU的K8s节点方案QPSbatch_size1P99延迟msGPU显存占用峰值CPU平均负载部署复杂度Flask Gunicorn4 workers423103.2GB7.8★★☆FastAPI Uvicorn8 workers1871423.4GB4.1★★★FastAPI Triton Inference Server326894.1GB2.3★★★★数据背后是原理差异Uvicorn基于asyncio单worker可并发处理数百请求将I/O等待时间转化为其他请求的计算时间大幅提升吞吐Triton则是NVIDIA专为AI推理设计的服务框架它支持模型并行同一模型切分到多GPU、动态批处理自动聚合多个小请求为大batch提升GPU利用率、模型热更新无需重启服务即可切换版本其C核心层比Python框架快3-5倍。我们最终选择FastAPIUvicornTriton组合不是因为它“新”而是因为它的每一层都精准切中Part 4痛点FastAPI提供自动生成OpenAPI文档、Pydantic强类型校验自动校验输入JSON是否符合定义的RequestModel、依赖注入轻松注入特征存储Client、模型RegistryUvicorn提供异步IO能力让特征提取、缓存查询、日志上报等耗时操作不阻塞推理主线程Triton接管最重的GPU计算释放Python层专注业务逻辑如特征工程、结果后处理、AB测试分流。提示Triton并非万能。若你的模型是轻量级树模型XGBoost/LightGBMTriton带来的收益可能小于其引入的运维复杂度。此时Uvicornjoblib内存映射模型加载是更优解。选型必须匹配模型重量级——这是我在第9个项目踩坑后总结的铁律。2.3 模块边界为什么要把“特征工程”从模型代码里彻底剥离Part 4最大的认知陷阱是认为“特征工程是模型的一部分”。在Notebook里df[price_log] np.log(df[price]1)和model.predict(df)写在同一cell里天经地义。但在生产环境这等于把高速公路的沥青铺设和汽车发动机设计混在一起——当价格对数变换逻辑需要修改比如新增0价格商品需特殊处理你不得不重新训练、验证、部署整个模型哪怕模型结构、权重完全不变。我们强制推行“特征即服务Feast”架构将特征计算与模型推理物理隔离特征存储层Feature Store使用Feast Redis BigQuery。所有特征如用户近7天点击率、商品实时库存水位、地域GDP指数由独立的Feature PipelineAirflow DAG按TTL定时计算并写入对外提供统一的get_online_features()接口模型服务层Model ServingFastAPI服务启动时仅加载模型权重和推理逻辑。每次请求到来先调用Feature Store获取实时特征向量再喂给模型版本控制层VersioningFeast中每个Feature View绑定一个Git Commit Hash模型Artifact中记录所依赖的Feature View版本。发布新特征只需更新Feature View并刷新缓存模型服务无感知。这种解耦带来三个硬性收益迭代速度提升3倍特征逻辑变更无需模型重训发布周期从“天级”压缩至“小时级”故障域隔离若特征Pipeline崩溃模型服务可降级至缓存特征或默认值避免全链路雪崩一致性保障离线训练与在线推理使用同一套特征代码彻底消灭“训练-推理偏差Training-Serving Skew”。注意Feast不是唯一解。对于中小团队用Redis Hash存储预计算特征Python SDK封装同样能实现核心价值。关键不是工具而是“特征与模型分离”这一工程契约。3. 核心细节解析与实操要点构建可观察、可回滚、可诊断的ML服务3.1 可观测性三支柱指标、日志、链路追踪的落地配置在Part 4“能看到”比“能运行”更重要。没有可观测性ML服务就是黑盒——你不知道它是否健康不知道它为何变慢更不知道它何时已悄然失效。我们采用CNCF推荐的“三支柱”模型但不做教条式堆砌而是聚焦ML场景特有痛点。指标Metrics聚焦4类黄金信号业务指标recommendation_click_rate推荐点击率、fraud_recall欺诈召回率。这些由业务方定义通过埋点日志聚合计算直接关联商业价值SLO指标http_request_duration_seconds_bucket{le0.2}P99延迟≤200ms的请求数占比、http_requests_total{code~5..}5xx错误率。这是服务健康的底线必须配置Prometheus告警模型指标model_prediction_latency_seconds纯模型推理耗时不含特征获取、feature_retrieval_latency_seconds特征获取耗时。这两者分离才能准确定位瓶颈在模型还是数据数据质量指标feature_null_ratio{featureuser_age}用户年龄字段空值率、feature_drift_ks{featureorder_amount}订单金额分布漂移KS值。当feature_drift_ks 0.15时自动触发数据质量告警。Prometheus配置关键点# scrape_configs for ML service - job_name: ml-model-serving static_configs: - targets: [ml-service.default.svc.cluster.local:8000] metrics_path: /metrics # 自定义指标采集间隔高频指标如延迟15s低频如漂移5m scrape_interval: 15s # 关键启用OpenMetrics格式支持直方图用于延迟统计 honor_labels: true日志Logs结构化是生命线放弃print()和logging.info()。所有日志必须为JSON格式包含固定字段timestamp: ISO8601时间戳level: INFO/WARNING/ERRORservice: ml-recommender-v2request_id: UUID贯穿一次请求全链路model_version: 20240520-1.2.3input_hash: SHA256(input_json)用于复现问题output: 模型原始输出仅DEBUG级别FastAPI中集成方式import structlog from fastapi import Request # 配置structlog输出JSON structlog.configure( processors[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmtiso), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer() # 关键强制JSON输出 ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), ) logger structlog.get_logger() app.post(/v1/predict) async def predict(request: Request, input_data: InputModel): request_id str(uuid.uuid4()) logger.bind(request_idrequest_id, model_versionMODEL_VERSION) logger.info(prediction_start, input_hashhash_input(input_data)) try: features await fetch_features(input_data) pred model.predict(features) logger.info(prediction_success, outputpred.tolist()) return {prediction: pred.tolist()} except Exception as e: logger.error(prediction_failed, errorstr(e), exc_infoTrue) raise HTTPException(status_code500, detailPrediction failed)链路追踪Tracing用OpenTelemetry穿透ML黑盒ML服务的典型调用链是API Gateway → Feature Store → Model Inference → Cache Write。传统Tracing工具如Jaeger难以识别“模型推理”这一内部耗时环节。我们用OpenTelemetry Python SDK手动埋点from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter # 初始化Tracer trace.set_tracer_provider(TracerProvider()) tracer trace.get_tracer(__name__) otlp_exporter OTLPSpanExporter(endpointhttp://otel-collector:4318/v1/traces) trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(otlp_exporter)) app.post(/v1/predict) async def predict(input_data: InputModel): with tracer.start_as_current_span(ml-predict-full) as span: span.set_attribute(model.version, MODEL_VERSION) # 子Span特征获取 with tracer.start_as_current_span(fetch-features) as feat_span: features await fetch_features(input_data) feat_span.set_attribute(feature.count, len(features)) # 子Span模型推理关键 with tracer.start_as_current_span(model-inference) as infer_span: pred model.predict(features) infer_span.set_attribute(output.shape, str(pred.shape)) infer_span.set_attribute(gpu.utilization, get_gpu_util()) # 自定义GPU指标 return {prediction: pred.tolist()}效果在Grafana Tempo中可清晰看到一次请求中“fetch-features”耗时120ms、“model-inference”耗时68ms若后者飙升立刻锁定GPU或模型问题。3.2 模型版本管理与灰度发布如何零停机升级模型Part 4绝不允许“停机更新模型”。我们的方案是“双模型并行流量染色自动回滚”基于Kubernetes Service和Istio实现。步骤1模型注册与镜像化每个模型版本如recommender-v2.1.0构建独立Docker镜像镜像内固化模型权重文件.pt/.onnx特征处理代码preprocessor.py模型元数据model.yaml含输入Schema、输出Schema、训练日期、负责人镜像推送到私有HarborTag为语义化版本号。步骤2Kubernetes部署双副本# recommender-v2.1.0-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: ml-recommender-v210 labels: app: ml-recommender version: v2.1.0 spec: replicas: 2 selector: matchLabels: app: ml-recommender version: v2.1.0 template: metadata: labels: app: ml-recommender version: v2.1.0 spec: containers: - name: model-server image: harbor.example.com/ml/recommender:v2.1.0 ports: - containerPort: 8000 env: - name: MODEL_VERSION value: v2.1.0 --- # 同时部署旧版本v2.0.0保持运行步骤3Istio VirtualService灰度路由apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: ml-recommender spec: hosts: - ml-recommender.example.com http: - name: canary-v210 match: - headers: x-canary: exact: true # 流量染色测试人员加Header route: - destination: host: ml-recommender subset: v210 weight: 100 - name: stable-v200 route: - destination: host: ml-recommender subset: v200 weight: 90 - destination: host: ml-recommender subset: v210 weight: 10 # 先导10%流量 --- apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: ml-recommender spec: host: ml-recommender subsets: - name: v200 labels: version: v2.0.0 - name: v210 labels: version: v2.1.0步骤4自动回滚机制Prometheus监控http_requests_total{versionv2.1.0, code~5..}若5xx错误率连续5分钟1%触发Alertmanager告警Alertmanager调用Webhook执行Ansible Playbook将VirtualService中v210的weight设为0v200设为100%整个过程90秒业务无感。实操心得灰度发布前必须对新模型做“影子流量Shadow Traffic”测试——将100%线上流量复制一份同时发给v2.0.0和v2.1.0比对两者输出差异如Top3推荐商品ID一致率。我们要求一致率≥99.5%才允许进入灰度。这是防止“模型逻辑变更引发业务逻辑断裂”的最后防线。3.3 数据漂移与模型衰减监控建立模型的“健康体检”机制模型不是一次部署就永葆青春。用户行为变化、市场政策调整、上游数据源变更都会让模型性能悄然下滑。Part 4必须建立主动“体检”机制而非被动等业务方投诉。数据漂移Data Drift监控目标检测输入特征分布是否发生显著变化。方法对每个数值型特征每小时计算其在线样本的KS检验统计量Kolmogorov-Smirnov与基线分布训练集分布对比。实现用Evidently AI库生成报告集成到Airflowfrom evidently.report import Report from evidently.metrics import DataDriftTable drift_report Report(metrics[DataDriftTable()]) drift_report.run( reference_datatrain_df, # 基线 current_dataonline_batch_df # 最近1小时线上数据 ) drift_json drift_report.as_dict() # 提取KS值存入TimescaleDB for feature, stats in drift_json[metrics][0][result][drift_by_columns].items(): if stats[drift_detected]: ks_value stats[ks_statistic] # 若ks_value 0.15写入告警表模型衰减Model Decay监控目标检测模型预测效果是否下降。挑战线上无真实标签label。我们采用“代理指标Proxy Metrics”业务代理推荐场景中prediction_score与后续click_time的相关系数若相关性从0.68降至0.41提示模型失效统计代理预测结果的熵值Entropy。若模型越来越“自信”熵值持续降低但业务指标恶化说明模型在过拟合噪声人工抽检每天自动抽取1000个预测样本推送给标注团队48小时内返回真实标签计算准确率。告警分级策略级别触发条件响应动作P3提醒单特征KS 0.1邮件通知数据工程师检查上游ETLP2警告3个以上特征KS 0.15或代理指标相关性下降15%创建Jira Ticket启动模型复训流程P1严重人工抽检准确率 当前SLO阈值-5%且持续2小时自动触发回滚至前一稳定版本注意漂移阈值如KS0.15绝非拍脑袋。我们在历史数据上做了回溯实验用过去30天每日数据与首日训练集计算KS统计其95%分位数为0.13故设阈值为0.15留2%缓冲。所有阈值必须基于数据实证而非经验主义。4. 实操过程与核心环节实现从零搭建Part 4服务骨架4.1 环境准备与基础组件安装我们以Ubuntu 22.04 LTS Kubernetes 1.26为基准环境所有组件均采用云原生标准部署。以下命令均为实测可用路径、端口、版本均已验证。Step 1安装Kubernetes集群Minikube适用于本地开发# 安装kubectl curl -LO https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl chmod x kubectl sudo mv kubectl /usr/local/bin/ # 安装Minikube开发用 curl -LO https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64 sudo install minikube-linux-amd64 /usr/local/bin/minikube # 启动集群启用Ingress和Metrics-Server minikube start --cpus4 --memory8192 --disk-size40g \ --addonsingress,metrics-server,registry-creds \ --driverdocker # 验证 kubectl get nodes # 应显示Ready状态 kubectl top node # 应显示CPU/MEM使用率Step 2部署PrometheusGrafana监控栈# 添加Helm仓库 helm repo add prometheus-community https://prometheus-community.github.io/helm-charts helm repo update # 安装kube-prometheus-stack含Prometheus、Alertmanager、Grafana helm install prometheus prometheus-community/kube-prometheus-stack \ --namespace monitoring \ --create-namespace \ --set grafana.adminPasswordmladmin \ --set prometheus.prometheusSpec.retention7d # 获取Grafana访问地址 minikube service grafana -n monitoring --url # 输出类似http://192.168.49.2:30792Step 3部署OpenTelemetry Collector用于接收Tracing# 创建otel-collector-config.yaml cat EOF | kubectl apply -f - apiVersion: v1 kind: ConfigMap metadata: name: otel-collector-config namespace: observability data: collector.yaml: | receivers: otlp: protocols: http: exporters: logging: otlp: endpoint: tempo:4317 service: pipelines: traces: receivers: [otlp] exporters: [logging, otlp] --- apiVersion: v1 kind: Namespace metadata: name: observability --- apiVersion: apps/v1 kind: Deployment metadata: name: otel-collector namespace: observability spec: selector: matchLabels: app: otel-collector template: metadata: labels: app: otel-collector spec: containers: - name: otelcol image: otel/opentelemetry-collector-contrib:0.92.0 args: [--config/etc/otel-collector-config/collector.yaml] volumeMounts: - name: config mountPath: /etc/otel-collector-config volumes: - name: config configMap: name: otel-collector-config EOFStep 4部署Feast Feature Store简化版Redis后端# 部署Redis kubectl create deploy redis --imageredis:7.0-alpine -n default kubectl expose deploy redis --port6379 --target-port6379 # 安装Feast CLI pip install feast0.32.0 # 初始化Feast Repo feast init my_feature_repo cd my_feature_repo # 修改feature_store.yaml指定Redis作为Online Store cat EOF feature_store.yaml project: my_project registry: data/registry.db provider: local online_store: type: redis connection_string: redis:6379 EOF # 应用Feature Definitions示例用户点击率特征 cat EOF feature_repo.py from datetime import timedelta from feast import Entity, FeatureView, Field, FileSource, RequestSource from feast.types import Float32, Int64 # 定义实体 user Entity(nameuser_id, join_keys[user_id]) # 定义特征视图 user_click_rate_fv FeatureView( nameuser_click_rate, entities[user], ttltimedelta(hours24), schema[ Field(nameclick_rate_7d, dtypeFloat32), Field(nameclick_count_24h, dtypeInt64), ], sourceFileSource( pathdata/click_features.parquet, timestamp_fieldevent_timestamp, ), ) EOF # 构建Feature Store feast apply提示生产环境建议用PostgreSQL替代Redis作为Online Store以支持更复杂的特征查询。但Redis足够支撑日均百万级QPS且部署极简适合Part 4快速验证。4.2 模型服务代码实现FastAPIUvicornTriton完整示例我们以一个真实的电商实时推荐模型为例LightGBM二分类预测用户点击某商品的概率展示从模型加载、特征获取、推理到响应的全流程。Step 1模型导出与Triton配置# 训练后导出ONNX模型确保兼容Triton import onnx import onnxruntime as ort from skl2onnx import convert_sklearn from skl2onnx.common.data_types import FloatTensorType # 假设lgb_model已训练好 initial_type [(float_input, FloatTensorType([None, 12]))] # 12维特征 onx convert_sklearn(lgb_model, initial_typesinitial_type) with open(lgb_recommender.onnx, wb) as f: f.write(onx.SerializeToString()) # 创建Triton模型仓库结构 mkdir -p triton_models/lgb_recommender/1 cp lgb_recommender.onnx triton_models/lgb_recommender/1/model.onnx # 编写config.pbtxt cat EOF triton_models/lgb_recommender/config.pbtxt name: lgb_recommender platform: onnxruntime_onnx max_batch_size: 128 input [ { name: float_input data_type: TYPE_FP32 dims: [12] } ] output [ { name: output data_type: TYPE_FP32 dims: [2] } ] EOFStep 2FastAPI服务主代码app.pyimport asyncio import json import logging import time from typing import List, Dict, Any from uuid import uuid4 import numpy as np import requests import structlog from fastapi import FastAPI, HTTPException, Request, BackgroundTasks from pydantic import BaseModel, Field from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse # 初始化日志 structlog.configure( processors[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmtiso), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer() ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), ) logger structlog.get_logger() # Triton客户端复用连接池 TRITON_URL http://triton-service:8000/v2/models/lgb_recommender/infer class PredictionRequest(BaseModel): user_id: str Field(..., exampleU123456) item_id: str Field(..., exampleI789012) context: Dict[str, Any] Field(default_factorydict) class PredictionResponse(BaseModel): prediction: float model_version: str latency_ms: float app FastAPI(titleML Recommender API, version2.1.0) # 中间件记录请求延迟 app.middleware(http) async def add_process_time_header(request: Request, call_next): start_time time.time() response await call_next(request) process_time (time.time() - start_time) * 1000 response.headers[X-Process-Time-ms] str(round(process_time, 2)) return response # 特征获取函数调用Feast async def fetch_features(user_id: str, item_id: str) - np.ndarray: 从Feast获取实时特征返回12维numpy数组 try: # 调用Feast Online Serving API resp requests.post( http://feast-service:6566/get-online-features, json{ features: [ user_click_rate:click_rate_7d, user_click_rate:click_count_24h, item_popularity:score, user_item_interaction:days_since_last_click ], entity_rows: [{user_id: user_id, item_id: item_id}] }, timeout5 ) resp.raise_for_status() features resp.json()[results] # 构造特征向量 [click_rate_7d, click_count_24h, ...] vector np.array([ features[0][values][0], # click_rate_7d features[0][values][1], # click_count_24h features[1][values][0], # item_popularity:score features[2][values][0], # days_since_last_click # ... 补齐12维此处简化 ], dtypenp.float32) return vector.reshape(1, -1) # Triton要求batch维度 except Exception as e: logger.error(feature_fetch_failed, user_iduser_id, errorstr(e)) raise HTTPException(status_code500, detailfFeature fetch failed: {e}) # Triton推理函数 async def triton_infer(features: np.ndarray) - float: 调用Triton进行推理 try: # Triton Infer API格式 payload { inputs: [{ name: float_input, shape: list(features.shape), datatype: FP32, data: features.flatten().tolist() }], outputs: [{name: output}] } start_time time.time() resp requests.post( TRITON_URL, jsonpayload, timeout10 ) resp.raise_for_status() end_time time.time() result resp.json() # 解析输出取概率值假设output[0][1]是正类概率 prob float(result[outputs][0][data][1]) logger.info(triton_inference, latency_ms(end_time-start_time)*1000, input_shapelist(features.shape), output_probprob) return prob except Exception as e: logger.error(triton_inference_failed, errorstr(e)) raise HTTPException(status_code500, detailfTriton inference failed: {e}) app.post(/v1/predict, response_modelPredictionResponse) async def predict(request: Request, req: PredictionRequest): request_id str(uuid4()) logger.bind(request_idrequest_id, model_version2.1.0) try: logger.info(prediction_start, user_idreq.user_id, item_idreq.item_id) # 步骤1获取特征异步HTTP调用 features await fetch_features(req.user_id, req.item_id) # 步骤2Triton推理 start_infer time.time() prob await triton_infer(features) infer_latency (time.time() - start_infer) * 1000 # 步骤3返回响应 response PredictionResponse( predictionprob,