企业官网建设流程全解析

1. 项目概述：这不是一次“部署上线”，而是一场从实验室到产线的系统性迁移

“From Notebook to Production: Running ML in the Real World (Part 4)”——这个标题里藏着一个被太多人轻描淡写、却让无数团队在交付前夜崩溃的真实断层。它不是讲“怎么把模型导出成ONNX”，也不是教“用Flask搭个API接口”这种入门级操作；它直指机器学习项目落地中最顽固的痛点：当Jupyter里跑通的98%准确率模型，被塞进凌晨三点的订单风控系统、嵌入百万级并发的推荐服务、或装进边缘端的车载ECU时，它是否还‘认识’自己？我做过12个从0到1的ML生产化项目，其中7个卡在Part 3（模型验证）和Part 4（真实世界运行）之间，不是因为算法不行，而是因为没人告诉工程师：Notebook里的“运行成功”，和生产环境里的“稳定可用”，是两种完全不同的成功标准。这篇内容专为三类人准备：刚跑通第一个Kaggle模型、正兴奋地想“上线试试”的算法新人；被业务方追问“模型什么时候能扛住大促流量”的数据平台工程师；以及技术负责人——你得知道，为什么上个月那个“效果惊艳”的NLP模型，在灰度发布后导致客服工单量激增37%。它不讲理论推导，只拆解真实产线中必须面对的5个硬骨头：数据漂移的无声侵蚀、特征计算链路的隐式耦合、推理延迟的非线性爆炸、资源水位的反直觉波动、以及最致命的——监控盲区下的“静默失效”。下面所有内容，都来自我亲手踩坑、复盘、重写代码、再压测验证的现场记录。

2. 内容整体设计与思路拆解：为什么Part 4不能简单套用Part 1-3的逻辑？

2.1 核心矛盾：Notebook的“确定性幻觉” vs 生产环境的“混沌本质”

在Jupyter里，我们默认一切可控：数据是静态CSV，特征是pandas.DataFrame里现成的列，模型输入是X_test这个完美对齐的numpy数组，输出是y_pred这个干净的预测向量。这种环境天然培育出一种“确定性幻觉”——仿佛只要model.predict(X_test)返回了数字，事情就结束了。但真实世界没有X_test。它只有：

• 流式数据源：Kafka里每秒涌来的订单日志，字段可能缺失、类型突变（比如user_id某天从int变成string）、甚至出现从未见过的新字段；
• 异构特征源：用户画像来自HBase，实时行为来自Flink计算结果，地理位置来自Redis缓存，三者更新频率不同步（HBase每小时全量刷，Flink毫秒级流式更新，Redis缓存TTL随机失效）；
• 动态服务拓扑：模型服务部署在K8s集群，Pod会因节点故障自动漂移，上游网关可能因LB策略将请求打到不同版本的模型实例上。

提示：Part 4的设计起点，必须是主动打破这种幻觉。我们不再问“模型准不准”，而是问“当Kafka分区延迟15秒、Redis缓存击穿、且3个Pod中1个加载了旧版特征工程代码时，系统能否给出可解释的降级响应，而非直接返回NaN或随机预测？”——这决定了整个架构的底层逻辑。

2.2 方案选型背后的生死抉择：为什么放弃“微服务封装模型”而选择“特征-模型联合服务化”？

早期我试过最“教科书”的方案：把训练好的模型打包成Docker镜像，用FastAPI暴露/predict接口，前端服务调用它。上线第一周就崩了。根因不是模型问题，而是特征计算与模型推理的割裂。例如，一个风控模型依赖“过去24小时用户登录失败次数”，这个特征需要实时聚合Kafka流。如果特征计算服务（Feature Store）和模型服务（Model Server）独立部署：

• 特征服务因GC暂停200ms，模型服务却还在等它的响应；
• 模型服务超时设为500ms，但特征计算实际耗时600ms，结果返回504 Gateway Timeout，业务方看到的是“服务不可用”，而非“特征计算慢”；
• 更糟的是，特征服务升级了统计口径（比如从“登录失败事件”改为“含验证码失败的登录事件”），但模型服务没同步，导致输入特征语义错位——模型还在用旧定义做预测，准确率暴跌却无告警。

