AI应用A/B测试与增长实验完全教程

本文系统讲解如何在AI应用中设计和实施A/B测试，涵盖统计学原理、模型版本实验、多臂老虎机算法及完整的增长实验框架。

目录

1. A/B测试基础与统计学原理
2. 样本量计算与实验周期
3. AI应用A/B测试的特殊性
4. Feature Flag管理
5. 多臂老虎机算法
6. 增长实验设计
7. 数据收集与分析
8. 显著性检验与结果解读
9. 持续迭代框架
10. 常用A/B测试工具对比
11. 总结与最佳实践

1. A/B测试基础与统计学原理

1.1 什么是A/B测试

A/B测试是一种随机对照实验，将用户随机分配到不同组（对照组A和实验组B），通过比较各组的关键指标来判断变更是否有效。在AI应用中，A/B测试常用于：

• 模型版本对比：新模型 vs 旧模型
• Prompt优化：不同Prompt模板的效果对比
• UI/UX变更：推荐算法、排序策略、界面布局
• 功能开关：新功能上线前的效果验证

1.2 核心统计概念

假设检验

A/B测试的统计基础是假设检验：

• 原假设（H₀）：实验组与对照组没有差异（新方案无效）
• 备择假设（H₁）：实验组与对照组存在差异（新方案有效）

import numpy as np
from scipy import stats

def two_sample_z_test(
    n_control: int,      # 对照组样本量
    n_treatment: int,    # 实验组样本量
    x_control: int,      # 对照组成功数
    x_treatment: int,    # 实验组成功数
    alpha: float = 0.05  # 显著性水平
) -> dict:
    """
    双样本Z检验（适用于比例型指标，如点击率、转化率）
    """
    p_control = x_control / n_control
    p_treatment = x_treatment / n_treatment

    # 合并比例
    p_pool = (x_control + x_treatment) / (n_control + n_treatment)

    # 标准误
    se = np.sqrt(p_pool * (1 - p_pool) * (1/n_control + 1/n_treatment))

    # Z统计量
    z_stat = (p_treatment - p_control) / se

    # p值（双尾检验）
    p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))

    # 置信区间
    se_diff = np.sqrt(
        p_control * (1 - p_control) / n_control +
        p_treatment * (1 - p_treatment) / n_treatment
    )
    ci_lower = (p_treatment - p_control) - 1.96 * se_diff
    ci_upper = (p_treatment - p_control) + 1.96 * se_diff

    return {
        "control_rate": round(p_control, 4),
        "treatment_rate": round(p_treatment, 4),
        "relative_lift": round((p_treatment - p_control) / p_control * 100, 2),
        "z_statistic": round(z_stat, 4),
        "p_value": round(p_value, 6),
        "significant": p_value < alpha,
        "ci_95": (round(ci_lower, 4), round(ci_upper, 4))
    }

# 示例：对比两个推荐算法的点击率
result = two_sample_z_test(
    n_control=10000, n_treatment=10000,
    x_control=320, x_treatment=380,
    alpha=0.05
)
print(result)
# {
#   'control_rate': 0.032, 'treatment_rate': 0.038,
#   'relative_lift': 18.75, 'z_statistic': 2.4819,
#   'p_value': 0.0131, 'significant': True,
#   'ci_95': (0.0013, 0.0107)
# }

关键统计概念

两类错误

                    真实情况
                 H₀为真    H₀为假
决策  拒绝H₀    I类错误(α)  正确决策(1-β)
      接受H₀    正确决策     II类错误(β)

• I类错误（假阳性）：新方案实际无效，但测试显示有效 → 上线无效果的功能
• II类错误（假阴性）：新方案实际有效，但测试未检测到 → 错失好方案

2. 样本量计算与实验周期

2.1 样本量公式

对于比例型指标（如转化率），所需样本量：

