企业官网建设流程全解析

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.memory=4Gi、limits.memory=6Gi、nvidia.com/gpu=1，并配合Vertical Pod Autoscaler动态调整，避免“一模型吃垮整台节点”。
• 行为契约迁移：从“predict()返回一个数字” → “接口明确定义SLA、错误码、降级策略”。例如，/v1/predict接口必须承诺P99延迟≤200ms，超时则自动降级至缓存策略；若模型加载失败，返回503+预置兜底响应，而非让请求hang住直至超时。

这三重契约，构成了Part 4的底层骨架。它不炫技，但像钢筋一样支撑起整个服务的稳定性。

2.2 架构选型：为什么我们放弃Flask+Gunicorn，选择FastAPI+Uvicorn+Triton

在Part 4的架构选型上，我坚持一个原则：拒绝通用，拥抱专用。很多团队沿用Flask，因为它“简单”“熟悉”“教程多”。但Flask本质是同步Web框架，其WSGI模型在高并发ML推理场景下存在天然瓶颈：每个请求独占一个worker进程，而ML推理（尤其深度学习）常含I/O等待（如加载大模型权重、读取特征存储）、GPU计算等待、网络等待。同步模型会让大量worker卡在等待态，资源利用率极低。

我们实测对比过三种方案（均部署于8核16GB+1×T4 GPU的K8s节点）：

数据背后是原理差异：

• Uvicorn基于asyncio，单worker可并发处理数百请求，将I/O等待时间转化为其他请求的计算时间，大幅提升吞吐；
• Triton则是NVIDIA专为AI推理设计的服务框架，它支持模型并行（同一模型切分到多GPU）、动态批处理（自动聚合多个小请求为大batch提升GPU利用率）、模型热更新（无需重启服务即可切换版本），其C++核心层比Python框架快3-5倍。

我们最终选择FastAPI+Uvicorn+Triton组合，不是因为它“新”，而是因为它的每一层都精准切中Part 4痛点：

• FastAPI提供自动生成OpenAPI文档、Pydantic强类型校验（自动校验输入JSON是否符合定义的RequestModel）、依赖注入（轻松注入特征存储Client、模型Registry）；
• Uvicorn提供异步IO能力，让特征提取、缓存查询、日志上报等耗时操作不阻塞推理主线程；
• Triton接管最重的GPU计算，释放Python层专注业务逻辑（如特征工程、结果后处理、AB测试分流）。

提示：Triton并非万能。若你的模型是轻量级树模型（XGBoost/LightGBM），Triton带来的收益可能小于其引入的运维复杂度。此时Uvicorn+joblib内存映射模型加载是更优解。选型必须匹配模型重量级——这是我在第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 Pipeline（Airflow DAG）按TTL定时计算并写入，对外提供统一的get_online_features()接口；
• 模型服务层（Model Serving）：FastAPI服务启动时，仅加载模型权重和推理逻辑。每次请求到来，先调用Feature Store获取实时特征向量，再喂给模型；
• 版本控制层（Versioning）：Feast中每个Feature View绑定一个Git Commit Hash，模型Artifact中记录所依赖的Feature View版本。发布新特征只需更新Feature View并刷新缓存，模型服务无感知。

这种解耦带来三个硬性收益：

