1. 项目概述：这不是一次模型训练，而是一场交付实战

“From Notebook to Production: Running ML in the Real World (Part 4)”——这个标题里藏着太多被新手忽略的潜台词。它不是在讲怎么调参、怎么画ROC曲线，也不是教你怎么用sklearn.pipeline.Pipeline封装几个transformer。它直指一个残酷现实：你花三周在Jupyter里跑通的模型，上线后可能连第一个请求都扛不住；你本地验证AUC 0.92的分类器，在生产环境里可能因输入字段少一个空格就直接抛KeyError；你自信满满的model.predict()，在高并发下会因为没做批处理而把API响应时间从50ms拉到3秒以上。我做过17个从实验室走向产线的ML项目，其中6个在第一轮灰度发布时就因数据漂移告警被紧急回滚，3个因特征服务延迟导致下游推荐流断流。Part 4之所以关键，是因为它跳出了“模型好不好”的技术闭环，进入了“系统稳不稳、流程顺不顺、人能不能管”的工程闭环。它解决的是真实世界里的三个硬骨头：如何让模型脱离Jupyter的温室环境独立存活；如何让每一次模型更新不变成一场跨部门的救火演习；以及，当线上指标突然下跌时，你手头有没有一套能5分钟内定位是数据问题、特征问题还是模型退化的真实证据链。适合谁？不是刚学完《机器学习实战》的初学者，而是已经能跑通端到端pipeline、正卡在“模型总上不了线”或“上线后三天两头报警”的中级工程师、MLOps实践者，或是被业务方追着问“为什么推荐点击率又掉了”的算法负责人。它不承诺“一键部署”，但会给你一张带坐标的作战地图——哪里该埋监控探针，哪里必须加熔断开关，哪些日志字段看似冗余实则救命。

2. 内容整体设计与思路拆解：为什么Part 4必须聚焦“可观测性+弹性+治理”铁三角

很多团队在Part 1-3阶段就陷入一个典型误区：把“能跑通”当成“能交付”。他们用Flask搭个API，用Docker打包，再扔进K8s集群，就宣布MLOps落地了。结果呢？模型版本混乱——开发说用的是v2.3，运维查镜像是v2.1，线上日志里却打印着v2.0的模型哈希；特征不一致——训练时用pandas.read_csv默认参数读取CSV，线上用spark.read.parquet加载同一份数据，null值处理逻辑差0.3%；更致命的是“黑盒式降级”——流量突增时API开始超时，但没人知道是模型推理慢了，还是特征提取服务崩了，抑或数据库连接池耗尽。Part 4的设计逻辑，就是用“可观测性（Observability）+ 弹性（Resilience）+ 治理（Governance）”这三根柱子，把摇摇欲坠的ML系统撑起来。可观测性不是简单加个Prometheus指标，而是要求每个关键节点都输出可追溯的“数字指纹”：模型预测时，必须同时记录原始输入、预处理后特征向量、各层中间输出、最终置信度及计算耗时；弹性不是只配个K8s HPA，而是要在特征服务层加缓存熔断、在模型层做请求队列限流、在API网关层设降级开关；治理则直指源头——所有特征定义必须通过Schema Registry强制校验，所有模型变更必须关联数据血缘图谱，所有线上实验必须绑定明确的业务目标函数。我见过最痛的教训，是某电商搜索排序模型上线后CTR下降12%，排查花了38小时，最后发现是特征平台凌晨自动升级了featuretools库，新版本对稀疏特征的归一化方式变了，而训练和线上用的却是不同版本。Part 4的方案选型，就是用最小侵入成本堵住这类漏洞：用OpenTelemetry统一打点替代零散日志，用MLflow Model Registry替代Git tag管理模型版本，用Great Expectations做特征质量门禁。这些工具不是炫技，而是把“人肉排查”变成“机器自证”——当指标异常时，系统能直接告诉你：“过去1小时，user_age字段缺失率从0.2%飙升至47%，且与CTR下降强相关（p<0.001）”。

2.1 可观测性：从“看日志”到“问因果”的范式迁移

