A/B 评测架构：流量分流不是随机，要在统计意义上等价 A/B 评测架构流量分流不是随机要在统计意义上等价一、个性化深度引言上线了一个新版本的文本分类模型A/B 测试跑了三天。新版模型的准确率指标比旧版高了 2.3 个百分点。看起来很明显。但深入分析流量分组后发现新版模型分配到的测试样本中短文本占比 68%旧版只有 41%。短文本天然更容易分类——这 2.3 个百分点的提升有 1.5 个是该流量歧视造成的。见证奇迹的时刻不是发现了这个统计偏差。而是意识到A/B 评测的流量分流核心不在于分成两半而在于两组流量的数据分布在统计意义上等价。这是评测可信度的地基。二、个性化原理剖析A/B 评测的流量分流需要满足三个等价性条件样本量等价两组流量规模相等或按预设比例分布等价两组样本的特征分布一致文本长度、主题、语言等时序等价两组样本在同时间段采集避免时间性偏差如白天/夜间用户行为差异关键设计一致性哈希分流使用hash(request_id) % N做流量分组同一请求多次调用路由到同一组保证用户体验一致。分层抽样补偿实时监控两组样本的特征分布出现偏差时通过配额调整补全。统计显著性检验收集足够样本后执行 t 检验或 bootstrapp 0.05 才认为差异可信。三、个性化代码实践import hashlib import json import math import random from collections import defaultdict from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple dataclass class TrafficSample: 设计原因流量样本的统一数据结构。 features 存储多维特征用于分布校验。 request_id: str text: str features: Dict[str, float] # {length: 120, difficulty: 0.7, ...} group: str # A or B label: Optional[str] None prediction: Optional[str] None timestamp: float 0.0 class ABTrafficSplitter: 设计原因流量分流器。 核心一致性哈希保证稳定性 分层补偿保证公平性。 def __init__( self, a_ratio: float 0.5, feature_dims: List[str] None, max_imbalance_threshold: float 0.1, ): self.a_ratio a_ratio self.feature_dims feature_dims or [ text_length, difficulty ] self.max_imbalance_threshold max_imbalance_threshold # 设计原因维护两组样本的特征分布统计 # 用于实时检测和补偿分布偏差。 self._a_stats: Dict[str, List[float]] defaultdict(list) self._b_stats: Dict[str, List[float]] defaultdict(list) def assign_group(self, request_id: str, features: Dict[str, float]) - str: 设计原因默认使用一致性哈希分流。 同一 request_id 永远路由到同一组保证用户体验一致。 hash_val int( hashlib.md5(request_id.encode()).hexdigest(), 16 ) normalized (hash_val % 10000) / 10000.0 return A if normalized self.a_ratio else B def assign_group_with_compensation( self, request_id: str, features: Dict[str, float] ) - str: 设计原因带分层补偿的分流。 检测到某维度分布偏差时通过配额调整做补偿。 # 设计原因先检查当前各组分布是否均衡。 imbalance self._check_imbalance(features) if imbalance: # 设计原因分布不均衡时优先分配给样本量少的组。 return A if self._need_more_a(features) else B else: # 设计原因分布均衡时使用一致性哈希。 return self.assign_group(request_id, features) def _check_imbalance(self, features: Dict[str, float]) - bool: 设计原因检查当前两组在关键维度上是否分布均衡。 任一维度偏差超过阈值即认为需要补偿。 if (len(self._a_stats.get(text_length, [])) 10 or len(self._b_stats.get(text_length, [])) 10): return False # 设计原因样本太少时不检查。 for dim in self.feature_dims: if dim not in features: continue a_vals self._a_stats.get(dim, []) b_vals self._b_stats.get(dim, []) if not a_vals or not b_vals: continue a_mean sum(a_vals) / len(a_vals) b_mean sum(b_vals) / len(b_vals) # 设计原因归一化差值适应不同量级的特征。 max_val max(abs(a_mean), abs(b_mean), 1e-6) diff abs(a_mean - b_mean) / max_val if diff self.max_imbalance_threshold: return True return False def _need_more_a(self, features: Dict[str, float]) - bool: 设计原因判断是否A组需要更多此类样本。 a_count len(self._a_stats.get(text_length, [])) b_count len(self._b_stats.get(text_length, [])) return a_count b_count def record_sample(self, sample: TrafficSample) - None: 设计原因记录已分配的样本特征供后续分布校验使用。 target ( self._a_stats if sample.group A else self._b_stats ) for dim, value in sample.features.items(): target[dim].append(value) def get_distribution_report(self) - Dict: 设计原因生成分布报告供人工审查两组是否等价。 report {} for dim in self.feature_dims: a_vals self._a_stats.get(dim, []) b_vals self._b_stats.get(dim, []) report[dim] { A_count: len(a_vals), B_count: len(b_vals), A_mean: sum(a_vals) / len(a_vals) if a_vals else 0, B_mean: sum(b_vals) / len(b_vals) if b_vals else 0, } return report class ABStatTest: 设计原因统计显著性检验。 使用 Welchs t-test 处理两组方差不等的场景。 staticmethod def welch_t_test( a_scores: List[float], b_scores: List[float] ) - Tuple[float, float]: 设计原因Welchs t-test 不假设两组方差相等 比 Students t-test 更适合真实场景的评测数据。 n_a, n_b len(a_scores), len(b_scores) mean_a sum(a_scores) / n_a mean_b sum(b_scores) / n_b var_a sum((x - mean_a) ** 2 for x in a_scores) / (n_a - 1) var_b sum((x - mean_b) ** 2 for x in b_scores) / (n_b - 1) se math.sqrt(var_a / n_a var_b / n_b) t_stat (mean_a - mean_b) / se if se 0 else 0 # 设计原因Welch-Satterthwaite 公式计算自由度。 num (var_a / n_a var_b / n_b) ** 2 denom ((var_a / n_a) ** 2 / (n_a - 1) (var_b / n_b) ** 2 / (n_b - 1)) df num / denom if denom 0 else 1 # 设计原因双尾检验的简化 p-value 计算。 p_value 2 * (1 - ABStatTest._t_cdf(abs(t_stat), df)) return t_stat, p_value staticmethod def _t_cdf(x: float, df: float, steps: int 1000) - float: 设计原因t 分布 CDF 的数值近似。 if df 1: return 0.5 # 简化实现 return 0.5 0.5 * math.erf(x / math.sqrt(2)) staticmethod def is_significant( a_scores: List[float], b_scores: List[float], alpha: float 0.05 ) - Dict: 设计原因返回完整的统计检验结果 包括 t 统计量、p 值、是否显著。 t_stat, p_value ABStatTest.welch_t_test(a_scores, b_scores) return { t_statistic: t_stat, p_value: p_value, significant: p_value alpha, alpha: alpha, n_a: len(a_scores), n_b: len(b_scores), mean_a: sum(a_scores) / len(a_scores), mean_b: sum(b_scores) / len(b_scores), }四、个性化边界权衡1. 一致性哈希 vs 随机分流一致性哈希保证用户会话内一致方便问题定位。随机分流流量分布更均匀但用户可能在不同请求间看到不同版本。推荐默认一致性哈希随机分流作为选项。2. 实时补偿 vs 离线重采样实时补偿调整反馈快但可能引入补偿偏差补偿样本不一定代表真实分布。离线重采样更精确但需要事后分析。建议实时轻度补偿阈值 0.2 事后重采样报告。3. 单维检验 vs 多维联合检验单维检验只比准确率简单但可能遗漏隐藏偏差。多维联合检验准确率 延迟 多样性更全面但解读复杂。推荐核心指标做显著性检验辅助指标做趋势监控。4. 固定样本量 vs 序贯检验固定样本量如 10000 条便于规划但可能收集了不必要的样本量。序贯检验每 500 条检查一次更高效但多次检验需要 Bonferroni 校正。建议序贯检验 alpha 校正。5. 自动决策 vs 人工审批自动决策p 0.05 即上线效率高但可能误判。人工审批增加审核环节但延迟上线。推荐自动标记统计显著 人工确认上线核心模型增加 launch review。五、总结A/B 评测的流量分流核心要求是两组样本在统计意义上等价。通过一致性哈希保证用户体验一致、分层抽样补偿特征分布偏差、Welchs t-test 检验统计显著性构建可信的评测基础设施。工程实践中需要在分流策略、补偿时机、检验方法上做权衡核心原则是流量分组不是分成两半而是保证两组在除去模型差异外所有其他因素等价。