LLM A/B 测试的特殊性 传统 Web 产品的 A/B 测试已经相当成熟——改个按钮颜色、调整文案、优化布局，通过 CTR 和转化率就能快速判断优劣。但 LLM 产品的 A/B 测试面临独特挑战：
输出非确定性：同一个输入，两次调用可能得到不同输出 质量维度多元：没有单一指标能衡量"回答好不好" 长尾效应：大部分对话可能表现类似，但少数关键场景差异巨大 学习效应：用户可能需要时间适应新模型的行为风格 A/B 测试完整流程 ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 假设构建 │ -> │ 实验设计 │ -> │ 执行与监控 │ -> │ 分析与决策 │ │ Hypothesis │ │ Design │ │ Execute │ │ Analyze │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ 一、假设构建 每个 A/B 测试都应始于一个清晰的假设：
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 LLM 特有的样本量考量 因素 影响 调整策略 输出非确定性 增加方差，需要更多样本 对同一输入运行多次取均值 用户异质性 不同用户群体效应不同 分层随机化 冷启动效应 新模型初期表现可能不稳定 设置预热期 时段效应 不同时间段对话质量不同 确保两组同时段运行 三、实验设计 随机化策略 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") 六、常见陷阱与解决方案 陷阱一：Peeking（偷看） 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"], } 决策速查表 情况 主要指标 护栏指标 决策 主要指标显著提升 ✅ ✅ 未突破 全量发布 主要指标显著提升 ✅ ❌ 突破 不发布，分析权衡 主要指标显著下降 ❌ ✅ 保持对照组 主要指标无显著差异 - ✅ 视成本决定 主要指标无显著差异 - ❌ 保持对照组 结语 LLM A/B 测试比传统 Web A/B 测试复杂得多，但核心原则不变：清晰的假设、足够的样本量、正确的统计检验、严格的护栏监控。最大的陷阱不是统计方法不对，而是没有想清楚要测什么就开始测。在 LLM 时代，一个好的 A/B 测试框架是产品迭代的基础设施——没有它，所有的"优化"都只是猜测。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
