第11章 在线服务

11.1 概述

前面所有章节构建的模型、策略和风控规则，最终都要通过一个在线 API 服务暴露给业务系统。当用户打开 App、浏览商品或即将离开时，业务系统调用这个 API，实时获取"给不给这个用户发券、发多少"的决策。

本章用 FastAPI 构建在线预测服务，实现从请求到决策的完整链路。

11.2 服务架构

sequenceDiagram
    participant Client as 业务系统
    participant API as FastAPI 服务
    participant AB as AB 分组
    participant Risk as 风控检查
    participant Model as 模型预测
    participant Opt as 面额决策

    Client->>API: POST /api/v1/decide
    API->>AB: assign_group(user_id)
    alt 对照组
        AB-->>API: group=control
        API-->>Client: should_send=false
    else 实验组
        AB-->>API: group=model/rule/explore
        API->>Risk: check(user_id, device_id, ip)
        alt 风控拦截
            Risk-->>API: blocked
            API-->>Client: should_send=false
        else 通过
            Risk-->>API: allowed
            API->>Model: predict(features)
            Model-->>API: uplift, sensitivity, gmv
            API->>Opt: select_amount
            Opt-->>API: coupon_amount
            API-->>Client: should_send=true, amount=15
        end
    end

决策流程严格按顺序执行：AB 分组 → 风控检查 → 模型预测 → 面额选择。任何一步判定"不发券"，都会短路返回，避免不必要的模型推理开销。

11.3 请求与响应模型

用 Pydantic 定义类型安全的数据模型：

<!-- src/serving/api.py -->
class UserFeatures(BaseModel):
    """单用户特征."""
    user_id: str
    device_id: str = ""
    ip: str = ""
    user_stage: str = UserStage.BROWSED_NO_CART

    # 浏览行为特征
    browse_pv_7d: float = 0
    browse_uv_items_1d: float = 0
    detail_page_view_ratio: float = 0
    sku_revisit_max: float = 0
    cart_add_count: float = 0
    search_query_count: float = 0
    browse_price_mean: float = 0
    browse_price_std: float = 0
    category_count: float = 0
    hours_since_register: float = 0
    hours_since_last_browse: float = 0

    # 用户基础特征
    device_os: str = "Android"
    device_price_tier: str = "mid"
    channel_source: str = "organic"
    city_tier: str = "2"


class CouponDecision(BaseModel):
    """发券决策结果."""
    user_id: str
    group: str = Field(description="AB实验分组")
    should_send: bool
    coupon_amount: int = 0
    reason: str = ""
    uplift_score: float = 0.0
    intent_score: float = 0.0

设计要点：

• 所有特征都有默认值，调用方可以只传部分字段
• reason 字段记录决策原因（control_group / risk_blocked / low_uplift / model_based / rule_based），便于后续分析
• uplift_score 和 intent_score 随结果返回，方便调试和监控

11.4 三个 API 端点

<!-- src/serving/api.py -->
# 健康检查
@app.get("/health", response_model=HealthResponse)
async def health_check() -> HealthResponse:
    return HealthResponse(status="ok", models_loaded=list(_models.keys()))

# 单用户决策
@app.post("/api/v1/decide", response_model=CouponDecision)
async def decide_single(user: UserFeatures) -> CouponDecision:
    return _decide(user)

# 批量决策
@app.post("/api/v1/decide/batch", response_model=BatchResponse)
async def decide_batch(req: BatchRequest) -> BatchResponse:
    decisions = [_decide(u) for u in req.users]
    return BatchResponse(decisions=decisions)

端点	方法	用途
/health	GET	健康检查，返回已加载的模型列表
/api/v1/decide	POST	单用户实时决策
/api/v1/decide/batch	POST	批量决策（离线调度用）

11.5 核心决策逻辑

_decide() 函数实现了完整的决策链：