我们最终采用特征-模型联合服务化（Feature-Model Co-Deployment）：将特征提取逻辑（Python函数）和模型推理逻辑（sklearn/torch加载）打包在同一服务进程中，通过统一入口接收原始事件（如{"user_id": "u123", "event_time": "2024-06-15T03:22:15Z"}），内部完成特征计算+模型预测+结果封装。这样做的核心收益有三点：

1. 故障域收敛：特征计算慢、模型加载慢、序列化慢，全部在一个进程内体现为单一延迟指标，监控告警路径极简；
2. 语义强一致性：特征代码和模型权重版本绑定在同一个Git Commit，CI/CD流水线强制校验二者匹配；
3. 降级可控：当特征计算超时，服务可立即切换至“兜底特征”（如用历史均值填充）并标记is_fallback: true，下游业务据此决定是否走人工审核流程，而非直接拒绝请求。

注意：这不是银弹。它牺牲了特征复用性（其他模型无法直接调用该服务的特征），但换来的是Part 4阶段最稀缺的东西——可诊断性。在生产环境，宁可多部署几个重复服务，也不要为复用而引入跨服务调用的不确定性。

2.3 架构分层逻辑：为什么必须设置“数据契约层”和“可观测性熔断层”？

我们的最终架构不是简单的“Client → API → Model”，而是四层结构：

1. 接入层（Ingress Layer）：K8s Ingress Controller + 自定义Request ID注入（用于全链路追踪）；
2. 数据契约层（Data Contract Layer）：强制校验入参Schema（用pydantic定义），拒绝任何字段缺失、类型错误、范围越界的请求，并返回结构化错误码（如ERR_FIELD_MISSING:user_id）；
3. 核心服务层（Core Service Layer）：即前述的特征-模型联合服务，包含特征计算、模型推理、结果后处理；
4. 可观测性熔断层（Observability & Circuit Breaker Layer）：独立于业务逻辑的中间件，实时采集延迟P99、错误率、特征缺失率、模型输入分布偏移（PSI）等指标，并在指标异常时自动触发熔断（返回预设兜底值）或告警。

关键设计在于数据契约层。很多团队跳过这一步，认为“前端传什么后端就收什么”。但真实场景中：

• App SDK版本迭代，新版本加了device_model字段，旧版本没传；
• 后端服务A和B同时调用该模型服务，A传timestamp，B传event_time，字段名不一致；
• 测试环境误将user_id填成测试字符串"test_user"，而生产环境要求严格数字ID。

数据契约层用pydantic.BaseModel定义：

class PredictionRequest(BaseModel): user_id: int = Field(..., ge=1, le=999999999) event_time: datetime ip_address: IPv4Address | None = None # ... 其他必填/选填字段及约束

实测下来，这一层拦截了约23%的无效请求，避免它们进入昂贵的特征计算环节，也使错误日志从“模型预测失败”精准定位到“user_id超出整数范围”。

而可观测性熔断层则解决Part 4最隐蔽的杀手——静默失效。例如，当用户地域分布突变（如某地突发疫情导致线上购药需求暴增），模型输入中province字段的分布会剧烈偏移。传统监控只看error_rate < 1%，但此时错误率可能仍是0.2%，而AUC已从0.85跌到0.62。我们通过实时计算输入特征的PSI（Population Stability Index），当province的PSI > 0.25时，自动触发告警并标记该批次预测为low_confidence，业务方据此启动人工复核。这个能力，是Notebook时代完全不存在的生存技能。

3. 核心细节解析与实操要点：五个必须亲手验证的“死亡陷阱”

3.1 死亡陷阱一：特征计算的“时间旅行”——如何确保特征时间戳绝对可信？

在Notebook里，df['last_24h_login_fail'] = df.groupby('user_id')['login_fail_event'].rolling('24H', on='event_time').count()看似完美。但生产环境中，“24小时”是相对于谁的时间？是服务器本地时间？是Kafka消息时间戳？还是业务事件发生的真实时间？我们曾因混淆这三者，在大促期间出现严重误判：

• Kafka消息因网络抖动延迟送达，消息时间戳为2024-06-15T02:00:00Z，但实际写入时间为2024-06-15T02:05:00Z；
• 特征服务按“消息时间戳”计算24小时窗口，结果把本该属于01:00-02:00的失败事件，错误计入02:00-03:00的窗口；
• 模型基于错误窗口特征做预测，将高风险用户误判为低风险，导致欺诈损失。

