LLM A/B 测试实践：统计显著性与业务指标

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  假设构建    │ -> │  实验设计    │ -> │  执行与监控  │ -> │  分析与决策  │
│  Hypothesis │    │  Design     │    │  Execute    │    │  Analyze    │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘

from dataclasses import dataclass
from typing import Optional

@dataclass
class ABTestHypothesis:
    """A/B 测试假设"""
    # 假设描述
    statement: str  # "将模型从 GPT-4o-mini 升级到 GPT-4o 后，
                    #  在复杂推理任务上的用户满意度将提升 10%"
    
    # 自变量
    treatment_description: str  # "使用 GPT-4o 替代 GPT-4o-mini"
    control_description: str    # "继续使用 GPT-4o-mini"
    
    # 因变量（核心指标）
    primary_metric: str         # "CSAT 评分"
    expected_effect: float      # 0.10 (提升10%)
    expected_direction: str     # "increase" or "decrease"
    
    # 次要指标
    secondary_metrics: list     # ["重试率", "平均对话轮次", "人工求助率"]
    
    # 护栏指标（不应恶化的指标）
    guardrail_metrics: list     # ["延迟 P95", "成本/请求", "安全违规率"]
    
    # 目标人群
    target_segment: str         # "all_users" 或 "power_users" 等
    
    # 最小可检测效应 (MDE)
    mde: float                  # 0.05 (最小可检测 5% 变化)

    def validate(self) -> list[str]:
        issues = []
        if not self.statement or len(self.statement) < 10:
            issues.append("假设描述太短")
        if not self.primary_metric:
            issues.append("缺少主要指标")
        if self.expected_effect <= 0:
            issues.append("预期效应应为正数")
        if not self.guardrail_metrics:
            issues.append("缺少护栏指标——可能导致意外回退")
        return issues

import math
from scipy import stats