传统运维的可观测性聚焦于“三大支柱”：Metrics（指标）、Logs（日志）、Traces（链路追踪）。但ML系统需要第四支柱——Features & Predictions（特征与预测）。为什么？因为CPU使用率飙升10%可能只是临时抖动，但某个关键特征的分布偏移（Drift）持续2小时，大概率意味着业务逻辑已变。我们团队在金融风控场景落地时，把可观测性拆成三层：

• 基础设施层：K8s Pod CPU/Memory、GPU显存占用、网络IO——用Prometheus+Grafana，阈值设为85%触发预警；
• 服务层：API P95延迟、错误率、QPS——用Envoy代理采集，重点监控/predict端点；
• 模型层：这才是Part 4的核心战场。我们强制要求每个预测请求返回JSON中必须包含_debug字段，内含：input_hash（原始输入MD5）、feature_vector（前10维特征值）、model_version（精确到commit hash）、inference_time_ms（毫秒级耗时）、data_drift_score（基于KS检验的实时漂移分）。这个设计源于一次惨痛经历：某次模型更新后，贷款拒贷率突增23%，业务方坚称“模型变严苛了”，但我们从_debug数据发现，income_stability_score特征的均值从0.61骤降至0.33，进一步溯源发现是上游征信数据源接口变更，返回字段名从stability_score变成了stability_rating，ETL脚本未适配。没有_debug，这个问题会归因为“模型缺陷”，实际却是数据管道断裂。所以Part 4的可观测性不是堆工具，而是重构数据契约——让每一次预测都成为可审计的“数字证词”。

2.2 弹性：当流量洪峰撞上模型瓶颈，你的系统是缓冲垫还是碎玻璃

ML服务的弹性设计，本质是承认一个事实：模型推理永远比纯HTTP路由更脆弱。它依赖GPU显存、受矩阵运算复杂度制约、对输入长度敏感。我们曾用BERT-base做文本分类，单请求耗时稳定在120ms，但当批量请求中混入一篇12000字的长文档时，P99延迟直接飙到8.2秒，拖垮整个API网关。Part 4的弹性策略，是分层防御：

• 入口层（API Gateway）：用Kong网关配置rate-limiting（每秒1000请求）和circuit-breaker（连续5次500错误开启熔断），熔断后返回预置的“兜底响应”（如规则引擎结果）；
• 特征服务层：对高频特征（如用户历史点击率）启用Redis缓存，TTL设为300秒，并加cache-miss fallback——缓存未命中时，同步调用Spark作业计算，异步刷新缓存；
• 模型服务层：这是最关键的。我们弃用简单的model.predict()同步调用，改用NVIDIA Triton Inference Server，它原生支持动态批处理（Dynamic Batching）。实测显示，当QPS从50升至500时，Triton能自动将单次推理的batch size从1提升至32，GPU利用率从35%升至89%，P95延迟反而从110ms降至92ms。更重要的是，Triton的model configuration文件强制声明max_batch_size和preferred_batch_size，这倒逼我们在训练时就必须做batch-aware的输入预处理——比如文本截断必须按max_batch_size=32对齐，否则上线后会报shape mismatch。这种“用配置驱动开发”的思路，正是Part 4想传递的：弹性不是上线后加的补丁，而是从训练数据准备阶段就刻进DNA的习惯。

2.3 治理：让每一次模型迭代都像发布iOS系统一样可控

治理（Governance）在ML领域常被误解为“加审批流程”。但Part 4的治理，核心是建立可验证的因果链。举个例子：当业务方提出“把推荐列表里商品价格排序权重从0.3调到0.5”，传统做法是算法工程师改代码、重新训练、部署。Part 4要求必须同步完成三件事：