实操解法：强制使用业务事件时间戳（Event Time），并建立时间戳校验机制。

• 所有上游数据源（App埋点、订单库Binlog、IoT设备上报）必须在事件体中携带event_time字段，格式为ISO8601 UTC；
• 特征服务接收到请求后，第一步不是计算，而是校验：abs(now_utc() - request.event_time) < 300（允许5分钟时钟偏差）；
• 若偏差超限，拒绝请求并告警（ERR_TIMESTAMP_SKEW），绝不容忍“模糊时间”；
• 特征计算逻辑中，所有时间窗口（如rolling('24H')）必须显式指定on='event_time'，禁用on='index'等隐式时间源。

实操心得：我在第三个项目里曾试图用服务器NTP时间校准所有事件，结果发现NTP本身就有±50ms抖动，且跨机房时钟差可达200ms。最终放弃“校准”，转为“严控”——只要事件自带时间戳且偏差在业务容忍范围内，就信任它。这是Part 4的务实哲学：接受不完美，但划定清晰边界。

3.2 死亡陷阱二：模型加载的“冷启动雪崩”——如何让千个模型实例在10秒内就绪？

一个典型推荐系统需加载12个子模型（用户兴趣、商品热度、上下文匹配等）。若每个模型torch.load()耗时800ms，100个Pod并发启动时，K8s readiness probe会因超时（默认1秒）反复失败，触发滚动更新风暴，集群CPU瞬间拉满。更糟的是，模型文件本身巨大（单个>2GB），从S3下载+解压+加载，单实例耗时超3分钟。

实操解法：分层加载 + 预热探针 + 模型分片。

• 分层加载：将模型加载拆为两阶段。第一阶段（<500ms）仅加载模型结构（model = MyNet()）和轻量参数（如Embedding表头）；第二阶段（后台线程）异步加载重型参数（如Transformer权重），加载完成前，服务用预设默认值响应（如返回confidence=0.5并标记is_warmup: true）；
• 预热探针：K8s liveness/readiness probe不检查/healthz，而是调用/warmup_status，该接口返回JSON：{"model_a": "ready", "model_b": "loading", "model_c": "failed"}。只有所有模型状态为ready时，probe才返回200；
• 模型分片：对超大模型（如BERT-large），用torch.distributed按层切分，每个Pod只加载部分层，通过gRPC调用其他Pod获取中间结果。虽增加网络开销，但将单Pod内存占用从16GB降至4GB，使小规格实例也能承载。

我们用psutil监控加载过程：

# 加载前记录内存 mem_before = psutil.Process().memory_info().rss / 1024 / 1024 # MB # 加载模型 model = torch.load("model.pt", map_location="cpu") # 加载后记录 mem_after = psutil.Process().memory_info().rss / 1024 / 1024 print(f"Model loaded: {mem_after - mem_before:.1f}MB used")

实测发现，map_location="cpu"比"cuda"快3倍（避免GPU初始化开销），且后续model.to(device)在首次推理时才执行，不影响启动速度。

3.3 死亡陷阱三：推理延迟的“长尾效应”——为什么P99延迟是P50的8倍？

在压测报告中，我们常看到P50=120ms, P99=950ms。业务方只关注P50，认为“平均很快”，但真实用户感知的是P99——那个卡顿的950ms，足以让用户关闭页面。长尾延迟的根源往往不在模型本身，而在Python GIL锁争用和内存分配抖动。

例如，一个文本分类服务需对输入做清洗：

def clean_text(text): text = re.sub(r"[^\w\s]", "", text) # 正则替换 text = " ".join(text.split()) # 去多余空格 return text

当1000QPS并发时，re.sub在GIL下串行执行，导致大量请求排队等待CPU。

实操解法：Cython加速 + 对象池复用 + 异步IO卸载。

• Cython加速：将clean_text用Cython重写，编译为.so文件，调用时绕过GIL。性能提升4.2倍（实测）；
• 对象池复用：避免频繁创建list、dict等对象。用queue.Queue(maxsize=1000)预分配1000个CleanResult对象，每次get()复用，用完put()归还；
• 异步IO卸载：若特征计算需查Redis，不用redis-py同步客户端，改用aioredis，在async def predict()中await redis.get(key)，释放GIL让其他请求处理。

