A/B 测试：模型效果验证

A/B 测试：模型效果验证

1. 技术分析

1.1 A/B 测试原理

A/B 测试是对比不同模型版本效果的方法：

A/B 测试流程 分流 → 实验 → 收集数据 → 统计分析 → 决策

1.2 A/B 测试类型

1.3 统计检验方法

统计检验方法 t检验: 均值差异 卡方检验: 比例差异 置信区间: 估计范围

2. 核心功能实现

2.1 分流器

import random import hashlib class TrafficSplitter: def __init__(self, splits): self.splits = splits self.total = sum(splits.values()) def split(self, user_id): hash_val = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16) random.seed(hash_val) r = random.random() * self.total cumulative = 0 for variant, weight in self.splits.items(): cumulative += weight if r < cumulative: return variant return list(self.splits.keys())[-1] class ConsistentSplitter: def __init__(self, variants, weights): self.variants = variants self.weights = weights self.cumulative_weights = [sum(weights[:i+1]) for i in range(len(weights))] def split(self, user_id): hash_val = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16) normalized = hash_val / (2**128 - 1) for i, variant in enumerate(self.variants): if normalized < self.cumulative_weights[i]: return variant return self.variants[-1] class DynamicSplitter: def __init__(self): self.variants = {} def add_variant(self, name, weight): self.variants[name] = weight def update_weight(self, name, weight): if name in self.variants: self.variants[name] = weight def split(self, user_id): total = sum(self.variants.values()) hash_val = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16) r = (hash_val % (total * 100)) / 100 cumulative = 0 for variant, weight in self.variants.items(): cumulative += weight if r < cumulative: return variant return list(self.variants.keys())[-1]

2.2 实验管理器

import pandas as pd class ExperimentManager: def __init__(self, name): self.name = name self.variants = {} self.results = [] def add_variant(self, name, model): self.variants[name] = model def run_experiment(self, data, labels): for variant_name, model in self.variants.items(): predictions = model.predict(data) accuracy = (predictions == labels).mean() self.results.append({ 'variant': variant_name, 'accuracy': accuracy, 'sample_size': len(data) }) def get_results(self): return pd.DataFrame(self.results) class OnlineExperimentManager: def __init__(self, name): self.name = name self.variant_data = {} def log_prediction(self, variant, prediction, actual): if variant not in self.variant_data: self.variant_data[variant] = {'correct': 0, 'total': 0} self.variant_data[variant]['total'] += 1 if prediction == actual: self.variant_data[variant]['correct'] += 1 def get_results(self): results = [] for variant, data in self.variant_data.items(): accuracy = data['correct'] / data['total'] if data['total'] > 0 else 0 results.append({ 'variant': variant, 'accuracy': accuracy, 'sample_size': data['total'] }) return pd.DataFrame(results)

2.3 统计分析

from scipy import stats import numpy as np class StatisticalAnalyzer: def __init__(self): pass def t_test(self, variant1_data, variant2_data): t_stat, p_value = stats.ttest_ind(variant1_data, variant2_data) return { 't_statistic': t_stat, 'p_value': p_value, 'significant': p_value < 0.05 } def chi_square_test(self, variant1_correct, variant1_total, variant2_correct, variant2_total): observed = [ [variant1_correct, variant1_total - variant1_correct], [variant2_correct, variant2_total - variant2_correct] ] chi_stat, p_value, _, _ = stats.chi2_contingency(observed) return { 'chi_statistic': chi_stat, 'p_value': p_value, 'significant': p_value < 0.05 } def calculate_confidence_interval(self, proportion, sample_size, confidence=0.95): z = stats.norm.ppf((1 + confidence) / 2) margin_of_error = z * np.sqrt((proportion * (1 - proportion)) / sample_size) return { 'lower_bound': proportion - margin_of_error, 'upper_bound': proportion + margin_of_error, 'confidence_level': confidence } def analyze_experiment(self, results_df): control = results_df[results_df['variant'] == 'control'] treatment = results_df[results_df['variant'] == 'treatment'] if len(control) == 0 or len(treatment) == 0: return None control_acc = control['accuracy'].values[0] treatment_acc = treatment['accuracy'].values[0] control_n = control['sample_size'].values[0] treatment_n = treatment['sample_size'].values[0] control_correct = int(control_acc * control_n) treatment_correct = int(treatment_acc * treatment_n) chi_result = self.chi_square_test(control_correct, control_n, treatment_correct, treatment_n) control_ci = self.calculate_confidence_interval(control_acc, control_n) treatment_ci = self.calculate_confidence_interval(treatment_acc, treatment_n) return { 'control_accuracy': control_acc, 'treatment_accuracy': treatment_acc, 'lift': treatment_acc - control_acc, 'chi_test': chi_result, 'control_ci': control_ci, 'treatment_ci': treatment_ci }

3. 性能对比

3.1 分流方法对比

3.2 统计检验对比

3.3 样本量计算

4. 最佳实践

4.1 A/B 测试流程

def run_ab_test(config): splitter = TrafficSplitter(config['splits']) experiment_manager = OnlineExperimentManager(config['name']) while True: user_id = get_next_user() data = get_user_data(user_id) variant = splitter.split(user_id) model = config['models'][variant] prediction = model.predict(data) actual = get_actual_result(user_id) experiment_manager.log_prediction(variant, prediction, actual) results = experiment_manager.get_results() if should_stop_test(results): break analyzer = StatisticalAnalyzer() analysis = analyzer.analyze_experiment(results) return analysis class ABTestWorkflow: def __init__(self, config): self.config = config self.splitter = TrafficSplitter(config['splits']) self.manager = OnlineExperimentManager(config['name']) self.analyzer = StatisticalAnalyzer() def run(self): while True: user_id = input("Enter user ID: ") features = input("Enter features: ").split() variant = self.splitter.split(user_id) model = self.config['models'][variant] prediction = model.predict([features])[0] actual = input("Enter actual result: ") self.manager.log_prediction(variant, prediction, actual) results = self.manager.get_results() print(results) if len(results) > 0 and all(r['sample_size'] > 100 for _, r in results.iterrows()): analysis = self.analyzer.analyze_experiment(results) print("Analysis:", analysis) if analysis['chi_test']['significant']: print(f"Variant {analysis['treatment_accuracy'] > analysis['control_accuracy'] and 'treatment' or 'control'} is better!") break

4.2 实验监控

class ExperimentMonitor: def __init__(self, experiment_manager): self.manager = experiment_manager def check_sample_size(self, min_sample_size=1000): results = self.manager.get_results() for _, row in results.iterrows(): if row['sample_size'] < min_sample_size: return False return True def check_significance(self): analyzer = StatisticalAnalyzer() results = self.manager.get_results() if len(results) < 2: return False analysis = analyzer.analyze_experiment(results) return analysis['chi_test']['significant'] def should_stop(self): return self.check_sample_size() and self.check_significance()

5. 总结

A/B 测试是验证模型效果的科学方法：

1. 分流策略：确保用户分配的公平性和一致性
2. 实验管理：收集和管理实验数据
3. 统计分析：使用统计方法验证差异
4. 决策依据：基于数据做出决策

对比数据如下：

• 一致性分流确保用户始终看到相同版本
• 卡方检验适合分类数据的对比
• 需要足够的样本量才能检测到小差异
• 推荐使用在线实验管理器进行实时监控