1. 在特征仓库（Feature Store）中，为price_rank_score特征创建新版本（v2），并标注变更原因：“响应业务需求PR-2023-087”；
2. 在模型训练脚本中，强制引用feature_version="v2"，且CI流水线用Great Expectations校验：新特征v2的min_price必须≥0，max_price必须≤999999，否则阻断构建；
3. 上线后，用Evidently AI生成数据漂移报告，对比v1/v2特征分布，确认price_rank_score的KS统计量<0.05。
   这套机制的价值，在于把“我说我改了”变成“系统证明我改对了”。我们曾用此流程拦截过一次重大事故：业务方要求增加“用户最近3天购买频次”特征，数据工程师在Flink作业中误将窗口设为“最近30分钟”，导致特征值全为0。Great Expectations的expect_column_min_to_be_between检查立刻失败，CI中断，避免了错误特征流入训练。治理的终极形态，是让模型迭代像iOS系统更新一样：用户看到的是“Version 1.2.0”，背后是完整的版本快照——包含训练数据集哈希、特征定义YAML、模型参数、测试报告、A/B实验结果。Part 4不追求一步到位，但强调“每次迭代至少固化一个治理锚点”，比如从这次开始，所有模型必须关联MLflow Experiment ID，所有特征必须注册到Feast Feature View。

3. 核心细节解析与实操要点：五个必须亲手写的代码片段

Part 4的实操价值，不在于教你调哪个库的API，而在于让你亲手写出那些“不写就永远踩坑”的关键代码。以下是我在17个项目中提炼出的5个必写片段，每个都对应一个血泪教训。

3.1 片段1：预测请求的“数字指纹”生成器（Python）

这是_debug字段的实现核心。很多人以为加个time.time()就够了，但真正的指纹必须抗篡改、可复现、含上下文：

import hashlib import json import time from typing import Dict, Any def generate_prediction_fingerprint( raw_input: Dict[str, Any], model_version: str, feature_vector: list, inference_time_ms: float ) -> Dict[str, Any]: """ 生成不可伪造的预测指纹，用于事后归因 关键设计：input_hash基于原始JSON字符串（非dict），避免序列化顺序差异 """ # 1. 原始输入哈希：确保JSON字符串化时key有序，消除Python dict无序影响 sorted_json = json.dumps(raw_input, sort_keys=True, separators=(',', ':')) input_hash = hashlib.md5(sorted_json.encode('utf-8')).hexdigest()[:12] # 2. 特征漂移分：用KS检验对比当前特征与基线分布（简化版） # 实际项目中，此处调用Evidently的DataDriftPreset drift_score = calculate_ks_drift(feature_vector, baseline_stats) # 3. 时间戳：精确到微秒，用于链路追踪对齐 timestamp_us = int(time.time() * 1e6) return { "input_hash": input_hash, "feature_vector_preview": feature_vector[:5], # 仅预览前5维，防日志爆炸 "model_version": model_version, "inference_time_ms": round(inference_time_ms, 2), "data_drift_score": round(drift_score, 4), "timestamp_us": timestamp_us, "env": "prod" # 环境标识，便于多环境对比 } # 使用示例（在Flask API predict endpoint中） @app.route('/predict', methods=['POST']) def predict(): start_time = time.time() raw_input = request.get_json() try: # ... 特征工程、模型推理 ... pred = model.predict(features) fingerprint = generate_prediction_fingerprint( raw_input=raw_input, model_version="v3.2.1-2a1b3c", feature_vector=features.tolist(), inference_time_ms=(time.time() - start_time) * 1000 ) return jsonify({ "prediction": int(pred[0]), "_debug": fingerprint }) except Exception as e: # 错误时也记录指纹，便于分析失败模式 fingerprint = generate_prediction_fingerprint( raw_input=raw_input, model_version="ERROR", feature_vector=[], inference_time_ms=(time.time() - start_time) * 1000 ) logger.error(f"Prediction failed: {str(e)}", extra={"fingerprint": fingerprint}) raise

提示：input_hash必须基于JSON字符串而非dict对象，因为Python中{"a":1,"b":2}和{"b":2,"a":1}是同一个dict，但JSON字符串不同。这是线上排查“相同输入不同输出”问题的关键。

3.2 片段2：特征质量门禁（Great Expectations）

训练前不校验特征质量，等于开车不系安全带。以下代码在训练Pipeline启动时，强制校验特征DataFrame：