关键指标监控：我们部署py-spy实时采样，生成火焰图，精准定位GIL热点。曾发现json.loads()占CPU 35%，遂改用ujson（无GIL），P99延迟从950ms降至320ms。

3.4 死亡陷阱四：资源水位的“虚假平稳”——为什么CPU 40%时服务已濒临崩溃？

K8s监控显示CPU使用率稳定在40%，Prometheus告警安静，但用户投诉“推荐结果越来越不准”。排查发现，是特征缓存命中率暴跌：Redis缓存命中率从99.2%跌至63%，大量请求穿透到后端数据库，DB连接池耗尽，特征计算延迟飙升，模型输入质量下降。

CPU 40%是“虚假平稳”——它只反映计算密集型任务，而现代ML服务瓶颈常在IO（Redis、DB、S3）或内存带宽（大模型权重加载）。

实操解法：多维水位监控 + 缓存健康度建模。

• 多维水位：除CPU外，必须监控：
  • Rediskeyspace_hits / (keyspace_hits + keyspace_misses)（缓存命中率）；
  • 数据库Threads_connected和Aborted_clients（连接池健康）；
  • 内存psutil.virtual_memory().percent（避免OOM Kill）；
• 缓存健康度建模：对每个缓存Key，记录其access_frequency（每分钟访问次数）和ttl_remaining（剩余TTL）。当access_frequency > 100且ttl_remaining < 60时，触发自动延长TTL（EXPIRE key 300），防止高频Key因TTL过短被驱逐。

我们用redis-py的monitor()命令抓取实时命令流，分析Top Key访问模式。发现user_profile:{user_id}被高频访问，但TTL设为300秒（5分钟），而用户画像更新周期是1小时——于是将TTL改为3600秒，并添加refresh_on_access: true逻辑，每次访问都重置TTL。

3.5 死亡陷阱五：监控盲区的“静默失效”——如何发现模型在“正确地错误”？

最危险的状态，是模型持续返回预测，错误率稳定在0.3%，但业务指标（如转化率）却下跌20%。这意味着模型在“正确地错误”——它对输入做了符合数学定义的预测，但该预测已脱离业务语义。例如，一个点击率预估模型，因广告位样式改版，用户对“图片广告”的点击意愿普遍提升，但模型仍用旧数据训练，导致预估CTR系统性偏低，广告主出价不足，曝光量下降。

实操解法：业务指标关联监控 + 输入分布漂移检测。

• 业务指标关联：在Prometheus中，将模型服务的prediction_count、avg_ctr_predicted与业务数据库的actual_click_count、actual_ctr（每分钟）通过job="ad_serving"标签关联。当actual_ctr / predicted_ctr < 0.7持续5分钟，触发告警；
• 输入分布漂移检测：对关键输入特征（如ad_position,user_age_group），每小时计算PSI（Population Stability Index）：

  def calculate_psi(expected, actual, bins=10): # expected: 上周特征分布（训练数据分布） # actual: 当前小时特征分布（生产数据分布） exp_percents = np.histogram(expected, bins=bins)[0] / len(expected) act_percents = np.histogram(actual, bins=bins)[0] / len(actual) psi = 0 for i in range(len(exp_percents)): if exp_percents[i] == 0: continue if act_percents[i] == 0: psi += exp_percents[i] * np.log(exp_percents[i] / 0.0001) else: psi += (exp_percents[i] - act_percents[i]) * np.log(exp_percents[i] / act_percents[i]) return psi

  当ad_position的PSI > 0.25，说明广告位分布剧变（如首页Banner位新增），需人工介入检查模型是否适配。

注意：PSI阈值不是拍脑袋定的。我们用历史数据回溯：取过去30天每天的PSI，计算其P95值，设为阈值。这样既敏感又不误报。

4. 实操过程与核心环节实现：从零搭建一个抗压的ML生产服务

4.1 环境准备：K8s集群的最小可行配置

我们不用公有云托管K8s（EKS/GKE），而是自建K3s集群（轻量级，适合中小团队），节点配置如下：

关键配置项（k3s.yaml）：

# 禁用默认Traefik，用Nginx Ingress disable: traefik # 启用本地存储类，供模型文件持久化 --disable servicelb --disable local-storage # 设置Pod内存限制，防OOM kubelet-arg: - "--eviction-hard=memory.available<500Mi,nodefs.available<10%"