def calculate_sample_size(
    baseline_rate: float,      # 基线转化率
    mde: float,                # 最小可检测效应（相对提升）
    alpha: float = 0.05,       # 显著性水平
    power: float = 0.80        # 统计功效
) -> int:
    """
    计算每组所需样本量

    参数:
        baseline_rate: 当前转化率，如0.05表示5%
        mde: 期望检测到的最小相对提升，如0.1表示10%相对提升
        alpha: 显著性水平
        power: 统计功效
    """
    from scipy.stats import norm

    p1 = baseline_rate
    p2 = baseline_rate * (1 + mde)

    # Z值
    z_alpha = norm.ppf(1 - alpha / 2)  # 双尾
    z_beta = norm.ppf(power)

    # 合并比例
    p_bar = (p1 + p2) / 2

    # 样本量公式
    n = (
        (z_alpha * np.sqrt(2 * p_bar * (1 - p_bar)) +
         z_beta * np.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2
        / (p2 - p1) ** 2
    )

    return int(np.ceil(n))

# 示例：基线转化率5%，期望检测到10%的相对提升
n = calculate_sample_size(
    baseline_rate=0.05,
    mde=0.10,  # 检测5% → 5.5%的变化
    alpha=0.05,
    power=0.80
)
print(f"每组需要样本量: {n:,}")  # 约 284,000

2.2 实验周期计算

def calculate_experiment_duration(
    sample_size_per_group: int,
    daily_traffic: int,
    num_groups: int = 2,
    traffic_fraction: float = 1.0  # 流量分配比例
) -> dict:
    """
    计算实验所需天数

    参数:
        sample_size_per_group: 每组所需样本量
        daily_traffic: 日均用户数
        num_groups: 实验组数
        traffic_fraction: 参与实验的流量比例
    """
    daily_experiment_traffic = daily_traffic * traffic_fraction
    daily_per_group = daily_experiment_traffic / num_groups
    days_needed = np.ceil(sample_size_per_group / daily_per_group)

    return {
        "days_needed": int(days_needed),
        "weeks_needed": round(days_needed / 7, 1),
        "total_sample_needed": sample_size_per_group * num_groups,
        "daily_traffic_per_group": int(daily_per_group)
    }

# 示例：日活10万，每组需要28.4万样本
duration = calculate_experiment_duration(
    sample_size_per_group=284000,
    daily_traffic=100000,
    num_groups=2,
    traffic_fraction=0.5  # 只用50%流量做实验
)
print(duration)
# {'days_needed': 12, 'weeks_needed': 1.7, 'total_sample_needed': 568000, ...}

2.3 样本量速查表

注：α=0.05，Power=0.80，每组样本量

3. AI应用A/B测试的特殊性

3.1 模型版本A/B测试

AI应用的A/B测试不同于传统UI变更，有其独特挑战：

from dataclasses import dataclass
from typing import Optional
import hashlib

@dataclass
class ModelExperiment:
    """模型A/B测试配置"""
    experiment_id: str
    model_variants: dict  # variant_name -> model_config
    traffic_split: dict   # variant_name -> percentage
    primary_metric: str   # 主要评估指标
    secondary_metrics: list  # 次要指标
    min_sample_size: int
    max_duration_days: int

class ModelABRouter:
    """模型版本路由器"""

    def __init__(self, experiment: ModelExperiment):
        self.experiment = experiment
        self._validate_splits()

    def _validate_splits(self):
        total = sum(self.experiment.traffic_split.values())
        if abs(total - 100) > 0.01:
            raise ValueError(f"Traffic split must sum to 100%, got {total}%")

    def assign_variant(self, user_id: str) -> str:
        """
        基于用户ID的确定性分配
        同一用户始终看到同一版本（一致性保证）
        """
        # 使用实验ID+用户ID的哈希确保不同实验的分配独立
        hash_input = f"{self.experiment.experiment_id}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        bucket = hash_value % 100

        cumulative = 0
        for variant, percentage in self.experiment.traffic_split.items():
            cumulative += percentage
            if bucket < cumulative:
                return variant
        return list(self.experiment.traffic_split.keys())[-1]

    def get_model_config(self, user_id: str) -> dict:
        variant = self.assign_variant(user_id)
        return {
            "variant": variant,
            "config": self.experiment.model_variants[variant]
        }

# 使用示例
experiment = ModelExperiment(
    experiment_id="exp_2024_llm_prompt_v2",
    model_variants={
        "control": {
            "model": "gpt-4-turbo",
            "prompt_version": "v1",
            "temperature": 0.7
        },
        "treatment": {
            "model": "gpt-4-turbo",
            "prompt_version": "v2",
            "temperature": 0.3
        }
    },
    traffic_split={"control": 50, "treatment": 50},
    primary_metric="answer_accuracy",
    secondary_metrics=["response_time", "user_satisfaction"],
    min_sample_size=5000,
    max_duration_days=14
)

router = ModelABRouter(experiment)
print(router.get_model_config("user_12345"))
# {'variant': 'treatment', 'config': {...}}

3.2 AI实验的特殊挑战

3.3 分层实验设计

当多个实验同时运行时，需要分层（Layer）设计避免相互干扰：

class LayeredExperimentSystem:
    """
    分层实验系统
    不同层的实验独立运行，同一层的实验互斥
    """

    def __init__(self):
        self.layers = {}  # layer_name -> {exp_id -> experiment}
        self.user_assignments = {}  # user_id -> {layer -> variant}

    def register_experiment(self, layer: str, experiment: ModelExperiment):
        if layer not in self.layers:
            self.layers[layer] = {}
        self.layers[layer][experiment.experiment_id] = experiment

    def assign(self, user_id: str) -> dict:
        assignments = {}
        for layer_name, experiments in self.layers.items():
            # 每层独立哈希，确保层间独立
            for exp_id, experiment in experiments.items():
                router = ModelABRouter(experiment)
                variant = router.assign_variant(user_id)
                assignments[f"{layer_name}:{exp_id}"] = variant
        self.user_assignments[user_id] = assignments
        return assignments

# 示例：三层实验系统
system = LayeredExperimentSystem()

# 模型层（互斥：同时只能测一个模型）
system.register_experiment("model", ModelExperiment(
    experiment_id="model_v2_test",
    model_variants={"control": {"model": "v1"}, "treatment": {"model": "v2"}},
    traffic_split={"control": 50, "treatment": 50},
    primary_metric="accuracy",
    secondary_metrics=[],
    min_sample_size=10000,
    max_duration_days=7
))

# Prompt层（独立于模型层）
system.register_experiment("prompt", ModelExperiment(
    experiment_id="prompt_optimization",
    model_variants={"control": {"prompt": "v1"}, "treatment": {"prompt": "v2"}},
    traffic_split={"control": 50, "treatment": 50},
    primary_metric="user_satisfaction",
    secondary_metrics=[],
    min_sample_size=5000,
    max_duration_days=14
))

assignments = system.assign("user_12345")
print(assignments)
# {'model:model_v2_test': 'treatment', 'prompt:prompt_optimization': 'control'}

4. Feature Flag管理

4.1 Feature Flag系统设计

Feature Flag是A/B测试的技术基础，允许动态控制功能的开启/关闭和流量分配：

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
from enum import Enum
import json

class FlagType(Enum):
    BOOLEAN = "boolean"      # 开/关
    PERCENTAGE = "percentage" # 百分比放量
    VARIANT = "variant"      # 多变体

@dataclass
class FeatureFlag:
    key: str
    name: str
    flag_type: FlagType
    enabled: bool = False
    default_value: Any = False
    # 百分比放量（0-100）
    rollout_percentage: int = 0
    # 多变体配置
    variants: dict = field(default_factory=dict)
    # 用户白名单
    whitelist: set = field(default_factory=set)
    # 用户黑名单
    blacklist: set = field(default_factory=set)
    # 创建/更新时间
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

class FeatureFlagService:
    def __init__(self):
        self.flags: dict[str, FeatureFlag] = {}

    def create_flag(self, flag: FeatureFlag):
        self.flags[flag.key] = flag

    def evaluate(self, flag_key: str, user_id: str, default: Any = None) -> Any:
        flag = self.flags.get(flag_key)
        if not flag or not flag.enabled:
            return default if default is not None else (flag.default_value if flag else None)

        # 黑名单直接返回默认值
        if user_id in flag.blacklist:
            return flag.default_value

        # 白名单直接返回
        if user_id in flag.whitelist:
            if flag.flag_type == FlagType.BOOLEAN:
                return True
            if flag.flag_type == FlagType.VARIANT:
                return flag.variants.get(user_id, flag.default_value)

        # 基于哈希的百分比分流
        hash_val = int(hashlib.md5(
            f"{flag_key}:{user_id}".encode()
        ).hexdigest(), 16) % 100

        if flag.flag_type == FlagType.BOOLEAN:
            return hash_val < flag.rollout_percentage
        elif flag.flag_type == FlagType.PERCENTAGE:
            return hash_val < flag.rollout_percentage
        elif flag.flag_type == FlagType.VARIANT:
            # 按变体权重分配
            cumulative = 0
            for variant_name, weight in flag.variants.items():
                cumulative += weight
                if hash_val < cumulative:
                    return variant_name
            return flag.default_value

        return flag.default_value

# 使用示例
ff_service = FeatureFlagService()

# 创建一个百分比放量的Feature Flag
ff_service.create_flag(FeatureFlag(
    key="new_recommendation_algorithm",
    name="新推荐算法",
    flag_type=FlagType.VARIANT,
    enabled=True,
    variants={"control": 50, "treatment": 50},
    default_value="control",
    whitelist={"beta_user_001", "beta_user_002"}
))

# 评估
variant = ff_service.evaluate("new_recommendation_algorithm", "user_12345")
print(f"用户分组: {variant}")

4.2 动态配置热更新

import json
import hashlib
import threading
import time

class DynamicFeatureFlagService(FeatureFlagService):
    """支持热更新的Feature Flag服务"""

    def __init__(self, config_source: str):
        super().__init__()
        self.config_source = config_source  # 文件路径或API URL
        self.config_hash = None
        self._start_watcher()

    def _start_watcher(self):
        """后台线程定期检查配置更新"""
        def watch():
            while True:
                try:
                    self._reload_if_changed()
                except Exception as e:
                    print(f"Config reload error: {e}")
                time.sleep(10)  # 每10秒检查一次

        thread = threading.Thread(target=watch, daemon=True)
        thread.start()

    def _reload_if_changed(self):
        with open(self.config_source, 'r') as f:
            content = f.read()

        new_hash = hashlib.md5(content.encode()).hexdigest()
        if new_hash == self.config_hash:
            return

        config = json.loads(content)
        self.flags.clear()
        for flag_def in config.get("flags", []):
            flag = FeatureFlag(
                key=flag_def["key"],
                name=flag_def["name"],
                flag_type=FlagType(flag_def["type"]),
                enabled=flag_def.get("enabled", False),
                rollout_percentage=flag_def.get("rollout_percentage", 0),
                variants=flag_def.get("variants", {}),
                whitelist=set(flag_def.get("whitelist", [])),
                blacklist=set(flag_def.get("blacklist", []))
            )
            self.create_flag(flag)

        self.config_hash = new_hash
        print(f"Reloaded {len(self.flags)} feature flags")

4.3 配置文件示例

{
  "flags": [
    {
      "key": "new_chat_model",
      "name": "新版对话模型",
      "type": "variant",
      "enabled": true,
      "variants": {
        "control": 50,
        "gpt4_turbo": 30,
        "claude_3": 20
      },
      "whitelist": ["internal_tester_001"],
      "blacklist": ["vip_user_001"]
    },
    {
      "key": "streaming_response",
      "name": "流式响应",
      "type": "percentage",
      "enabled": true,
      "rollout_percentage": 20
    }
  ]
}

5. 多臂老虎机算法

5.1 传统A/B测试 vs 多臂老虎机

传统A/B测试的问题：在实验期间，即使某个变体明显更优，仍需继续分配流量给较差的变体，造成机会成本。

多臂老虎机（Multi-Armed Bandit, MAB）算法通过动态调整流量分配，最小化这种损失：

5.2 Epsilon-Greedy算法

import random
import numpy as np

class EpsilonGreedy:
    """
    Epsilon-Greedy多臂老虎机
    以epsilon的概率随机探索，1-epsilon的概率选择当前最优
    """

    def __init__(self, n_arms: int, epsilon: float = 0.1):
        self.n_arms = n_arms
        self.epsilon = epsilon
        self.counts = np.zeros(n_arms)       # 每个臂被选择的次数
        self.values = np.zeros(n_arms)       # 每个臂的平均奖励

    def select_arm(self) -> int:
        """选择一个臂"""
        if random.random() < self.epsilon:
            # 探索：随机选择
            return random.randint(0, self.n_arms - 1)
        else:
            # 利用：选择当前最优
            return int(np.argmax(self.values))

    def update(self, arm: int, reward: float):
        """更新选中臂的奖励"""
        self.counts[arm] += 1
        n = self.counts[arm]
        # 增量更新平均值
        self.values[arm] = self.values[arm] * (n - 1) / n + reward / n

    def get_stats(self) -> dict:
        return {
            "arm_counts": self.counts.tolist(),
            "arm_values": self.values.tolist(),
            "best_arm": int(np.argmax(self.values)),
            "total_pulls": int(self.counts.sum())
        }

# 模拟实验
mab = EpsilonGreedy(n_arms=3, epsilon=0.1)

# 真实转化率（未知）
true_rates = [0.05, 0.08, 0.03]

for i in range(10000):
    arm = mab.select_arm()
    reward = 1 if random.random() < true_rates[arm] else 0
    mab.update(arm, reward)

print(mab.get_stats())
# arm_1被选择次数最多，因为它的真实转化率最高

5.3 Thompson Sampling

Thompson Sampling是贝叶斯方法，通常比Epsilon-Greedy收敛更快：

import numpy as np
from scipy import stats

class ThompsonSampling:
    """
    Thompson Sampling多臂老虎机
    基于Beta分布的贝叶斯方法
    """

    def __init__(self, n_arms: int):
        self.n_arms = n_arms
        # Beta分布参数（先验：alpha=1, beta=1 即均匀分布）
        self.alpha = np.ones(n_arms)  # 成功次数 + 1
        self.beta = np.ones(n_arms)   # 失败次数 + 1

    def select_arm(self) -> int:
        """从每个臂的Beta分布中采样，选择采样值最大的臂"""
        samples = np.array([
            np.random.beta(self.alpha[i], self.beta[i])
            for i in range(self.n_arms)
        ])
        return int(np.argmax(samples))

    def update(self, arm: int, reward: float):
        """更新Beta分布参数"""
        if reward > 0:
            self.alpha[arm] += 1
        else:
            self.beta[arm] += 1

    def get_estimated_rates(self) -> list:
        """获取每个臂的估计转化率"""
        return [
            self.alpha[i] / (self.alpha[i] + self.beta[i])
            for i in range(self.n_arms)
        ]

# 模拟对比
def simulate_mab(algorithm, true_rates: list, n_rounds: int = 10000):
    total_reward = 0
    optimal_arm = np.argmax(true_rates)
    optimal_count = 0

    for _ in range(n_rounds):
        arm = algorithm.select_arm()
        reward = 1 if np.random.random() < true_rates[arm] else 0
        algorithm.update(arm, reward)
        total_reward += reward
        if arm == optimal_arm:
            optimal_count += 1

    return {
        "total_reward": total_reward,
        "optimal_arm_rate": optimal_count / n_rounds,
        "estimated_rates": algorithm.get_estimated_rates()
            if hasattr(algorithm, 'get_estimated_rates')
            else algorithm.values.tolist()
    }

# 对比两种算法
true_rates = [0.05, 0.08, 0.03]

ts = ThompsonSampling(3)
eg = EpsilonGreedy(3, epsilon=0.1)

ts_result = simulate_mab(ts, true_rates)
eg_result = simulate_mab(eg, true_rates)

print(f"Thompson Sampling: {ts_result}")
print(f"Epsilon-Greedy: {eg_result}")

5.4 UCB算法

class UCB1:
    """
    Upper Confidence Bound算法
    基于"乐观面对不确定性"原则
    """

    def __init__(self, n_arms: int):
        self.n_arms = n_arms
        self.counts = np.zeros(n_arms)
        self.values = np.zeros(n_arms)
        self.total_pulls = 0

    def select_arm(self) -> int:
        # 如果有未尝试的臂，优先选择
        for arm in range(self.n_arms):
            if self.counts[arm] == 0:
                return arm

        # UCB公式：利用项 + 探索项
        ucb_values = np.zeros(self.n_arms)
        for arm in range(self.n_arms):
            exploitation = self.values[arm]
            exploration = np.sqrt(
                2 * np.log(self.total_pulls) / self.counts[arm]
            )
            ucb_values[arm] = exploitation + exploration

        return int(np.argmax(ucb_values))

    def update(self, arm: int, reward: float):
        self.counts[arm] += 1
        self.total_pulls += 1
        n = self.counts[arm]
        self.values[arm] = self.values[arm] * (n - 1) / n + reward / n

6. 增长实验设计

6.1 实验框架：ICE评分

在众多实验想法中，优先执行最有价值的实验：

@dataclass
class Experiment:
    name: str
    hypothesis: str
    description: str
    impact: int        # 1-10: 对核心指标的潜在影响
    confidence: int    # 1-10: 对成功的信心
    ease: int          # 1-10: 实施难度（越容易分越高）
    primary_metric: str
    secondary_metrics: list
    estimated_duration_days: int
    estimated_traffic: int

class ICEPrioritizer:
    """ICE评分模型"""

    @staticmethod
    def score(experiment: Experiment) -> float:
        return (experiment.impact + experiment.confidence + experiment.ease) / 3

    @staticmethod
    def rank(experiments: list) -> list:
        scored = [
            (exp, ICEPrioritizer.score(exp))
            for exp in experiments
        ]
        scored.sort(key=lambda x: x[1], reverse=True)
        return [(exp, round(score, 1)) for exp, score in scored]

# 使用
experiments = [
    Experiment(
        name="优化Prompt模板",
        hypothesis="更结构化的Prompt能提升回答准确率10%",
        description="将现有Prompt改为Chain-of-Thought格式",
        impact=8, confidence=7, ease=9,
        primary_metric="answer_accuracy",
        secondary_metrics=["response_time"],
        estimated_duration_days=7,
        estimated_traffic=50000
    ),
    Experiment(
        name="新增模型GPT-4o",
        hypothesis="更强模型能提升用户满意度",
        description="用GPT-4o替代GPT-3.5-turbo",
        impact=9, confidence=8, ease=3,
        primary_metric="user_satisfaction",
        secondary_metrics=["cost_per_query"],
        estimated_duration_days=14,
        estimated_traffic=100000
    ),
    Experiment(
        name="推荐结果重新排序",
        hypothesis="个性化排序能提升点击率",
        description="引入用户历史行为的个性化排序",
        impact=6, confidence=5, ease=6,
        primary_metric="click_through_rate",
        secondary_metrics=["engagement_time"],
        estimated_duration_days=10,
        estimated_traffic=80000
    )
]

ranked = ICEPrioritizer.rank(experiments)
for exp, score in ranked:
    print(f"[ICE={score}] {exp.name}: {exp.hypothesis}")

6.2 实验设计模板

@dataclass
class ExperimentDesign:
    """实验设计文档模板"""

    # 基本信息
    name: str
    owner: str
    start_date: str
    end_date: str

    # 假设
    hypothesis: str
    # 例："将推荐算法从协同过滤改为深度学习模型，
    #       预计点击率提升10%以上"

    # 变量
    independent_variable: str  # 自变量：你改变的
    dependent_variables: list  # 因变量：你测量的
    control_variant: str       # 对照组描述
    treatment_variants: list   # 实验组描述

    # 样本
    target_population: str     # 目标用户群
    sample_size: int
    traffic_split: dict        # variant -> percentage

    # 指标
    primary_metric: str
    guardrail_metrics: list    # 护栏指标（不能恶化的指标）
    # 例：["page_load_time < 2s", "error_rate < 0.1%"]

    # 统计设计
    alpha: float = 0.05
    power: float = 0.80
    mde: float = 0.05

    def to_document(self) -> str:
        return f"""
# 实验设计文档: {self.name}

## 基本信息
- **负责人**: {self.owner}
- **时间**: {self.start_date} ~ {self.end_date}

## 假设
{self.hypothesis}

## 实验设计
- **自变量**: {self.independent_variable}
- **因变量**: {', '.join(self.dependent_variables)}
- **对照组**: {self.control_variant}
- **实验组**: {', '.join(self.treatment_variants)}

## 样本与分流
- **目标人群**: {self.target_population}
- **样本量**: {self.sample_size:,}
- **流量分配**: {self.traffic_split}

## 评估指标
- **主要指标**: {self.primary_metric}
- **护栏指标**: {', '.join(self.guardrail_metrics)}

## 统计参数
- α = {self.alpha}, Power = {self.power}, MDE = {self.mde}
"""

7. 数据收集与分析

7.1 事件追踪系统

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json

@dataclass
class ExperimentEvent:
    """实验事件"""
    event_id: str
    user_id: str
    experiment_id: str
    variant: str
    event_type: str      # exposure, conversion, custom
    event_name: str      # page_view, click, purchase, etc.
    value: float = 0.0
    properties: dict = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)