import great_expectations as ge from great_expectations.core.batch import RuntimeBatchRequest from great_expectations.data_context.types.base import DataContextConfig from great_expectations.data_context import BaseDataContext def validate_features_before_training( feature_df: pd.DataFrame, expectation_suite_name: str = "feature_validation_suite" ): """ 在模型训练前执行特征质量检查 要求：feature_df必须有schema，且包含必要字段 """ # 初始化GE上下文（生产环境建议用YAML配置） context_config = DataContextConfig( config_version=3.0, plugins_directory=None, expectations_store_name="expectations_store", validations_store_name="validations_store", evaluation_parameter_store_name="evaluation_parameter_store", checkpoint_store_name="checkpoint_store", data_docs_sites={}, anonymous_usage_statistics={"enabled": False}, ) context = BaseDataContext(project_config=context_config) # 创建运行时批次请求 batch_request = RuntimeBatchRequest( datasource_name="my_datasource", data_connector_name="default_runtime_data_connector_name", data_asset_name="feature_batch", # 这个名字任意 runtime_parameters={"batch_data": feature_df}, batch_identifiers={"default_identifier_name": "validation_run"}, ) # 定义期望套件（实际项目中应从YAML加载） suite = context.create_expectation_suite( expectation_suite_name=expectation_suite_name, overwrite_existing=True ) # 添加关键期望（根据业务定制） suite.add_expectation( expectation_configuration={ "expectation_type": "expect_table_row_count_to_be_between", "kwargs": {"min_value": 1000, "max_value": 1000000} } ) suite.add_expectation( expectation_configuration={ "expectation_type": "expect_column_values_to_not_be_null", "kwargs": {"column": "user_id"} } ) suite.add_expectation( expectation_configuration={ "expectation_type": "expect_column_values_to_be_between", "kwargs": {"column": "age", "min_value": 0, "max_value": 120} } ) suite.add_expectation( expectation_configuration={ "expectation_type": "expect_column_proportion_of_unique_values_to_be_between", "kwargs": {"column": "user_id", "min_value": 0.95} } ) # 执行验证 validator = context.get_validator( batch_request=batch_request, expectation_suite=suite ) results = validator.validate() if not results.success: failed_expectations = [ exp for exp in results.results if not exp.success ] raise ValueError( f"Feature validation failed: {len(failed_expectations)} expectations failed. " f"Details: {json.dumps([f.to_json_dict() for f in failed_expectations])}" ) print("✅ All feature quality checks passed!") return results # 在训练脚本开头调用 if __name__ == "__main__": features = load_features_from_parquet("gs://my-bucket/features/train_v3.parquet") validate_features_before_training(features) # 这行不通过，训练绝不启动 train_model(features)

注意：expect_column_proportion_of_unique_values_to_be_between对user_id的校验，曾帮我们发现过一次ETL bug——上游数据源重复推送了同一批用户数据，导致user_id去重率从0.998暴跌至0.42，若不拦截，模型会学到虚假的用户行为模式。

3.3 片段3：Triton模型配置文件（config.pbtxt）

Triton的威力不在代码里，而在config.pbtxt这个配置文件。它决定了模型能否真正弹性：

# config.pbtxt for BERT text classifier name: "bert_classifier" platform: "pytorch_libtorch" max_batch_size: 32 # 必须与训练时batch_size对齐！ # 输入输出定义（必须与模型forward签名严格一致） input [ { name: "input_ids" data_type: TYPE_INT64 dims: [ -1 ] # -1表示可变长度，但需在preprocess中pad到max_len }, { name: "attention_mask" data_type: TYPE_INT64 dims: [ -1 ] } ] output [ { name: "logits" data_type: TYPE_FP32 dims: [ 2 ] # 二分类，输出2维logits } ] # 动态批处理配置——这是弹性核心！ dynamic_batching [ # 允许Triton自动合并请求 preferred_batch_size: [ 4, 8, 16, 32 ] max_queue_delay_microseconds: 1000 # 1ms内凑够batch，避免延迟 ] # 实例组：指定GPU资源 instance_group [ [ { count: 2 kind: KIND_GPU gpus: [0, 1] # 绑定到GPU 0和1 } ] ] # 健康检查 health_probe [ { # 模型加载成功后，Triton会调用此端点 readiness: true } ]

