用户 ID 末两位 00-04（约 5%）→ 新模型
用户 ID 末两位 05-99（约 95%）→ 旧模型

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

from scipy import stats
import math

def calculate_sample_size(
    baseline_rate: float,   # 基准指标，如 0.75 表示 75% 满意率
    mde: float,             # 最小可检测效果，如 0.05 表示提升 5 个百分点
    alpha: float = 0.05,    # 显著性水平（误判概率），通常取 0.05
    power: float = 0.80     # 统计功效（发现真实效果的概率），通常取 0.80
) -> int:
    """计算 A/B 测试所需的每组样本量"""
    effect_size = abs(mde) / math.sqrt(
        (baseline_rate * (1 - baseline_rate) +
         (baseline_rate + mde) * (1 - baseline_rate - mde)) / 2
    )
    z_alpha = stats.norm.ppf(1 - alpha / 2)   # 双尾检验
    z_beta = stats.norm.ppf(power)
    n = ((z_alpha + z_beta) / effect_size) ** 2
    return math.ceil(n)

# 示例：基准满意率 75%，希望检测到 5 个百分点的提升
n = calculate_sample_size(baseline_rate=0.75, mde=0.05)
print(f"每组需要约 {n} 个样本（共 {n*2} 个）")
# 输出：每组需要约 1264 个样本（共 2528 个）

# 如果每天有 500 次请求，至少需要 2528/500 ≈ 5 天才能完成测试

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

import hashlib
import time
from typing import Literal
from dataclasses import dataclass
import redis

@dataclass
class ABConfig:
    """A/B 测试配置，存储在 Redis 中，可动态修改"""
    experiment_id: str           # 实验 ID，如 "model_upgrade_v2"
    model_a: str                 # 控制组（旧模型）
    model_b: str                 # 实验组（新模型）
    traffic_to_b: float          # 分配给 B 组的流量比例，如 0.05 = 5%
    enabled: bool = True

class ModelABRouter:
    """
    模型 A/B 路由器
    核心逻辑：根据用户 ID 的哈希值决定分组，保证同一用户始终在同一组
    """

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def get_model_for_user(
        self,
        user_id: str,
        experiment_id: str = "model_upgrade"
    ) -> tuple[str, Literal["A", "B"]]:
        """
        根据用户 ID 返回应使用的模型
        返回：(模型名称, 分组标记)
        """
        config = self._get_config(experiment_id)
        if not config.enabled:
            return config.model_a, "A"

        # 用 user_id + experiment_id 计算哈希值
        # 这保证：同一用户在同一实验中始终被分配到同一组
        hash_input = f"{user_id}:{experiment_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)

        # 哈希值对 100 取模，得到 0-99 的均匀分布
        bucket = hash_value % 100

        if bucket < config.traffic_to_b * 100:
            return config.model_b, "B"
        else:
            return config.model_a, "A"

    def _get_config(self, experiment_id: str) -> ABConfig:
        """从 Redis 读取实验配置（支持运行时动态修改）"""
        key = f"ab:config:{experiment_id}"
        data = self.redis.hgetall(key)
        if not data:
            # 默认配置：100% 流量给旧模型
            return ABConfig(
                experiment_id=experiment_id,
                model_a="gpt-4o",
                model_b="gpt-4.1",
                traffic_to_b=0.0,
                enabled=False
            )
        return ABConfig(
            experiment_id=experiment_id,
            model_a=data[b"model_a"].decode(),
            model_b=data[b"model_b"].decode(),
            traffic_to_b=float(data[b"traffic_to_b"]),
            enabled=data[b"enabled"].decode() == "true"
        )

# 在 FastAPI 路由中使用
from fastapi import FastAPI, Depends
from openai import AsyncOpenAI

app = FastAPI()
openai_client = AsyncOpenAI()

@app.post("/chat")
async def chat(request: ChatRequest, router: ModelABRouter = Depends(get_router)):
    # 路由决策
    model_name, group = router.get_model_for_user(
        user_id=request.user_id,
        experiment_id="model_upgrade_v2"
    )

    # 记录分组信息，用于后续统计分析
    start_time = time.time()
    response = await openai_client.chat.completions.create(
        model=model_name,
        messages=request.messages
    )
    latency_ms = (time.time() - start_time) * 1000

    # 埋点：记录本次请求的实验数据
    await record_ab_event(
        user_id=request.user_id,
        experiment_id="model_upgrade_v2",
        group=group,
        model=model_name,
        latency_ms=latency_ms,
        prompt_tokens=response.usage.prompt_tokens,
        completion_tokens=response.usage.completion_tokens
    )

    return {"answer": response.choices[0].message.content, "group": group}

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

from dataclasses import dataclass
import redis
import logging

@dataclass
class RollbackTrigger:
    """自动回滚触发条件配置"""
    error_rate_threshold: float = 0.05      # 错误率超过 5% 触发回滚
    latency_p99_threshold_ms: float = 30000  # P99 延迟超过 30 秒触发回滚
    satisfaction_drop_threshold: float = 0.1 # 满意率下降超过 10 个百分点

class AutoRollbackMonitor:
    """
    自动回滚监控器
    每分钟检查一次关键指标，触发阈值时自动回滚
    """

    def __init__(self, redis_client: redis.Redis, trigger: RollbackTrigger):
        self.redis = redis_client
        self.trigger = trigger

    async def check_and_rollback(self, experiment_id: str) -> bool:
        """
        返回 True 表示触发了回滚
        """
        stats_b = await self._get_stats("B", experiment_id)
        stats_a = await self._get_stats("A", experiment_id)

        # 检查 B 组的错误率
        if stats_b.error_rate > self.trigger.error_rate_threshold:
            await self._do_rollback(
                experiment_id,
                reason=f"B组错误率 {stats_b.error_rate:.1%} 超过阈值 {self.trigger.error_rate_threshold:.1%}"
            )
            return True

        # 检查 B 组的 P99 延迟
        if stats_b.latency_p99 > self.trigger.latency_p99_threshold_ms:
            await self._do_rollback(
                experiment_id,
                reason=f"B组P99延迟 {stats_b.latency_p99:.0f}ms 超过阈值"
            )
            return True

        # 检查满意率是否显著下降（与 A 组对比）
        satisfaction_drop = stats_a.satisfaction_rate - stats_b.satisfaction_rate
        if satisfaction_drop > self.trigger.satisfaction_drop_threshold:
            await self._do_rollback(
                experiment_id,
                reason=f"B组满意率比A组低 {satisfaction_drop:.1%}"
            )
            return True

        return False

    async def _do_rollback(self, experiment_id: str, reason: str):
        """执行回滚：将 B 组流量设为 0，发送告警"""
        key = f"ab:config:{experiment_id}"
        self.redis.hset(key, "traffic_to_b", 0)
        self.redis.hset(key, "enabled", "false")
        logging.critical(f"[AutoRollback] 实验 {experiment_id} 已回滚。原因: {reason}")
        # 触发告警通知（接入第11章的告警系统）
        await send_alert(
            level="critical",
            title=f"AI 模型灰度发布自动回滚",
            message=reason
        )

# 以下为命令行示例，非程序员可跳过，重点看说明
redis-cli HSET ab:config:gpt41_upgrade \
  model_a "gpt-4o" \
  model_b "gpt-4.1" \
  traffic_to_b 0.0 \
  enabled false

redis-cli HSET ab:config:gpt41_upgrade traffic_to_b 0.05 enabled true

# 全量发布后，保留旧配置 30 天用于紧急回滚
redis-cli HSET ab:config:gpt41_upgrade traffic_to_b 1.0
# 30 天后确认稳定，归档实验配置