class ExperimentTracker:
    """实验数据追踪器"""

    def __init__(self):
        self.events: list[ExperimentEvent] = []
        self.exposures: dict = {}  # user_id -> variant (去重)

    def track_exposure(self, user_id: str, experiment_id: str, variant: str):
        """记录用户进入实验"""
        key = f"{experiment_id}:{user_id}"
        if key in self.exposures:
            return  # 防止重复曝光

        self.exposures[key] = variant
        self._emit(ExperimentEvent(
            event_id=f"exp_{datetime.now().timestamp()}",
            user_id=user_id,
            experiment_id=experiment_id,
            variant=variant,
            event_type="exposure",
            event_name="experiment_exposure"
        ))

    def track_conversion(self, user_id: str, experiment_id: str,
                         event_name: str, value: float = 1.0,
                         properties: dict = None):
        """记录转化事件"""
        key = f"{experiment_id}:{user_id}"
        variant = self.exposures.get(key)
        if not variant:
            return  # 用户未进入实验

        self._emit(ExperimentEvent(
            event_id=f"conv_{datetime.now().timestamp()}",
            user_id=user_id,
            experiment_id=experiment_id,
            variant=variant,
            event_type="conversion",
            event_name=event_name,
            value=value,
            properties=properties or {}
        ))

    def _emit(self, event: ExperimentEvent):
        self.events.append(event)
        # 生产环境：发送到Kafka/数据仓库
        # self.kafka_producer.send("experiment_events", event)

    def get_experiment_data(self, experiment_id: str) -> dict:
        """获取实验数据汇总"""
        from collections import defaultdict

        variant_data = defaultdict(lambda: {"exposures": 0, "conversions": 0, "values": []})

        for event in self.events:
            if event.experiment_id != experiment_id:
                continue
            if event.event_type == "exposure":
                variant_data[event.variant]["exposures"] += 1
            elif event.event_type == "conversion":
                variant_data[event.variant]["conversions"] += 1
                variant_data[event.variant]["values"].append(event.value)

        return dict(variant_data)