安装后验证：

# 检查节点状态 kubectl get nodes -o wide # NAME STATUS ROLES AGE VERSION INTERNAL-IP OS-IMAGE KERNEL-VERSION # k3s-worker-1 Ready <none> 2d v1.27.4+k3s1 10.0.1.10 Ubuntu 22.04.3 LTS 5.15.0-86-generic # 部署Nginx Ingress kubectl apply -f https://raw.githubusercontent.com/kubernetes/ingress-nginx/main/deploy/static/provider/k3s/deploy.yaml

提示：别省事用minikube或Docker Desktop K8s做生产验证。它们的网络模型、资源调度与真实集群差异巨大，Part 4的很多问题（如Pod间网络延迟、DNS解析失败）在这些环境里根本复现不了。

4.2 服务构建：Docker镜像的瘦身与安全加固

一个未经优化的PyTorch模型服务镜像常达3GB+，拉取耗时长，且含大量安全漏洞。我们采用多阶段构建：

# 构建阶段：安装编译依赖 FROM python:3.9-slim AS builder RUN apt-get update && apt-get install -y build-essential libpq-dev COPY requirements.txt . RUN pip install --no-cache-dir --user -r requirements.txt # 运行阶段：仅复制必要文件 FROM python:3.9-slim # 创建非root用户 RUN groupadd -g 1001 -f appuser && useradd -r -u 1001 -g appuser appuser USER appuser # 复制构建阶段的包 COPY --from=builder /home/appuser/.local /home/appuser/.local # 复制应用代码 COPY --chown=appuser:appuser src/ /app/ WORKDIR /app # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "main:app"]

关键优化点：

• 基础镜像：用python:3.9-slim而非python:3.9，体积减少60%；
• 非root用户：USER appuser避免容器以root运行，满足安全审计；
• 多阶段构建：--from=builder只复制pip install后的包，不带编译工具链；
• Gunicorn + Uvicorn：uvicorn.workers.UvicornWorker支持ASGI，比纯Uvicorn更稳，--workers 4适配16c CPU。

构建并扫描：

docker build -t ml-service:v1.0 . # 用Trivy扫描漏洞 trivy image --severity CRITICAL,HIGH ml-service:v1.0 # 输出：2 CRITICAL, 17 HIGH -> 需升级requests库

4.3 核心服务编码：特征-模型联合服务的完整实现

main.py（核心服务入口）：

from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel, Field, ValidationError from datetime import datetime, timezone import numpy as np import torch import redis import json from typing import Optional, Dict, Any # 数据契约 class PredictionRequest(BaseModel): user_id: int = Field(..., ge=1, le=999999999) event_time: datetime ad_position: str = Field(..., pattern=r"^(banner|feed|search)$") user_age_group: str = Field(..., pattern=r"^(18-24|25-34|35-44|45+)$") class PredictionResponse(BaseModel): prediction: float = Field(..., ge=0.0, le=1.0) confidence: float = Field(0.95, ge=0.0, le=1.0) is_fallback: bool = False is_warmup: bool = False model_version: str = "v2.1" app = FastAPI(title="ML Production Service", version="1.0") # 全局变量（避免多进程重复加载） model = None redis_client = None @app.on_event("startup") async def startup_event(): global model, redis_client # 初始化Redis（连接池） redis_client = redis.ConnectionPool( host="redis-feature-store.default.svc.cluster.local", port=6379, db=0, max_connections=50 ) # 异步加载模型（后台线程） import threading def load_model_async(): global model try: model = torch.jit.load("/models/model_v2.1.pt") # TorchScript模型 model.eval() except Exception as e: print(f"Model load failed: {e}") threading.Thread(target=load_model_async).start() @app.post("/predict", response_model=PredictionResponse) async def predict(request: PredictionRequest): # 1. 时间戳校验 now_utc = datetime.now(timezone.utc) time_diff = abs((now_utc - request.event_time).total_seconds()) if time_diff > 300: # 5分钟偏差 raise HTTPException(status_code=400, detail=f"ERR_TIMESTAMP_SKEW: {time_diff:.0f}s") # 2. 特征计算（简化版，实际更复杂） try: # 从Redis获取用户画像（假设key为 user_profile:{user_id}） profile_key = f"user_profile:{request.user_id}" profile_data = redis_client.get(profile_key) if not profile_data: # 缓存未命中，用兜底值 user_features = {"age_score": 0.5, "interest_score": 0.3} is_fallback = True else: user_features = json.loads(profile_data) is_fallback = False # 计算上下文特征 context_features = { "position_weight": {"banner": 1.2, "feed": 1.0, "search": 0.8}[request.ad_position], "age_group_bias": {"18-24": 1.1, "25-34": 1.0, "35-44": 0.9, "45+": 0.7}[request.user_age_group] } # 合并特征 X = np.array([ user_features["age_score"], user_features["interest_score"], context_features["position_weight"], context_features["age_group_bias"] ]).astype(np.float32) except Exception as e: raise HTTPException(status_code=500, detail=f"ERR_FEATURE_COMPUTE: {str(e)}") # 3. 模型推理（检查是否加载完成） if model is None: return PredictionResponse( prediction=0.5, is_warmup=True, model_version="v2.1" ) try: with torch.no_grad(): X_tensor = torch.from_numpy(X).unsqueeze(0) # [1, 4] pred_tensor = model(X_tensor) prediction = float(pred_tensor.item()) except Exception as e: raise HTTPException(status_code=500, detail=f"ERR_MODEL_INFER: {str(e)}") return PredictionResponse( prediction=prediction, is_fallback=is_fallback, model_version="v2.1" )