class SampleSizeCalculator:
    """A/B 测试样本量计算器"""
    
    @staticmethod
    def for_proportion(
        baseline_rate: float,
        mde: float,  # 最小可检测效应（绝对值）
        alpha: float = 0.05,
        power: float = 0.80,
        two_sided: bool = True,
    ) -> dict:
        """
        比例类指标的样本量计算（如 CSAT 满意率、重试率）
        
        参数:
            baseline_rate: 基线比例（如当前 CSAT = 0.75）
            mde: 最小可检测效应（如 0.05 表示检测 5% 绝对变化）
            alpha: 显著性水平
            power: 统计功效
        """
        z_alpha = stats.norm.ppf(1 - alpha / 2 if two_sided else 1 - alpha)
        z_beta = stats.norm.ppf(power)
        
        p1 = baseline_rate
        p2 = baseline_rate + mde
        p_avg = (p1 + p2) / 2
        
        n = ((z_alpha * math.sqrt(2 * p_avg * (1 - p_avg)) +
              z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / (p2 - p1) ** 2
        
        return {
            "sample_per_group": math.ceil(n),
            "total_sample": math.ceil(n * 2),
            "baseline_rate": baseline_rate,
            "mde": mde,
            "alpha": alpha,
            "power": power,
            "expected_duration_days": math.ceil(n * 2 / 1000),  # 假设每天1000用户
        }

    @staticmethod
    def for_continuous(
        baseline_mean: float,
        baseline_std: float,
        mde: float,
        alpha: float = 0.05,
        power: float = 0.80,
    ) -> dict:
        """
        连续型指标的样本量计算（如评分均值、延迟）
        
        参数:
            baseline_mean: 基线均值
            baseline_std: 基线标准差
            mde: 最小可检测效应（绝对变化量）
        """
        z_alpha = stats.norm.ppf(1 - alpha / 2)
        z_beta = stats.norm.ppf(power)
        
        n = 2 * ((z_alpha + z_beta) * baseline_std / mde) ** 2
        
        return {
            "sample_per_group": math.ceil(n),
            "total_sample": math.ceil(n * 2),
            "baseline_mean": baseline_mean,
            "baseline_std": baseline_std,
            "mde": mde,
        }

# 示例：计算 CSAT 从 75% 提升到 80% 所需的样本量
calc = SampleSizeCalculator()
result = calc.for_proportion(baseline_rate=0.75, mde=0.05)
# 输出: sample_per_group ≈ 2554, total_sample ≈ 5108

import hashlib
import random

class ABTestRouter:
    """A/B 测试流量分配器"""
    
    def __init__(self, experiment_id: str, traffic_split: dict = None):
        self.experiment_id = experiment_id
        # 默认 50/50 分配
        self.traffic_split = traffic_split or {"control": 0.5, "treatment": 0.5}
        self.salt = experiment_id  # 用于哈希的盐值

    def assign(self, user_id: str) -> str:
        """
        基于用户 ID 的确定性分配
        同一用户始终分到同一组
        """
        # 使用哈希确保确定性
        hash_input = f"{self.salt}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        bucket = (hash_value % 10000) / 10000.0
        
        cumulative = 0.0
        for group, ratio in self.traffic_split.items():
            cumulative += ratio
            if bucket < cumulative:
                return group
        return list(self.traffic_split.keys())[-1]  # fallback

    def assign_with_layering(self, user_id: str, 
                              existing_experiments: list[str]) -> str:
        """
        分层分配：避免与正在运行的其他实验冲突
        """
        # 将已有实验纳入哈希
        hash_input = f"{self.salt}:{user_id}:{':'.join(sorted(existing_experiments))}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        bucket = (hash_value % 10000) / 10000.0
        
        cumulative = 0.0
        for group, ratio in self.traffic_split.items():
            cumulative += ratio
            if bucket < cumulative:
                return group
        return list(self.traffic_split.keys())[-1]


class StratifiedRouter:
    """分层随机化：确保关键维度均衡"""
    
    STRATA = ["user_type", "region", "device", "usage_level"]
    
    def __init__(self, experiment_id: str, strata_weights: dict = None):
        self.experiment_id = experiment_id
        self.strata_weights = strata_weights or {}
        self.assignments = {}  # 缓存分配结果

    def assign(self, user_id: str, user_attributes: dict) -> str:
        # 计算分层 key
        strata_key = "|".join(
            str(user_attributes.get(s, "unknown")) for s in self.STRATA
        )
        
        # 分层内随机分配
        hash_input = f"{self.experiment_id}:{strata_key}:{user_id}"
        hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16)
        return "treatment" if hash_value % 2 == 0 else "control"

@dataclass
class ABTestConfig:
    experiment_id: str
    name: str
    hypothesis: ABTestHypothesis
    
    # 流量分配
    traffic_split: dict = field(default_factory=lambda: {"control": 0.5, "treatment": 0.5})
    total_traffic_pct: float = 1.0  # 使用多少比例的总流量
    
    # 时间设置
    start_date: str = ""
    end_date: str = ""
    min_duration_days: int = 14  # 最短运行天数
    warmup_days: int = 2         # 预热天数（数据不计入分析）
    
    # 指标配置
    primary_metric: str = ""
    secondary_metrics: list = field(default_factory=list)
    guardrail_metrics: list = field(default_factory=list)
    
    # 统计参数
    alpha: float = 0.05
    power: float = 0.80
    mde: float = 0.05
    
    # 停止规则
    early_stop_on_guardrail: bool = True
    guardrail_thresholds: dict = field(default_factory=lambda: {
        "latency_p95_ms": 5000,     # P95 延迟不超过 5s
        "safety_violation_rate": 0.001,  # 安全违规率不超过 0.1%
        "cost_per_session": 0.15,   # 每次会话成本不超过 $0.15
    })

from scipy import stats
import numpy as np

class SignificanceTester:
    """A/B 测试显著性检验"""
    
    def test_proportion(self, control_success: int, control_total: int,
                        treatment_success: int, treatment_total: int,
                        alpha: float = 0.05) -> dict:
        """
        比例类指标的双比例 Z 检验
        """
        p_control = control_success / control_total
        p_treatment = treatment_success / treatment_total
        p_pooled = (control_success + treatment_success) / (control_total + treatment_total)
        
        se = np.sqrt(p_pooled * (1 - p_pooled) * (1/control_total + 1/treatment_total))
        z_stat = (p_treatment - p_control) / se if se > 0 else 0
        p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))
        
        # 置信区间
        diff = p_treatment - p_control
        ci_lower = diff - stats.norm.ppf(1 - alpha/2) * se
        ci_upper = diff + stats.norm.ppf(1 - alpha/2) * se
        
        # 效应量
        relative_lift = diff / p_control if p_control > 0 else 0
        
        return {
            "test_type": "two_proportion_z_test",
            "control_rate": round(p_control, 4),
            "treatment_rate": round(p_treatment, 4),
            "absolute_diff": round(diff, 4),
            "relative_lift": f"{relative_lift*100:.2f}%",
            "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)],
            "winner": "treatment" if (p_value < alpha and diff > 0) else 
                      ("control" if (p_value < alpha and diff < 0) else "no_significant_diff"),
        }

    def test_continuous(self, control_values: list[float],
                        treatment_values: list[float],
                        alpha: float = 0.05) -> dict:
        """
        连续型指标的检验（t 检验或 Mann-Whitney U 检验）
        """
        control = np.array(control_values)
        treatment = np.array(treatment_values)
        
        # 正态性检验决定使用哪种检验
        if len(control) >= 30 and len(treatment) >= 30:
            # 大样本使用 Welch t 检验
            t_stat, p_value = stats.ttest_ind(treatment, control, equal_var=False)
            test_name = "welch_t_test"
        else:
            # 小样本使用 Mann-Whitney U
            u_stat, p_value = stats.mannwhitneyu(treatment, control, alternative="two-sided")
            t_stat = u_stat
            test_name = "mann_whitney_u"
        
        diff = treatment.mean() - control.mean()
        # 置信区间（基于 t 分布）
        pooled_se = np.sqrt(treatment.var(ddof=1)/len(treatment) + control.var(ddof=1)/len(control))
        df = len(treatment) + len(control) - 2
        ci_lower = diff - stats.t.ppf(1-alpha/2, df) * pooled_se
        ci_upper = diff + stats.t.ppf(1-alpha/2, df) * pooled_se
        
        return {
            "test_type": test_name,
            "control_mean": round(control.mean(), 4),
            "control_std": round(control.std(), 4),
            "treatment_mean": round(treatment.mean(), 4),
            "treatment_std": round(treatment.std(), 4),
            "absolute_diff": round(diff, 4),
            "relative_lift": f"{(diff/control.mean())*100:.2f}%" if control.mean() != 0 else "N/A",
            "statistic": round(t_stat, 4),
            "p_value": round(p_value, 6),
            "significant": p_value < alpha,
            "ci_95": [round(ci_lower, 4), round(ci_upper, 4)],
        }

    def sequential_test(self, daily_data: list[dict], alpha: float = 0.05) -> dict:
        """
        序贯检验：每天检查是否可以提前停止
        使用 Bonferroni 校正控制总犯错误率
        """
        num_checks = len(daily_data)
        adjusted_alpha = alpha / num_checks  # Bonferroni 校正
        
        results = []
        for i, day_data in enumerate(daily_data):
            result = self.test_proportion(
                day_data["control_success"], day_data["control_total"],
                day_data["treatment_success"], day_data["treatment_total"],
                alpha=adjusted_alpha
            )
            result["day"] = i + 1
            result["adjusted_alpha"] = adjusted_alpha
            results.append(result)
            
            # 如果已显著或护栏指标触发，可以停止
            if result["significant"]:
                return {
                    "stopped_early": True,
                    "stopped_at_day": i + 1,
                    "final_result": result,
                    "all_daily_results": results,
                }
        
        return {
            "stopped_early": False,
            "final_result": results[-1] if results else None,
            "all_daily_results": results,
        }