7.2 指标计算

class MetricsCalculator:
    """实验指标计算器"""

    @staticmethod
    def conversion_rate(data: dict) -> dict:
        """计算各变体的转化率"""
        results = {}
        for variant, info in data.items():
            exposures = info["exposures"]
            conversions = info["conversions"]
            rate = conversions / exposures if exposures > 0 else 0
            results[variant] = {
                "exposures": exposures,
                "conversions": conversions,
                "rate": round(rate, 6)
            }
        return results

    @staticmethod
    def revenue_metrics(data: dict) -> dict:
        """计算收入相关指标"""
        results = {}
        for variant, info in data.items():
            values = info.get("values", [])
            if not values:
                results[variant] = {"total": 0, "mean": 0, "median": 0}
                continue
            results[variant] = {
                "total": round(sum(values), 2),
                "mean": round(np.mean(values), 2),
                "median": round(np.median(values), 2),
                "std": round(np.std(values), 2),
                "count": len(values)
            }
        return results

    @staticmethod
    def segment_analysis(data: dict, segments: dict) -> dict:
        """分群分析"""
        # segments: user_id -> segment_name
        segment_data = defaultdict(lambda: defaultdict(lambda: {
            "exposures": 0, "conversions": 0
        }))

        for event_data in data:
            segment = segments.get(event_data["user_id"], "unknown")
            variant = event_data["variant"]
            if event_data["event_type"] == "exposure":
                segment_data[segment][variant]["exposures"] += 1
            elif event_data["event_type"] == "conversion":
                segment_data[segment][variant]["conversions"] += 1

        return dict(segment_data)