1. 迭代速度提升3倍：特征逻辑变更无需模型重训，发布周期从“天级”压缩至“小时级”；
2. 故障域隔离：若特征Pipeline崩溃，模型服务可降级至缓存特征或默认值，避免全链路雪崩；
3. 一致性保障：离线训练与在线推理使用同一套特征代码，彻底消灭“训练-推理偏差（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{le="0.2"}（P99延迟≤200ms的请求数占比）、http_requests_total{code=~"5.."}（5xx错误率）。这是服务健康的底线，必须配置Prometheus告警；
• 模型指标：model_prediction_latency_seconds（纯模型推理耗时，不含特征获取）、feature_retrieval_latency_seconds（特征获取耗时）。这两者分离，才能准确定位瓶颈在模型还是数据；
• 数据质量指标：feature_null_ratio{feature="user_age"}（用户年龄字段空值率）、feature_drift_ks{feature="order_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"/"ERROR"
• service: "ml-recommender-v2"
• request_id: UUID（贯穿一次请求全链路）
• model_version: "20240520-1.2.3"
• input_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(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer() # 关键：强制JSON输出 ], context_class=dict, logger_factory=structlog.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_id=request_id, model_version=MODEL_VERSION) logger.info("prediction_start", input_hash=hash_input(input_data)) try: features = await fetch_features(input_data) pred = model.predict(features) logger.info("prediction_success", output=pred.tolist()) return {"prediction": pred.tolist()} except Exception as e: logger.error("prediction_failed", error=str(e), exc_info=True) raise HTTPException(status_code=500, detail="Prediction 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(endpoint="http://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、训练日期、负责人）
• 镜像推送到私有Harbor，Tag为语义化版本号。

步骤2：Kubernetes部署双副本

# 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（保持运行）

步骤3：Istio 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{version="v2.1.0", code=~"5.."}，若5xx错误率连续5分钟>1%，触发Alertmanager告警；
• Alertmanager调用Webhook，执行Ansible Playbook，将VirtualService中v210的weight设为0，v200设为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库生成报告，集成到Airflow：

from evidently.report import Report from evidently.metrics import DataDriftTable drift_report = Report(metrics=[DataDriftTable()]) drift_report.run( reference_data=train_df, # 基线 current_data=online_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小时内返回真实标签，计算准确率。

告警分级策略

注意：漂移阈值（如KS=0.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 --cpus=4 --memory=8192 --disk-size=40g \ --addons=ingress,metrics-server,registry-creds \ --driver=docker # 验证 kubectl get nodes # 应显示Ready状态 kubectl top node # 应显示CPU/MEM使用率

Step 2：部署Prometheus+Grafana监控栈

# 添加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.adminPassword='mladmin' \ --set prometheus.prometheusSpec.retention='7d' # 获取Grafana访问地址 minikube service grafana -n monitoring --url # 输出类似：http://192.168.49.2:30792

Step 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 EOF

Step 4：部署Feast Feature Store（简化版Redis后端）

# 部署Redis kubectl create deploy redis --image=redis:7.0-alpine -n default kubectl expose deploy redis --port=6379 --target-port=6379 # 安装Feast CLI pip install feast==0.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(name="user_id", join_keys=["user_id"]) # 定义特征视图 user_click_rate_fv = FeatureView( name="user_click_rate", entities=[user], ttl=timedelta(hours=24), schema=[ Field(name="click_rate_7d", dtype=Float32), Field(name="click_count_24h", dtype=Int64), ], source=FileSource( path="data/click_features.parquet", timestamp_field="event_timestamp", ), ) EOF # 构建Feature Store feast apply

提示：生产环境建议用PostgreSQL替代Redis作为Online Store，以支持更复杂的特征查询。但Redis足够支撑日均百万级QPS，且部署极简，适合Part 4快速验证。

4.2 模型服务代码实现：FastAPI+Uvicorn+Triton完整示例

我们以一个真实的电商实时推荐模型为例（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_types=initial_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] } ] EOF

Step 2：FastAPI服务主代码（app.py）

import 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(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer() ], context_class=dict, logger_factory=structlog.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(..., example="U123456") item_id: str = Field(..., example="I789012") context: Dict[str, Any] = Field(default_factory=dict) class PredictionResponse(BaseModel): prediction: float model_version: str latency_ms: float app = FastAPI(title="ML Recommender API", version="2.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}] }, timeout=5 ) 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维（此处简化） ], dtype=np.float32) return vector.reshape(1, -1) # Triton要求batch维度 except Exception as e: logger.error("feature_fetch_failed", user_id=user_id, error=str(e)) raise HTTPException(status_code=500, detail=f"Feature 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, json=payload, timeout=10 ) 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_shape=list(features.shape), output_prob=prob) return prob except Exception as e: logger.error("triton_inference_failed", error=str(e)) raise HTTPException(status_code=500, detail=f"Triton inference failed: {e}") @app.post("/v1/predict", response_model=PredictionResponse) async def predict(request: Request, req: PredictionRequest): request_id = str(uuid4()) logger.bind(request_id=request_id, model_version="2.1.0") try: logger.info("prediction_start", user_id=req.user_id, item_id=req.item_id) # 步骤1：获取特征（异步HTTP调用） features = await fetch_features(req.user_id, req.item_id) # 步骤2：Triton推理 start_infer = time.time() prob = await triton_infer(features) infer_latency = (time.time() - start_infer) * 1000 # 步骤3：返回响应 response = PredictionResponse( prediction=prob,