当前位置：首页 > news >正文

AI产品A/B测试的实验设计：分流策略、指标选择与统计显著性的工程化实现

news 2026/7/23 7:29:01

AI产品A/B测试的实验设计：分流策略、指标选择与统计显著性的工程化实现

一、AI产品的A/B测试独特性：输出不是"蓝色按钮或红色按钮"

传统互联网产品的A/B测试对比的是UI元素（按钮颜色、布局），输出是可精确度量的（点击率、转化率）。AI产品的A/B测试对比的是模型版本、Prompt策略或Agent工作流，输出是"生成质量"这样的模糊概念。如何将"这篇日记写得好不好"转化为可量化的指标，是AI产品A/B测试的核心挑战。

二、实验设计的三层架构

分流层使用一致性哈希确保同一用户始终分到同一组（避免体验不一致）。指标采集层分离了三类指标：客观系统指标（延迟、成本）、用户行为指标（编辑次数、复制率）和质量感知指标（评分、次日留存）。统计分析层使用独立样本t检验判断组间差异是否显著。

三、分流策略与关键实现

# ab_testing/splitter.py """用户分流引擎 设计意图： 1. 使用一致性哈希确保同一用户ID始终分到同一实验组 2. 支持多实验并行，哈希空间互不重叠 3. 流量分配通过实验配置动态调整，无需重启服务 """ import hashlib from dataclasses import dataclass from enum import Enum from typing import Optional class Variant(Enum): CONTROL = "control" # 对照组 TREATMENT_A = "treatment_a" # 实验组A TREATMENT_B = "treatment_b" # 实验组B @dataclass class ExperimentConfig: experiment_id: str variants: dict[Variant, float] # 各变体的流量占比 salt: str # 哈希盐值，确保不同实验独立 min_sample_size: int # 最小样本量 def total_ratio(self) -> float: return sum(self.variants.values()) class UserSplitter: """一致性哈希分流器""" # 哈希环槽位数，2^14=16384的精度足以分配百分比级流量 RING_SIZE = 16384 def assign(self, user_id: str, config: ExperimentConfig) -> Optional[Variant]: """返回用户被分配到哪个实验变体""" # 使用MD5生成分布式哈希 hash_input = f"{config.salt}:{user_id}" hash_bytes = hashlib.md5(hash_input.encode()).digest() # 取前2字节映射到0-RING_SIZE范围 bucket = int.from_bytes(hash_bytes[:2], 'big') % self.RING_SIZE # 按累计占比确定bucket落在哪个变体的区间 cumulative = 0 total = config.total_ratio() for variant, ratio in config.variants.items(): cumulative += ratio # 将累计占比映射为hash区间 boundary = int(self.RING_SIZE * cumulative / total) if bucket < boundary: return variant return None # 未被分配（如流量占比<100%时剩余的用户） def should_collect(self, config: ExperimentConfig) -> bool: """判断是否还需收集数据（样本量是否达到要求）""" # 实际实现中从数据库读取当前各变体的样本量 return True # ab_testing/metrics.py """实验指标计算 指标分为三类： 1. 系统指标：延迟、Token消耗（客观、有标准答案） 2. 行为指标：编辑次数、分享率（客观、无标准答案） 3. 感知指标：用户评分、次日留存（主观、收集成本高） """ from scipy import stats import numpy as np class ExperimentAnalyzer: def analyze( self, control_values: list[float], treatment_values: list[float], alpha: float = 0.05, ) -> dict: """独立样本t检验分析两组差异显著性""" t_stat, p_value = stats.ttest_ind( treatment_values, control_values, equal_var=False ) # 计算效应量（Cohen's d） pooled_std = np.sqrt( (np.std(control_values, ddof=1) ** 2 + np.std(treatment_values, ddof=1) ** 2) / 2 ) cohens_d = ( (np.mean(treatment_values) - np.mean(control_values)) / pooled_std if pooled_std > 0 else 0 ) return { "t_statistic": float(t_stat), "p_value": float(p_value), "significant": p_value < alpha, "cohens_d": float(cohens_d), "effect_size": self._classify_effect(cohens_d), "control_mean": float(np.mean(control_values)), "treatment_mean": float(np.mean(treatment_values)), "relative_change": float( (np.mean(treatment_values) - np.mean(control_values)) / max(np.mean(control_values), 0.001) ), } def _classify_effect(self, d: float) -> str: """按Cohen's d分类效应大小""" if abs(d) < 0.2: return "可忽略" elif abs(d) < 0.5: return "小" elif abs(d) < 0.8: return "中" else: return "大"

一致性哈希保证了用户ID与变体的稳定映射——同一用户不会在一次对话中看到新旧两个版本的AI输出，避免混淆感知。