8. 显著性检验与结果解读

8.1 完整的实验结果分析

from scipy import stats
import numpy as np

class ExperimentAnalyzer:
    """实验结果分析器"""

    @staticmethod
    def analyze_proportions(control_data: dict, treatment_data: dict,
                            alpha: float = 0.05) -> dict:
        """
        分析比例型指标（转化率、点击率等）
        """
        n_c, x_c = control_data["exposures"], control_data["conversions"]
        n_t, x_t = treatment_data["exposures"], treatment_data["conversions"]

        p_c = x_c / n_c
        p_t = x_t / n_t

        # Z检验
        p_pool = (x_c + x_t) / (n_c + n_t)
        se = np.sqrt(p_pool * (1 - p_pool) * (1/n_c + 1/n_t))
        z = (p_t - p_c) / se
        p_value = 2 * (1 - stats.norm.cdf(abs(z)))

        # 置信区间
        se_diff = np.sqrt(p_c*(1-p_c)/n_c + p_t*(1-p_t)/n_t)
        ci = ((p_t - p_c) - 1.96*se_diff, (p_t - p_c) + 1.96*se_diff)

        # 相对提升
        relative_lift = (p_t - p_c) / p_c * 100 if p_c > 0 else 0

        return {
            "control_rate": round(p_c, 6),
            "treatment_rate": round(p_t, 6),
            "absolute_diff": round(p_t - p_c, 6),
            "relative_lift_pct": round(relative_lift, 2),
            "z_statistic": round(z, 4),
            "p_value": round(p_value, 6),
            "ci_95": (round(ci[0], 6), round(ci[1], 6)),
            "significant": p_value < alpha,
            "recommendation": "deploy" if p_value < alpha and p_t > p_c else "no_change"
        }

    @staticmethod
    def analyze_continuous(control_values: list, treatment_values: list,
                           alpha: float = 0.05) -> dict:
        """
        分析连续型指标（响应时间、评分等）
        使用t检验
        """
        # 方差齐性检验
        levene_stat, levene_p = stats.levene(control_values, treatment_values)
        equal_var = levene_p > 0.05

        # t检验
        t_stat, p_value = stats.ttest_ind(
            control_values, treatment_values,
            equal_var=equal_var
        )

        # 效应量（Cohen's d）
        pooled_std = np.sqrt(
            (np.std(control_values)**2 + np.std(treatment_values)**2) / 2
        )
        cohens_d = (np.mean(treatment_values) - np.mean(control_values)) / pooled_std

        return {
            "control_mean": round(np.mean(control_values), 4),
            "treatment_mean": round(np.mean(treatment_values), 4),
            "control_std": round(np.std(control_values), 4),
            "treatment_std": round(np.std(treatment_values), 4),
            "t_statistic": round(t_stat, 4),
            "p_value": round(p_value, 6),
            "cohens_d": round(cohens_d, 4),
            "effect_size": "small" if abs(cohens_d) < 0.2
                          else "medium" if abs(cohens_d) < 0.8
                          else "large",
            "significant": p_value < alpha,
            "equal_var_assumed": equal_var
        }

    @staticmethod
    def check_guardrail_metrics(metrics: dict, thresholds: dict) -> dict:
        """检查护栏指标是否被突破"""
        violations = []
        for metric, threshold in thresholds.items():
            current = metrics.get(metric)
            if current is None:
                continue
            if ">" in threshold:
                limit = float(threshold.replace(">", ""))
                if current <= limit:
                    violations.append(f"{metric}={current} <= {limit}")
            elif "<" in threshold:
                limit = float(threshold.replace("<", ""))
                if current >= limit:
                    violations.append(f"{metric}={current} >= {limit}")

        return {
            "passed": len(violations) == 0,
            "violations": violations
        }