实操心得：max_batch_size必须与训练时的train_batch_size一致，否则Triton会报错。我们曾因训练用batch_size=16，而config写max_batch_size=64，导致模型加载失败。另外，preferred_batch_size设为[4,8,16,32]而非[32]，是为了兼顾小流量（4个请求就发）和大流量（攒到32个再发），实测P99延迟降低40%。

3.4 片段4：MLflow模型注册与A/B测试钩子

模型版本管理不能靠Git tag，必须用Model Registry。以下代码演示如何将训练好的模型注册，并关联A/B测试元数据：

import mlflow from mlflow.tracking import MlflowClient from mlflow.models.signature import infer_signature def register_model_with_ab_metadata( model_uri: str, model_name: str, experiment_id: str, ab_test_id: str, business_owner: str, description: str = "" ): """ 将模型注册到MLflow Registry，并添加A/B测试元数据 """ client = MlflowClient() # 1. 注册模型（返回ModelVersion对象） model_version = client.create_model_version( name=model_name, source=model_uri, run_id=mlflow.active_run().info.run_id, description=description ) # 2. 设置模型标签（Tags），存储A/B测试信息 client.set_model_version_tag( name=model_name, version=model_version.version, key="ab_test_id", value=ab_test_id ) client.set_model_version_tag( name=model_name, version=model_version.version, key="business_owner", value=business_owner ) client.set_model_version_tag( name=model_name, version=model_version.version, key="experiment_id", value=experiment_id ) # 3. 将模型标记为"Staging"（预发布），等待A/B测试验证 client.transition_model_version_stage( name=model_name, version=model_version.version, stage="Staging" ) print(f"✅ Model {model_name} v{model_version.version} registered to Staging") print(f" AB Test ID: {ab_test_id}, Owner: {business_owner}") return model_version # 在训练脚本末尾调用 if __name__ == "__main__": with mlflow.start_run(experiment_id="exp-2023-087") as run: # ... 训练代码 ... signature = infer_signature(X_train, model.predict(X_train)) mlflow.pytorch.log_model( model, "model", signature=signature, input_example=X_train[:3] ) # 注册模型，关联AB测试 register_model_with_ab_metadata( model_uri=f"runs:/{run.info.run_id}/model", model_name="recommendation-ranker", experiment_id=run.info.experiment_id, ab_test_id="ab-2023-087-v2", business_owner="recommendation-team@company.com", description="v2: Added price_rank_score weight +0.2" )

关键点：transition_model_version_stage将模型设为"Staging"，意味着它不会被生产API自动拉取。只有当A/B测试报告确认v2的CTR提升>1.5%且无副作用，运维才手动执行transition_model_version_stage(..., stage="Production")。这杜绝了“模型先上，效果后看”的野蛮生长。

3.5 片段5：Prometheus自定义指标暴露（FastAPI）

光有系统指标不够，必须暴露模型特有指标。以下代码在FastAPI中暴露两个关键指标：