class MetricFramework:
    """A/B 测试指标分层框架"""
    
    METRIC_TREE = {
        "北极星指标": {
            "description": "最能反映产品价值的单一指标",
            "llm_examples": ["每周活跃对话用户数 (WAC)", "人均每日对话轮次"],
            "sensitivity": "低 — 需要较长时间观察",
        },
        "主要指标": {
            "description": "直接反映假设是否成立的指标",
            "llm_examples": ["CSAT", "任务完成率", "回答准确率"],
            "sensitivity": "中 — 通常 2-4 周可检测",
        },
        "次要指标": {
            "description": "帮助理解主要指标变化原因",
            "llm_examples": ["重试率", "对话深度", "人工求助率"],
            "sensitivity": "高 — 快速响应变化",
        },
        "护栏指标": {
            "description": "确保不出现严重回退",
            "llm_examples": ["延迟 P95", "安全违规率", "每次会话成本"],
            "sensitivity": "高 — 需要实时监控",
        },
        "调试指标": {
            "description": "用于诊断问题，不用于决策",
            "llm_examples": ["Token 使用量", "API 错误率", "特定类别表现"],
            "sensitivity": "高",
        }
    }

    @staticmethod
    def recommend_metrics(scenario: str) -> dict:
        recommendations = {
            "model_upgrade": {
                "primary": "回答准确率（基于评估集）",
                "secondary": ["CSAT", "重试率", "对话深度"],
                "guardrails": ["延迟 P95", "安全违规率", "成本/请求"],
            },
            "prompt_change": {
                "primary": "任务完成率",
                "secondary": ["用户重述率", "回答长度分布"],
                "guardrails": ["安全违规率", "Token 使用量"],
            },
            "feature_addition": {
                "primary": "功能采纳率",
                "secondary": ["CSAT", "会话深度", "NPS"],
                "guardrails": ["延迟 P95", "错误率"],
            },
            "ui_change": {
                "primary": "任务完成率",
                "secondary": ["CSAT", "CES", "会话深度"],
                "guardrails": ["页面加载时间", "错误率"],
            },
        }
        return recommendations.get(scenario, "Unknown scenario")