8.2 多重比较校正

当同时测试多个变体或多个指标时，需要进行多重比较校正：

def bonferroni_correction(p_values: list, alpha: float = 0.05) -> dict:
    """Bonferroni校正"""
    n = len(p_values)
    adjusted_alpha = alpha / n
    return {
        "adjusted_alpha": adjusted_alpha,
        "significant": [p < adjusted_alpha for p in p_values],
        "method": "bonferroni"
    }

def benjamini_hochberg(p_values: list, alpha: float = 0.05) -> dict:
    """Benjamini-Hochberg FDR校正"""
    n = len(p_values)
    sorted_indices = np.argsort(p_values)
    sorted_p = np.array(p_values)[sorted_indices]

    # 计算BH阈值
    thresholds = [(i + 1) / n * alpha for i in range(n)]

    # 找到最大的k使得 p_(k) <= k/n * alpha
    significant = np.zeros(n, dtype=bool)
    max_k = -1
    for k in range(n - 1, -1, -1):
        if sorted_p[k] <= thresholds[k]:
            max_k = k
            break

    if max_k >= 0:
        for i in range(max_k + 1):
            significant[i] = True

    # 恢复原始顺序
    result = np.zeros(n, dtype=bool)
    for i, idx in enumerate(sorted_indices):
        result[idx] = significant[i]

    return {
        "significant": result.tolist(),
        "method": "benjamini_hochberg",
        "fdr_level": alpha
    }

# 示例：5个实验的p值
p_values = [0.01, 0.03, 0.04, 0.06, 0.08]
print("Bonferroni:", bonferroni_correction(p_values))
print("BH:", benjamini_hochberg(p_values))

8.3 结果解读指南

实验结果决策树：