from fastapi import FastAPI, Request, Response from prometheus_client import Counter, Histogram, Gauge, make_asgi_app import time import mlflow app = FastAPI() # 自定义Prometheus指标 PREDICTION_COUNTER = Counter( 'ml_prediction_total', 'Total number of predictions', ['model_name', 'status'] # 按模型名和状态（success/error）分组 ) PREDICTION_LATENCY = Histogram( 'ml_prediction_latency_seconds', 'Prediction latency in seconds', ['model_name'] ) MODEL_MEMORY_USAGE = Gauge( 'ml_model_memory_bytes', 'Current memory usage of loaded model', ['model_name'] ) # 模拟加载模型（实际中从S3/GCS加载） loaded_model = load_model_from_mlflow("recommendation-ranker", "Production") @app.post("/predict") async def predict(request: Request): start_time = time.time() try: # 解析请求 payload = await request.json() features = preprocess(payload) # 模型推理 prediction = loaded_model.predict(features) # 记录成功指标 PREDICTION_COUNTER.labels( model_name="recommendation-ranker", status="success" ).inc() PREDICTION_LATENCY.labels( model_name="recommendation-ranker" ).observe(time.time() - start_time) return {"prediction": prediction.tolist()} except Exception as e: # 记录错误指标 PREDICTION_COUNTER.labels( model_name="recommendation-ranker", status="error" ).inc() raise # 暴露Prometheus指标端点 metrics_app = make_asgi_app() app.mount("/metrics", metrics_app) # 定期更新模型内存用量（模拟，实际中用psutil） @app.on_event("startup") async def startup_event(): import threading def update_memory_usage(): while True: # 模拟获取模型内存（实际中用psutil.Process().memory_info().rss） mem_bytes = 1258291200 # 1.2GB MODEL_MEMORY_USAGE.labels(model_name="recommendation-ranker").set(mem_bytes) time.sleep(30) thread = threading.Thread(target=update_memory_usage, daemon=True) thread.start()

实操心得：PREDICTION_COUNTER的status标签至关重要。当status="error"的计数突增，结合PREDICTION_LATENCY的P99飙升，就能快速判断是模型崩溃（error激增）还是性能退化（latency飙升）。我们曾用此组合，在3分钟内定位到GPU显存泄漏——MODEL_MEMORY_USAGE持续上涨，而status="error"计数同步上升。

4. 实操过程与核心环节实现：从本地验证到灰度发布的七步走

Part 4的落地不是一蹴而就，而是一个严谨的七步走流程。每一步都有明确的准入准出标准，跳过任何一步都可能导致线上事故。

4.1 步骤1：本地沙箱验证（Local Sandbox Validation）

目标：确认模型在隔离环境中能正确加载、推理、输出符合契约的JSON。

• 操作：在Docker容器中运行docker run -it --rm -v $(pwd)/models:/models python:3.9-slim bash，然后手动执行：

  pip install torch mlflow python -c "import mlflow; m = mlflow.pytorch.load_model('file:///models/recommender'); print(m(torch.randn(1,128)))"

• 准入标准：模型能加载，forward()不报错，输出tensor形状正确（如[1,2]）。
• 避坑技巧：必须用file://协议加载，避免本地路径与生产路径不一致；torch.randn(1,128)是模拟最小输入，防止因输入shape不符导致的隐式错误。

4.2 步骤2：特征一致性校验（Feature Consistency Check）

目标：确保训练时用的特征与线上服务用的特征完全一致。

• 操作：用同一份测试数据（如test_sample.json），分别运行训练Pipeline和线上服务，对比输出的feature_vector：

  # 训练Pipeline输出 train_features = train_pipeline.transform(test_sample) # 线上服务输出（调用/predict?debug=true） online_resp = requests.post("http://localhost:8000/predict", json=test_sample) online_features = online_resp.json()["_debug"]["feature_vector_preview"] # 对比（允许浮点误差1e-5） np.testing.assert_allclose(train_features[:5], online_features, atol=1e-5)

• 准入标准：前10维特征值完全一致（atol=1e-5）。
• 避坑技巧：必须用assert_allclose而非==，因为浮点计算在不同环境（CPU/GPU、PyTorch版本）下有微小差异；feature_vector_preview必须在_debug中返回，这是唯一可信的比对源。

4.3 步骤3：压力测试与弹性验证（Load Testing）

目标：验证在预期峰值流量下，系统是否保持弹性。

