引言 Agent 的非确定性使得"感觉更好"不能作为决策依据。一个 Prompt 的微调可能提升某类任务的表现，却悄悄损害了另一类。2026年，A/B 测试已成为 Agent 优化的科学方法——用数据说话，用统计检验做决策。
一、Agent A/B 测试的特殊性 与传统 Web A/B 测试不同，Agent A/B 测试面临独特挑战：
维度 传统 A/B 测试 Agent A/B 测试 指标 点击率、转化率 输出质量、任务完成率、用户满意度 测量 确定性（点击=1/不点击=0） 非确定性（同一输入可能不同输出） 变量 UI 元素 Prompt、模型、工具、温度 噪声 低 高（LLM 输出方差大） 样本量 百万级 千级（成本限制） 指标延迟 即时 分钟级（需要完整执行） 二、实验设计框架 2.1 假设构建 @dataclass class ExperimentHypothesis: """实验假设""" name: str description: str independent_variable: str # 自变量（如 temperature） control_value: any # 对照组值（如 0.3） treatment_value: any # 实验组值（如 0.5） dependent_variables: list[str] # 因变量（如 task_completion_rate） expected_effect: str # 预期效果 min_detectable_effect: float # 最小可检测效应 (MDE) statistical_power: float # 统计功效 (通常 0.8) significance_level: float # 显著性水平 (通常 0.05) def required_sample_size(self) -> int: """计算所需样本量""" # 基于双比例检验的样本量计算 p1 = self.baseline_rate # 基线成功率 p2 = p1 + self.min_detectable_effect # 预期成功率 z_alpha = 1.96 # α=0.05 z_beta = 0.84 # power=0.8 n = ( (z_alpha * (2*p1*(1-p1))**0.5 + z_beta * (p1*(1-p1) + p2*(1-p2))**0.5) ** 2 ) / (p2 - p1) ** 2 return int(n) + 1 # 示例 hypothesis = ExperimentHypothesis( name="temperature_optimization", description="将 temperature 从 0.3 调至 0.5，预期能提升创意写作任务的用户满意度", independent_variable="temperature", control_value=0.3, treatment_value=0.5, dependent_variables=["user_satisfaction", "task_completion_rate"], expected_effect="满意度提升 5%", min_detectable_effect=0.05, statistical_power=0.8, significance_level=0.05, baseline_rate=0.75 # 当前满意度 75% ) # 所需样本量 ≈ 2,435 per group 2.2 实验配置 @dataclass class ExperimentConfig: experiment_id: str name: str hypothesis: ExperimentHypothesis traffic_allocation: float # 实验占总流量比例 (0-1) control_split: float # 对照组在实验流量中的比例 (通常 0.5) targeting_rules: list[Rule] # 目标用户筛选 duration_days: int # 预计运行天数 metrics: list[Metric] # 追踪指标 guardrail_metrics: list[Metric] # 护栏指标（不可恶化） early_stop_rules: list[Rule] # 提前停止规则 cost_budget: float # 实验成本预算 # 护栏指标示例 GUARDRAIL_METRICS = [ Metric(name="error_rate", type="counter", max_threshold=0.05), Metric(name="p95_latency", type="histogram", max_threshold_ms=10000), Metric(name="cost_per_request", type="histogram", max_threshold=0.15), Metric(name="toxic_output_rate", type="counter", max_threshold=0.01), ] 三、流量分配系统 class ExperimentRouter: """实验流量路由""" def __init__(self, redis_client): self.redis = redis_client async def assign( self, user_id: str, agent_name: str ) -> VariantAssignment: """为用户分配实验变体""" # 1. 获取活跃实验 experiments = await self._get_active_experiments(agent_name) for exp in experiments: # 2. 检查目标规则 if not self._matches_targeting(user_id, exp.targeting_rules): continue # 3. 检查是否已分配 existing = await self._get_assignment(user_id, exp.experiment_id) if existing: return existing # 保持一致性 # 4. 一致性哈希分配 bucket = self._hash_bucket(user_id, exp.experiment_id) # 5. 决定是否进入实验 if bucket < exp.traffic_allocation: # 在实验内部分配对照组/实验组 inner_bucket = self._hash_bucket( f"{user_id}:{exp.experiment_id}", "inner" ) if inner_bucket < exp.control_split: variant = "control" else: variant = "treatment" else: variant = "excluded" # 不参与实验 assignment = VariantAssignment( experiment_id=exp.experiment_id, user_id=user_id, variant=variant, config=exp.get_variant_config(variant), assigned_at=datetime.now() ) await self._save_assignment(assignment) return assignment # 没有匹配的实验 return VariantAssignment(variant="default", config={}) def _hash_bucket(self, key: str, salt: str = "") -> float: """一致性哈希，返回 0-1 之间的值""" h = hashlib.sha256(f"{key}:{salt}".encode()).hexdigest() return int(h[:8], 16) / 0xFFFFFFFF 四、指标收集与统计检验 4.1 指标收集器 class ExperimentMetricsCollector: """实验指标收集器""" async def record( self, experiment_id: str, user_id: str, variant: str, metrics: dict ): """记录单次实验观测""" event = { "experiment_id": experiment_id, "user_id": user_id, "variant": variant, "timestamp": time.time(), **metrics # task_completed, satisfaction_score, latency_ms, tokens_used, cost } # 写入时序数据库 await self.influxdb.write( measurement="experiment_events", tags={"experiment_id": experiment_id, "variant": variant}, fields=metrics, timestamp=event["timestamp"] ) async def aggregate( self, experiment_id: str, metric_name: str ) -> dict: """聚合实验指标""" return { "control": await self._compute_stats(experiment_id, "control", metric_name), "treatment": await self._compute_stats(experiment_id, "treatment", metric_name) } async def _compute_stats( self, exp_id: str, variant: str, metric: str ) -> MetricStats: values = await self.influxdb.query( f'SELECT "{metric}" FROM "experiment_events" ' f'WHERE "experiment_id" = \'{exp_id}\' ' f'AND "variant" = \'{variant}\'' ) return MetricStats( n=len(values), mean=statistics.mean(values), std=statistics.stdev(values) if len(values) > 1 else 0, median=statistics.median(values), p25=np.percentile(values, 25), p75=np.percentile(values, 75), p95=np.percentile(values, 95), ) 4.2 统计检验 from scipy import stats import numpy as np class StatisticalTester: """统计显著性检验""" def test_proportion( self, control_successes: int, control_total: int, treatment_successes: int, treatment_total: int, alpha: float = 0.05 ) -> TestResult: """比例检验（用于完成率等二值指标）""" # 卡方检验 contingency = [ [control_successes, control_total - control_successes], [treatment_successes, treatment_total - treatment_successes] ] chi2, p_value, _, _ = stats.chi2_contingency(contingency) # 效应量 p_control = control_successes / control_total p_treatment = treatment_successes / treatment_total effect_size = p_treatment - p_control # 置信区间 se = np.sqrt(p_control*(1-p_control)/control_total + p_treatment*(1-p_treatment)/treatment_total) ci_lower = effect_size - 1.96 * se ci_upper = effect_size + 1.96 * se return TestResult( test="chi_square", p_value=p_value, significant=p_value < alpha, effect_size=effect_size, confidence_interval=(ci_lower, ci_upper), control_rate=p_control, treatment_rate=p_treatment, interpretation=self._interpret( p_value, alpha, effect_size, p_control, p_treatment ) ) def test_continuous( self, control_values: list[float], treatment_values: list[float], alpha: float = 0.05 ) -> TestResult: """连续值检验（用于满意度分数、延迟等）""" # 正态性检验 _, p_normal_ctrl = stats.shapiro(control_values) _, p_normal_treat = stats.shapiro(treatment_values) if p_normal_ctrl > 0.05 and p_normal_treat > 0.05: # 正态分布：使用 t 检验 statistic, p_value = stats.ttest_ind( control_values, treatment_values, equal_var=False # Welch's t-test ) test_name = "welch_t_test" else: # 非正态：使用 Mann-Whitney U 检验 statistic, p_value = stats.mannwhitneyu( control_values, treatment_values, alternative='two-sided' ) test_name = "mann_whitney_u" # 效应量 (Cohen's d) pooled_std = np.sqrt( ((len(control_values)-1) * np.var(control_values, ddof=1) + (len(treatment_values)-1) * np.var(treatment_values, ddof=1)) / (len(control_values) + len(treatment_values) - 2) ) cohens_d = (np.mean(treatment_values) - np.mean(control_values)) / pooled_std return TestResult( test=test_name, p_value=p_value, significant=p_value < alpha, effect_size=cohens_d, control_mean=np.mean(control_values), treatment_mean=np.mean(treatment_values), interpretation=self._interpret_continuous( p_value, alpha, cohens_d, np.mean(control_values), np.mean(treatment_values) ) ) def _interpret(self, p_value, alpha, effect, p_ctrl, p_treat): if p_value >= alpha: return f"无统计显著差异 (p={p_value:.4f} ≥ {alpha})。建议继续收集数据或增大样本量。" direction = "提升" if effect > 0 else "下降" return ( f"统计显著 (p={p_value:.4f} < {alpha})。" f"实验组{direction}了{abs(effect)*100:.1f}个百分点" f"（{p_ctrl:.1%} → {p_treat:.1%}）。" ) 4.3 序贯检验（支持提前停止） class SequentialTester: """序贯检验：允许在实验过程中提前判断""" def __init__(self, alpha: float = 0.05, power: float = 0.8, num_looks: int = 5): # Bonferroni 校正 self.adjusted_alpha = alpha / num_looks self.looks = num_looks self.current_look = 0 def should_stop_early( self, control_data: list, treatment_data: list, sample_size_ratio: float # 当前样本量 / 计划样本量 ) -> EarlyStopDecision: """检查是否可以提前停止""" self.current_look = int(sample_size_ratio * self.looks) result = StatisticalTester().test_continuous( control_data, treatment_data, self.adjusted_alpha ) # 护栏指标检查 guardrail_ok = self._check_guardrails(control_data, treatment_data) if not guardrail_ok: return EarlyStopDecision( should_stop=True, reason="护栏指标恶化，建议立即停止实验", winner="control" ) if result.significant: if result.effect_size > 0: return EarlyStopDecision( should_stop=True, reason=f"实验组显著优于对照组 (p={result.p_value:.4f})", winner="treatment" ) else: return EarlyStopDecision( should_stop=True, reason=f"实验组显著劣于对照组 (p={result.p_value:.4f})", winner="control" ) # 计算当前功效 current_power = self._compute_power( len(control_data), result.effect_size ) if current_power > 0.8 and not result.significant: return EarlyStopDecision( should_stop=True, reason=f"功效充足({current_power:.1%})但无显著差异，停止实验", winner="tie" ) return EarlyStopDecision(should_stop=False) 五、LLM 特有的 A/B 测试方法 5.1 LLM-as-Judge A/B 测试 class LLMJudgeABTest: """使用 LLM 作为评判者的 A/B 测试""" async def judge_pair( self, prompt: str, response_a: str, response_b: str, criteria: list[str] ) -> JudgmentResult: """让 LLM 判断哪个回答更好""" judge_prompt = f"""You are an impartial judge. Compare two responses to the same prompt. Prompt: {prompt} Response A: {response_a} Response B: {response_b} Criteria: {', '.join(criteria)} Evaluate which response is better. Consider: 1. Accuracy and correctness 2. Completeness 3. Clarity and structure 4. Adherence to instructions Respond in JSON: {{ "winner": "A" | "B" | "tie", "confidence": 0.0-1.0, "reasoning": "explanation", "scores": {{"A": float, "B": float}} }}""" response = await self.judge_llm.invoke(judge_prompt, temperature=0.0) return JudgmentResult(**json.loads(response.content)) async def run_experiment( self, test_cases: list[TestCase], control_agent: Agent, treatment_agent: Agent, num_judges: int = 3 # 多评判者取平均 ) -> ExperimentResult: results = [] for case in test_cases: # 生成两组回答 response_ctrl = await control_agent.run(case.input) response_treat = await treatment_agent.run(case.input) # 多评判者投票 judgments = [] for i in range(num_judges): judge = self.judges[i] judgment = await judge.judge_pair( case.input, response_ctrl, response_treat, case.criteria ) judgments.append(judgment) # 多数投票 winner = self._majority_vote(judgments) results.append({ "test_id": case.id, "winner": winner, "confidence": np.mean([j.confidence for j in judgments]), }) # 统计分析 wins_treatment = sum(1 for r in results if r["winner"] == "treatment") wins_control = sum(1 for r in results if r["winner"] == "control") ties = sum(1 for r in results if r["winner"] == "tie") # Bradley-Terry 模型检验 bt_stat = self._bradley_terry_test(wins_treatment, wins_control, ties) return ExperimentResult( wins_treatment=wins_treatment, wins_control=wins_control, ties=ties, p_value=bt_stat.p_value, significant=bt_stat.p_value < 0.05, avg_confidence=np.mean([r["confidence"] for r in results]) ) 六、实验报告自动化 class ExperimentReporter: """自动化实验报告生成""" async def generate_report( self, experiment_id: str ) -> ExperimentReport: exp = await self.repo.get(experiment_id) metrics = await self.collector.aggregate_all(experiment_id) test_results = {} for metric_name, data in metrics.items(): if metric_name in ["task_completed", "user_thumbs_up"]: # 比例检验 result = self.tester.test_proportion( data["control"].successes, data["control"].total, data["treatment"].successes, data["treatment"].total ) else: # 连续值检验 result = self.tester.test_continuous( data["control"].values, data["treatment"].values ) test_results[metric_name] = result # 护栏指标检查 guardrail_status = self._check_guardrails(metrics, exp.guardrail_metrics) # 生成决策建议 recommendation = self._generate_recommendation( test_results, guardrail_status, exp.hypothesis ) return ExperimentReport( experiment=exp, sample_sizes={ "control": metrics["task_completed"]["control"].total, "treatment": metrics["task_completed"]["treatment"].total, }, results=test_results, guardrail_status=guardrail_status, recommendation=recommendation, summary=self._generate_summary(test_results, recommendation), generated_at=datetime.now() ) def _generate_recommendation(self, results, guardrails, hypothesis): primary = results.get(hypothesis.dependent_variables[0]) if not primary.significant: return Recommendation( action="continue_or_stop", reason=f"主指标无显著差异 (p={primary.p_value:.4f})。" f"建议：若已达到计划样本量则停止；否则继续收集数据。" ) if primary.effect_size > 0 and guardrails.all_passed: return Recommendation( action="ship", reason=f"主指标显著提升 (p={primary.p_value:.4f}, " f"效应量={primary.effect_size:.3f})。" f"护栏指标全部通过。建议全量发布。" ) if primary.effect_size < 0: return Recommendation( action="do_not_ship", reason=f"主指标显著下降 (p={primary.p_value:.4f})。不建议发布。" ) if not guardrails.all_passed: return Recommendation( action="do_not_ship", reason=f"主指标虽提升但护栏指标恶化：{guardrails.violated}。不建议发布。" ) 七、A/B 测试 Checklist □ 实验假设明确（自变量、因变量、预期效果） □ 样本量计算完成（MDE、power、alpha） □ 流量分配使用一致性哈希（同一用户体验一致） □ 护栏指标已定义并监控 □ 统计检验方法匹配指标类型（比例/连续） □ 序贯检验支持提前停止 □ LLM-as-Judge 评判使用多评判者 □ 实验报告自动生成 □ 决策建议基于数据而非直觉 □ 实验结果归档可追溯 结语 A/B 测试是 Agent 优化的科学基石。在 LLM 的非确定性世界里，直觉是不可靠的，只有统计检验才能区分真实效果和随机噪声。投资 A/B 测试平台不是开销，而是回报率最高的基础设施投资。让每一次 Prompt 修改、每一次模型升级都有数据支撑，这就是 Agent 工程的成熟标志。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