class PeekingWarning:
    """偷看陷阱演示与解决方案"""
    
    def demonstrate_peeking_problem(self):
        """
        如果每天检查 p 值并在 p<0.05 时停止，
        实际犯错误率远超 5%
        """
        # 模拟 1000 次实验（A 和 B 实际无差异）
        false_positive_count = 0
        for _ in range(1000):
            control = np.random.binomial(1, 0.10, 500)  # 转化率 10%
            treatment = np.random.binomial(1, 0.10, 500)
            
            # 每天检查（假设每天 50 个样本）
            for day in range(1, 11):
                c_slice = control[:day*50]
                t_slice = treatment[:day*50]
                # 简化：使用卡方检验
                _, p_value = stats.chi2_contingency([
                    [c_slice.sum(), len(c_slice) - c_slice.sum()],
                    [t_slice.sum(), len(t_slice) - t_slice.sum()]
                ])
                if p_value < 0.05:
                    false_positive_count += 1
                    break
        
        actual_fpr = false_positive_count / 1000
        return {
            "nominal_alpha": 0.05,
            "actual_false_positive_rate": actual_fpr,
            "inflation_factor": actual_fpr / 0.05,
            "solution": "使用序贯检验或 Alpha Spending 函数"
        }

class SimpsonParadoxCheck:
    """辛普森悖论检测"""
    
    def check(self, data: pd.DataFrame, 
              group_col: str, metric_col: str,
              strata_col: str) -> dict:
        """
        检测是否存在辛普森悖论：
        整体趋势与分层趋势相反
        """
        # 整体差异
        overall = data.groupby(group_col)[metric_col].mean()
        overall_diff = overall.get("treatment", 0) - overall.get("control", 0)
        
        # 分层差异
        strata_results = {}
        reversals = []
        for stratum, stratum_data in data.groupby(strata_col):
            stratum_overall = stratum_data.groupby(group_col)[metric_col].mean()
            stratum_diff = stratum_overall.get("treatment", 0) - stratum_overall.get("control", 0)
            strata_results[stratum] = {
                "control_n": len(stratum_data[stratum_data[group_col] == "control"]),
                "treatment_n": len(stratum_data[stratum_data[group_col] == "treatment"]),
                "control_mean": stratum_overall.get("control", 0),
                "treatment_mean": stratum_overall.get("treatment", 0),
                "diff": stratum_diff,
            }
            # 检测方向反转
            if (overall_diff > 0 and stratum_diff < 0) or \
               (overall_diff < 0 and stratum_diff > 0):
                reversals.append(stratum)
        
        return {
            "overall_diff": overall_diff,
            "strata": strata_results,
            "reversals": reversals,
            "simpson_paradox": len(reversals) > 0,
            "recommendation": "分层分析而非整体分析" if reversals else "整体分析可行",
        }

class ABTestDecision:
    """A/B 测试决策框架"""
    
    @staticmethod
    def decide(primary_result: dict, guardrail_results: dict,
               secondary_results: dict, config: ABTestConfig) -> dict:
        
        # 1. 检查护栏指标
        guardrail_breaches = []
        for metric, result in guardrail_results.items():
            threshold = config.guardrail_thresholds.get(metric)
            if threshold and result.get("treatment_mean", 0) > threshold:
                guardrail_breaches.append({
                    "metric": metric,
                    "treatment_value": result["treatment_mean"],
                    "threshold": threshold,
                })
        
        if guardrail_breaches:
            return {
                "decision": "STOP",
                "reason": "护栏指标被突破",
                "breaches": guardrail_breaches,
            }
        
        # 2. 检查主要指标
        if not primary_result["significant"]:
            return {
                "decision": "CONTINUE" if primary_result.get("p_value", 1) > 0.3 else "INCONCLUSIVE",
                "reason": "主要指标未达统计显著性",
                "p_value": primary_result["p_value"],
                "recommendation": "继续收集数据或增大样本量",
            }
        
        # 3. 主要指标显著
        winner = primary_result["winner"]
        if winner == "treatment":
            # 检查次要指标是否支持
            secondary_support = all(
                r.get("relative_lift", "").startswith("-") is False
                for r in secondary_results.values()
            )
            return {
                "decision": "SHIP",
                "reason": "主要指标显著提升，护栏指标未突破",
                "confidence": "high" if secondary_support else "medium",
                "primary_lift": primary_result["relative_lift"],
                "secondary_support": secondary_support,
            }
        else:
            return {
                "decision": "HOLD",
                "reason": "主要指标显著下降",
                "primary_drop": primary_result["relative_lift"],
            }