<!-- src/serving/api.py -->
def _decide(user: UserFeatures) -> CouponDecision:
    # 1. AB 分组
    group = _ab_test.assign_group(user.user_id)

    # 2. 对照组不发券
    if group == "control":
        return CouponDecision(
            user_id=user.user_id, group=group,
            should_send=False, reason="control_group"
        )

    # 3. 风控检查
    if _anti_fraud is not None:
        allowed, fraud_reason = _anti_fraud.check(
            user_id=user.user_id,
            device_id=user.device_id,
            ip=user.ip,
        )
        if not allowed:
            return CouponDecision(
                user_id=user.user_id, group=group,
                should_send=False,
                reason=f"risk_blocked:{fraud_reason}",
            )

    # 4. 构造特征 DataFrame
    feature_dict = user.model_dump(
        exclude={"user_id", "device_id", "ip", "user_stage"}
    )
    features_df = pd.DataFrame([feature_dict])

    # 5. 规则组或模型不可用 → 规则兜底
    if group == "rule" or not _models:
        coupon_amount = STAGE_DEFAULT_AMOUNTS.get(user.user_stage, 10)
        return CouponDecision(
            user_id=user.user_id, group=group,
            should_send=True, coupon_amount=coupon_amount,
            reason="rule_based",
        )

    # 6. 模型预测
    intent_score = float(_models["intent"].predict(features_df)[0])
    uplift_score = float(_models["uplift"].predict_uplift(features_df)[0])

    # Uplift 分太低 → 不发券
    if uplift_score <= default_config.uplift_threshold:
        return CouponDecision(
            user_id=user.user_id, group=group,
            should_send=False, reason="low_uplift",
            uplift_score=uplift_score, intent_score=intent_score,
        )

    # 7. 确定面额（根据敏感度）
    sensitivity = float(_models["sensitivity"].predict(features_df)[0])
    amounts = sorted(default_config.coupon_amounts)
    idx = int(sensitivity * (len(amounts) - 1))
    coupon_amount = amounts[min(idx, len(amounts) - 1)]

    return CouponDecision(
        user_id=user.user_id, group=group,
        should_send=True, coupon_amount=coupon_amount,
        reason="model_based",
        uplift_score=uplift_score, intent_score=intent_score,
    )

面额选择策略

在线服务中的面额选择采用简化版逻辑：根据价格敏感度在面额集合中索引。

sensitivity ∈ [0, 1]
amounts = [5, 10, 15, 20, 30, 50]

idx = int(sensitivity × 5)
  sensitivity=0.0 → idx=0 → ¥5
  sensitivity=0.3 → idx=1 → ¥10
  sensitivity=0.5 → idx=2 → ¥15
  sensitivity=0.8 → idx=4 → ¥30
  sensitivity=1.0 → idx=5 → ¥50

在线 vs 离线面额选择

离线分配（第 8 章）用完整的预算约束优化（贪心/拉格朗日），考虑全局预算限制。在线服务由于实时性要求，采用简化的敏感度索引。实际生产中，更常见的做法是离线算好每个用户的面额，在线直接查表。

11.6 模型加载与生命周期

<!-- src/serving/api.py -->
@app.on_event("startup")
async def load_models() -> None:
    global _models, _optimizer, _ab_test, _anti_fraud

    model_dir = Path("models")

    if (model_dir / "intent_model.pkl").exists():
        _models["intent"] = IntentModel.load(
            str(model_dir / "intent_model.pkl")
        )

    if (model_dir / "uplift_model.pkl").exists():
        _models["uplift"] = UpliftModel.load(
            str(model_dir / "uplift_model.pkl")
        )

    # sensitivity_model, gmv_model 类似...

    _optimizer = BudgetOptimizer(config)
    _ab_test = ABTestFramework(config)
    _anti_fraud = AntiFraudChecker(config)

关键设计：

1. 启动时加载：模型在服务启动时一次性加载到内存，避免每次请求都读磁盘
2. 容错加载：每个模型独立判断 .pkl 文件是否存在。模型缺失时不会崩溃，而是降级到规则策略
3. 全局状态：_models 字典存放已加载的模型，/health 端点可以查看加载状态

11.7 启动服务

# 启动开发服务器
uvicorn src.serving.api:app --reload --port 8000

# 测试健康检查
curl http://localhost:8000/health
# {"status":"ok","models_loaded":["intent","uplift","sensitivity","gmv"]}

# 测试单用户决策
curl -X POST http://localhost:8000/api/v1/decide \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u_000001",
    "browse_pv_7d": 25,
    "detail_page_view_ratio": 0.6,
    "cart_add_count": 2,
    "hours_since_register": 48
  }'

本章小结

• 在线服务用 FastAPI 实现，提供 /health、/api/v1/decide、/api/v1/decide/batch 三个端点
• 决策链路：AB 分组 → 风控检查 → 模型预测 → 面额选择，任何一步可短路返回
• Pydantic 保证请求/响应的类型安全，所有特征有默认值
• 模型启动时一次性加载，缺失时自动降级到规则策略
• 面额选择采用简化的敏感度索引，生产环境可升级为离线预计算 + 在线查表

思考题

1. 当前的批量接口是串行处理每个用户。如果要支持 1000 个用户的批量请求，如何优化？
2. 模型更新时需要重启服务吗？如何实现热加载（不停服更新模型）？
3. 在线服务的面额选择和离线预算优化的面额选择可能不一致。如何解决这个矛盾？