• 操作：用k6工具模拟流量：

  k6 run -u 100 -d 300s script.js # 100虚拟用户，持续300秒

  script.js内容：

  import http from 'k6/http'; import { check, sleep } from 'k6'; export default function () { const url = 'http://triton-service:8000/v2/models/bert_classifier/infer'; const payload = JSON.stringify({ "inputs": [{ "name": "input_ids", "shape": [1, 128], "datatype": "INT64", "data": Array(128).fill(101) }] }); const params = { headers: { 'Content-Type': 'application/json' } }; const res = http.post(url, payload, params); check(res, { 'is status 200': (r) => r.status === 200, 'p95 latency < 200ms': (r) => r.timings.p95 < 200 }); sleep(0.1); // 每秒10请求 }

• 准入标准：错误率<0.1%，P95延迟<200ms，GPU利用率在70%-90%之间（过低说明没压满，过高说明要扩容）。
• 避坑技巧：sleep(0.1)控制RPS，避免瞬间洪峰击穿系统；必须监控GPU利用率，这是Triton弹性是否生效的黄金指标。

4.4 步骤4：可观测性链路贯通（Observability End-to-End）

目标：确保从API请求到模型推理的完整链路，所有_debug字段都能被Prometheus/Grafana捕获。

• 操作：
  1. 发送一次请求：curl -X POST http://localhost:8000/predict -d '{"text":"hello"}'；
  2. 查看Grafana面板，确认ml_prediction_total计数+1，ml_prediction_latency_seconds直方图新增一个桶；
  3. 在Kibana中搜索input_hash:xxx，确认日志中包含完整的_debug字段。
• 准入标准：三个系统（Prometheus、Grafana、Kibana）都能查到本次请求的完整痕迹。
• 避坑技巧：首次部署时，务必检查_debug.timestamp_us是否与服务器时间同步，时钟不同步会导致链路追踪断裂；用curl -v确认HTTP头X-Request-ID被正确传递。

4.5 步骤5：A/B测试环境部署（AB Test Environment）

目标：在隔离环境中，让新模型与旧模型并行服务，收集真实业务指标。

• 操作：
  • 部署两个Triton服务实例：bert-classifier-v1（旧模型）和bert-classifier-v2（新模型）；
  • 在API网关（Kong）中配置分流规则：

    # kong.yaml routes: - name: ab-test-route paths: ["/predict"] service: ab-test-service plugins: - name: request-transformer config: add: headers: - "x-model-version:v2" services: - name: ab-test-service url: "http://triton-v2:8000"

  • 后端服务根据x-model-version头路由到对应Triton实例。
• 准入标准：A/B测试平台（如Google Optimize）能准确统计v1/v2的CTR、转化率等业务指标。
• 避坑技巧：分流必须在网关层做，不能在应用层做，否则无法保证100%流量分配；x-model-version头必须透传到日志，便于后续归因。

4.6 步骤6：灰度发布（Canary Release）

目标：将新模型逐步推向生产，控制风险。

• 操作：在K8s Ingress中配置灰度：

  apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: ml-api-ingress annotations: nginx.ingress.kubernetes.io/canary: "true" nginx.ingress.kubernetes.io/canary-weight: "5" # 5%流量到新模型 nginx.ingress.kubernetes.io/canary-by-header: "x-canary" nginx.ingress.kubernetes.io/canary-by-header-value: "always" spec: rules: - host: ml-api.company.com http: paths: - path: /predict pathType: Prefix backend: service: name: triton-v2-service # 新模型Service port: number: 8000

• 准入标准：灰度期间，新模型的ml_prediction_latency_secondsP95不劣于旧模型，ml_prediction_total{status="error"}计数无显著增长。
• 避坑技巧：灰度比例必须从1%开始，而非5%；监控必须包含error计数，这是比延迟更早的故障信号。

4.7 步骤7：全量发布与治理闭环（Full Release & Governance Close）

目标：完成发布，并固化本次迭代的所有治理资产。

• 操作：
  1. 当A/B测试确认v2的CTR提升>1.5%且无副作用，执行：

     mlflow models transition-model-version-stage \ --name "recommendation-ranker" \ --version 12 \ --stage "Production"

  2. 更新Feature Store中price_rank_score的文档，注明“v2起权重+0.2”；
  3. 在Confluence中归档本次发布报告，包含：A/B测试截图、漂移检测报告、压力测试结果。
• 准入标准：MLflow UI中，recommendation-ranker的Production版本号更新为12；Feature Store文档已更新；Confluence报告已发布。
• 避坑技巧：`transition