p值 < α ?
├── 是 → 统计显著
│   ├── 护栏指标通过？
│   │   ├── 是 → ✅ 推荐上线
│   │   └── 否 → ⚠️ 需要调查护栏指标恶化原因
│   └── 效应量足够大？
│       ├── 是 → 业务意义明确
│       └── 否 → 统计显著但业务价值有限
└── 否 → 统计不显著
    ├── 样本量足够？
    │   ├── 是 → ❌ 变更可能无效
    │   └── 否 → ⏳ 继续收集数据或增大样本
    └── 观察到正向趋势？
        ├── 是 → 考虑延长实验
        └── 否 → 放弃该变更

9. 持续迭代框架

9.1 实验驱动的增长飞轮

class GrowthExperimentFramework:
    """增长实验框架"""

    def __init__(self):
        self.experiment_log = []
        self.learnings = []

    def run_cycle(self, cycle_name: str) -> dict:
        """运行一个实验周期"""

        # 1. 分析现状，发现问题
        insights = self.analyze_current_state()

        # 2. 生成实验假设
        hypotheses = self.generate_hypotheses(insights)

        # 3. ICE评分排序
        prioritized = ICEPrioritizer.rank(hypotheses)

        # 4. 设计实验
        designs = [self.design_experiment(exp) for exp, _ in prioritized[:3]]

        # 5. 实施实验
        results = [self.run_experiment(design) for design in designs]

        # 6. 分析结果，提取洞察
        for result in results:
            learning = self.extract_learning(result)
            self.learnings.append(learning)

        return {
            "cycle": cycle_name,
            "experiments_run": len(results),
            "significant_results": sum(1 for r in results if r.get("significant")),
            "learnings": self.learnings[-len(results):]
        }

    def analyze_current_state(self) -> dict:
        """分析当前产品状态，发现增长机会"""
        # 实际实现：查询数据仓库
        return {
            "funnel": {
                "awareness": 100000,
                "signup": 10000,
                "activation": 5000,
                "retention_d7": 2000,
                "revenue": 500
            },
            "bottleneck": "signup_to_activation",
            "improvement_areas": ["onboarding_flow", "first_value_time"]
        }

    def generate_hypotheses(self, insights: dict) -> list:
        """基于洞察生成实验假设"""
        return [
            Experiment(
                name="简化注册流程",
                hypothesis="减少注册步骤从5步到2步，预计激活率提升20%",
                description="使用手机号一键注册替代邮箱注册",
                impact=9, confidence=7, ease=8,
                primary_metric="activation_rate",
                secondary_metrics=["signup_completion"],
                estimated_duration_days=7,
                estimated_traffic=50000
            ),
            Experiment(
                name="新用户引导优化",
                hypothesis="交互式引导比静态教程提升次日留存15%",
                description="将PDF教程改为交互式任务引导",
                impact=8, confidence=6, ease=5,
                primary_metric="d1_retention",
                secondary_metrics=["task_completion"],
                estimated_duration_days=14,
                estimated_traffic=30000
            )
        ]

    def design_experiment(self, experiment: Experiment) -> ExperimentDesign:
        """将实验想法转化为实验设计"""
        n = calculate_sample_size(
            baseline_rate=0.05,
            mde=0.15,
            alpha=0.05,
            power=0.80
        )
        return ExperimentDesign(
            name=experiment.name,
            owner="growth_team",
            start_date="2024-01-15",
            end_date="2024-01-29",
            hypothesis=experiment.hypothesis,
            independent_variable="onboarding_flow",
            dependent_variables=[experiment.primary_metric],
            control_variant="现有5步注册流程",
            treatment_variants=["2步手机号注册"],
            target_population="新注册用户",
            sample_size=n,
            traffic_split={"control": 50, "treatment": 50},
            primary_metric=experiment.primary_metric,
            guardrail_metrics=["error_rate < 1%", "page_load < 2s"]
        )

    def run_experiment(self, design: ExperimentDesign) -> dict:
        """执行实验（实际环境中是真实运行）"""
        # 模拟
        return {"significant": True, "lift": 18.5}

    def extract_learning(self, result: dict) -> dict:
        """从实验结果中提取可复用的知识"""
        return {
            "timestamp": datetime.now().isoformat(),
            "result": result,
            "learning": "简化注册流程显著提升激活率",
            "action": "全量上线新注册流程",
            "next_steps": "优化首次使用体验"
        }

9.2 实验知识库

class ExperimentKnowledgeBase:
    """实验知识库——记录所有实验的历史和洞察"""

    def __init__(self):
        self.entries = []

    def add_entry(self, experiment_id: str, result: dict, learning: str):
        self.entries.append({
            "experiment_id": experiment_id,
            "date": datetime.now().isoformat(),
            "result": result,
            "learning": learning
        })

    def search(self, keyword: str) -> list:
        return [
            e for e in self.entries
            if keyword.lower() in str(e).lower()
        ]

    def get_success_patterns(self) -> list:
        """提取成功实验的共同模式"""
        successes = [
            e for e in self.entries
            if e["result"].get("significant") and
               e["result"].get("lift", 0) > 0
        ]
        return successes

    def generate_report(self) -> str:
        total = len(self.entries)
        successful = sum(
            1 for e in self.entries
            if e["result"].get("significant")
        )
        return f"""
实验知识库报告
==============
总实验数: {total}
成功实验: {successful}
成功率: {successful/total*100:.1f}%

最近5条实验:
""" + "\n".join(
            f"- [{e['date'][:10]}] {e['experiment_id']}: {e['learning']}"
            for e in self.entries[-5:]
        )

10. 常用A/B测试工具对比

10.1 工具对比表

10.2 开源方案：GrowthBook集成

# GrowthBook Python SDK 示例
from growthbook import GrowthBook

def init_growthbook(user_id: str) -> GrowthBook:
    gb = GrowthBook(
        api_host="https://cdn.growthbook.io",
        client_key="sdk-abc123",
        attributes={
            "id": user_id,
            "country": "CN",
            "device": "mobile"
        }
    )
    return gb

