Agent A/B 测试平台搭建：从实验设计到统计显著性

@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

@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

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),
        )

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%}）。"
        )

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)

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}。不建议发布。"
            )

□ 实验假设明确（自变量、因变量、预期效果）
□ 样本量计算完成（MDE、power、alpha）
□ 流量分配使用一致性哈希（同一用户体验一致）
□ 护栏指标已定义并监控
□ 统计检验方法匹配指标类型（比例/连续）
□ 序贯检验支持提前停止
□ LLM-as-Judge 评判使用多评判者
□ 实验报告自动生成
□ 决策建议基于数据而非直觉
□ 实验结果归档可追溯