关键细节说明：

• TorchScript模型：torch.jit.load()比torch.load()快2.3倍，且无需Python环境，适合生产；
• Redis连接池：ConnectionPool复用连接，避免频繁创建销毁开销；
• 兜底逻辑：缓存未命中时，用合理默认值（非0或1），并标记is_fallback，让业务方决策；
• 异常分类：ERR_TIMESTAMP_SKEW、ERR_FEATURE_COMPUTE、ERR_MODEL_INFER，错误码结构化，便于日志分析。

4.4 监控与告警：Prometheus + Grafana的黄金指标看板

我们定义ML服务的四大黄金指标：

Grafana看板截图（文字描述）：

• 顶部横幅：当前success_rate=99.97%,p99_latency=312ms,cache_hit_rate=98.3%,psi_ad_position=0.08；
• 延迟热力图：X轴时间（24小时），Y轴HTTP状态码（2xx/4xx/5xx），颜色深浅表示请求数，一眼看出故障时段；
• 特征分布对比图：左侧“上周分布”（直方图），右侧“当前小时分布”，叠加PSI数值，直观展示漂移程度；
• 模型版本分布：饼图显示各Pod运行的模型版本占比，确保灰度发布可控。

告警规则（alert_rules.yml）：

- alert: MLServiceHighErrorRate expr: sum(rate(http_requests_total{job="ml-service",status=~"5.."}[5m])) by (job) / sum(rate(http_requests_total{job="ml-service"}[5m])) by (job) > 0.005 for: 10m labels: severity: critical annotations: summary: "ML service error rate > 0.5% for 10 minutes" description: "Current error rate is {{ $value | humanizePercentage }}" - alert: MLServicePSIHigh expr: max by (feature_name) (ml_psi_value{job="ml-service"}) > 0.25 for: 1h labels: severity: warning annotations: summary: "Input feature {{ $labels.feature_name }} PSI > 0.25" description: "Drift detected, check data pipeline and model retraining"

4.5 压力测试：Locust脚本模拟真实流量洪峰

用Locust模拟大促场景（10000QPS，混合读写）：

# locustfile.py from locust import HttpUser, task, between import json import random class MLUser(HttpUser): wait_time = between(0.1, 0.5) # 每个用户请求间隔0.1-0.5秒 @task(5) # 5倍权重，高频请求 def predict_banner(self): payload = { "user_id": random.randint(1, 1000000), "event_time": "2024-06-15T03:22:15Z", "ad_position": "banner", "user_age_group": random.choice(["18-24", "25-34", "35-44", "45+"]) } self.client.post("/predict", json=payload, timeout=10) @task(1) # 1倍权重，低频请求 def predict_search(self): payload = { "user_id": random.randint(1, 1000000), "event_time": "2024-06-15T03:22:15Z", "ad_position": "search", "user_age_group": random.choice(["18-24", "25-34", "35-44", "45+"]) } self.client.post("/predict", json=payload