def get_recommendation_model(user_id: str) -> str:
    gb = init_growthbook(user_id)

    # 自动获取实验分组
    variant = gb.get_feature_value("recommendation-model", "control")

    if variant == "treatment":
        return "deep_learning_v2"
    else:
        return "collaborative_filtering_v1"

# 使用
model = get_recommendation_model("user_12345")

10.3 自研轻量方案

class SimpleABTestingPlatform:
    """轻量级自研A/B测试平台"""

    def __init__(self, db_path: str = ":memory:"):
        import sqlite3
        self.db = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS experiments (
                id TEXT PRIMARY KEY,
                name TEXT,
                variants TEXT,  -- JSON
                traffic_split TEXT,  -- JSON
                status TEXT DEFAULT 'draft',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS assignments (
                user_id TEXT,
                experiment_id TEXT,
                variant TEXT,
                assigned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                PRIMARY KEY (user_id, experiment_id)
            )
        """)
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT,
                experiment_id TEXT,
                variant TEXT,
                event_name TEXT,
                value REAL DEFAULT 0,
                properties TEXT,  -- JSON
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.db.commit()

    def create_experiment(self, exp_id: str, name: str,
                          variants: dict, traffic_split: dict):
        import json
        self.db.execute(
            "INSERT INTO experiments (id, name, variants, traffic_split, status) VALUES (?, ?, ?, ?, ?)",
            (exp_id, name, json.dumps(variants), json.dumps(traffic_split), "running")
        )
        self.db.commit()

    def assign(self, exp_id: str, user_id: str) -> str:
        import json
        # 检查是否已分配
        row = self.db.execute(
            "SELECT variant FROM assignments WHERE user_id=? AND experiment_id=?",
            (user_id, exp_id)
        ).fetchone()
        if row:
            return row[0]

        # 获取流量分配
        exp = self.db.execute(
            "SELECT traffic_split FROM experiments WHERE id=? AND status='running'",
            (exp_id,)
        ).fetchone()
        if not exp:
            return "control"

        split = json.loads(exp[0])
        hash_val = int(hashlib.md5(f"{exp_id}:{user_id}".encode()).hexdigest(), 16) % 100

        cumulative = 0
        variant = "control"
        for v, pct in split.items():
            cumulative += pct
            if hash_val < cumulative:
                variant = v
                break

        self.db.execute(
            "INSERT OR IGNORE INTO assignments (user_id, experiment_id, variant) VALUES (?, ?, ?)",
            (user_id, exp_id, variant)
        )
        self.db.commit()
        return variant

    def track(self, user_id: str, exp_id: str, event_name: str,
              value: float = 0, properties: dict = None):
        import json
        # 获取用户分组
        row = self.db.execute(
            "SELECT variant FROM assignments WHERE user_id=? AND experiment_id=?",
            (user_id, exp_id)
        ).fetchone()
        if not row:
            return

        self.db.execute(
            "INSERT INTO events (user_id, experiment_id, variant, event_name, value, properties) VALUES (?, ?, ?, ?, ?, ?)",
            (user_id, exp_id, row[0], event_name, value, json.dumps(properties or {}))
        )
        self.db.commit()

    def get_results(self, exp_id: str) -> dict:
        rows = self.db.execute("""
            SELECT variant,
                   COUNT(DISTINCT CASE WHEN event_name='exposure' THEN user_id END) as exposures,
                   COUNT(DISTINCT CASE WHEN event_name='conversion' THEN user_id END) as conversions,
                   SUM(CASE WHEN event_name='conversion' THEN value ELSE 0 END) as total_value
            FROM events
            WHERE experiment_id=?
            GROUP BY variant
        """, (exp_id,)).fetchall()

        results = {}
        for variant, exposures, conversions, total_value in rows:
            rate = conversions / exposures if exposures > 0 else 0
            results[variant] = {
                "exposures": exposures,
                "conversions": conversions,
                "conversion_rate": round(rate, 6),
                "total_value": round(total_value, 2)
            }
        return results

# 使用示例
platform = SimpleABTestingPlatform()

platform.create_experiment(
    exp_id="new_search_algo",
    name="新搜索算法测试",
    variants={"control": "bm25", "treatment": "dense_retrieval"},
    traffic_split={"control": 50, "treatment": 50}
)

# 用户进入实验
for i in range(1000):
    uid = f"user_{i}"
    variant = platform.assign("new_search_algo", uid)
    platform.track(uid, "new_search_algo", "exposure")
    # 模拟转化
    import random
    rate = 0.05 if variant == "control" else 0.065
    if random.random() < rate:
        platform.track(uid, "new_search_algo", "conversion", value=10.0)

# 查看结果
results = platform.get_results("new_search_algo")
for variant, data in results.items():
    print(f"{variant}: {data}")

11. 总结与最佳实践

核心原则

1. 先有假设，再做实验：不要为了A/B测试而A/B测试
2. 一次只测一个变量：确保因果关系清晰
3. 样本量先行：实验前计算需要多少数据
4. 关注护栏指标：主要指标提升不能以牺牲其他指标为代价
5. 记录一切：每个实验的假设、设计、结果和学习都要归档

常见陷阱

AI应用实验检查清单

• 明确实验假设和预期效果
• 计算样本量和实验周期
• 确认分流方案（用户维度 vs 请求维度）
• 设置Feature Flag和流量分配
• 验证数据埋点准确性
• 确认护栏指标和告警阈值
• 检查模型确定性（相同输入相同输出）
• 考虑延迟指标的观测周期
• 准备回滚方案
• 预注册分析计划

推荐技术栈

本文介绍了AI应用A/B测试的完整知识体系，从统计学原理到工程实现，从传统假设检验到多臂老虎机算法。核心思想是：用数据驱动决策，用实验验证假设，用框架加速迭代。在AI应用中，模型版本管理和不确定性量化是A/B测试的关键差异点，需要特别关注。