模型评估与 A/B 测试

为什么需要科学的模型评估

很多团队在微调完模型后，只做了几道手工测试题，感觉"效果好多了"，就兴冲冲地部署上线。结果上线后发现：在某些场景确实变好了，但在另一些场景反而变差了；用户的整体满意度并没有提升；甚至因为模型行为的变化，导致一些老用户投诉。这就是缺乏科学评估体系导致的典型问题。

科学的模型评估体系包含两个层面：离线评估（在部署前用标准化测试集评估，回答"模型在受控环境下表现如何"）和在线评估（部署后用A/B测试评估真实用户反馈，回答"模型在真实场景中表现如何"）。两个层面缺一不可——离线评估好不代表线上好（存在分布偏移），线上表现好但不知道具体好在哪也不利于持续优化。

离线评估指标设计

离线评估的核心是构建高质量的测试集。测试集应该：与训练集完全独立（不能有任何重叠）、覆盖所有重要的业务场景、每个场景有足够的样本量（至少50-100条）、标注质量高（每条数据有标准答案或参考标准）。评估指标根据任务类型选择：分类任务（意图识别）→ Accuracy、F1、混淆矩阵；生成任务（对话、写作）→ BLEU/ROUGE（参考价值有限）→ 更推荐LLM-as-Judge评分；检索任务→ Recall@k、MRR、NDCG。

import json, random, numpy as np
from scipy import stats
from openai import OpenAI

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")

class ModelEvaluator:
    def __init__(self):
        self.test_set = []
        self.results = {"baseline":[], "candidate":[]}

    def load_test_set(self, path):
        """加载测试集"""
        with open(path, 'r', encoding='utf-8') as f:
            self.test_set = json.load(f)
        print(f"加载 {len(self.test_set)} 条测试数据")

    def llm_judge(self, question, answer_a, answer_b, criteria):
        """LLM-as-Judge：让DeepSeek评判两个回答的优劣"""
        prompt = f"""你是一个公正的评审专家。请比较以下两个回答。

问题：{question}

回答A：{answer_a}

回答B：{answer_b}

评审标准：{criteria}

请以JSON格式输出：
{{
  "winner": "A" | "B" | "tie",
  "score_a": 1-10,
  "score_b": 1-10,
  "reasoning": "简短说明"
}}"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.1
        )
        return json.loads(response.choices[0].message.content)

    def evaluate_pairwise(self, baseline_model, candidate_model, num_samples=100):
        """成对评估：基座 vs 微调"""
        samples = random.sample(self.test_set, min(num_samples, len(self.test_set)))
        wins_a = wins_b = ties = 0
        for i, item in enumerate(samples):
            answer_a = baseline_model(item["instruction"], item.get("input",""))
            answer_b = candidate_model(item["instruction"], item.get("input",""))
            judgment = self.llm_judge(
                item["instruction"], answer_a, answer_b,
                item.get("criteria", "准确性、完整性、流畅性")
            )
            if judgment["winner"] == "A": wins_a += 1
            elif judgment["winner"] == "B": wins_b += 1
            else: ties += 1
            if (i+1) % 20 == 0:
                print(f"进度: {i+1}/{len(samples)}, A胜:{wins_a} B胜:{wins_b} 平:{ties}")
        return {"A_wins":wins_a, "B_wins":wins_b, "ties":ties, "win_rate_B":wins_b/(wins_a+wins_b+ties)}

class ABTestDesigner:
    def __init__(self):
        self.control_group = []
        self.treatment_group = []

    def calculate_sample_size(self, baseline_rate, expected_lift, alpha=0.05, power=0.8):
        """计算所需样本量"""
        from scipy.stats import norm
        z_alpha = norm.ppf(1 - alpha/2)
        z_beta = norm.ppf(power)
        p1 = baseline_rate
        p2 = baseline_rate * (1 + expected_lift)
        p_pooled = (p1 + p2) / 2
        n = (z_alpha * np.sqrt(2*p_pooled*(1-p_pooled)) + z_beta * np.sqrt(p1*(1-p1)+p2*(1-p2)))**2 / (p2-p1)**2
        return int(np.ceil(n))

    def check_significance(self, control_data, treatment_data, metric_name="satisfaction"):
        """检查统计显著性"""
        control = np.array(control_data)
        treatment = np.array(treatment_data)
        t_stat, p_value = stats.ttest_ind(treatment, control)
        effect = (treatment.mean() - control.mean()) / control.mean()
        ci = stats.t.interval(0.95, len(treatment)-1, loc=treatment.mean(), scale=stats.sem(treatment))
        return {
            "metric": metric_name,
            "control_mean": control.mean(),
            "treatment_mean": treatment.mean(),
            "relative_lift": f"{effect:.2%}",
            "p_value": p_value,
            "significant": p_value < 0.05,
            "ci_95": (ci[0], ci[1])
        }

    def run_ab_test(self, model_a, model_b, traffic_split=0.5, duration_days=7):
        """运行A/B测试"""
        print(f"A/B测试设计：")
        print(f"  对照组（A）：{model_a}")
        print(f"  实验组（B）：{model_b}")
        print(f"  流量分配：{traffic_split*100:.0f}%/{100-traffic_split*100:.0f}%")
        print(f"  建议运行时间：{duration_days} 天")
        n = self.calculate_sample_size(0.7, 0.05)
        print(f"  每组最少样本量：{n}")
        return {"status":"running", "expected_completion":f"{duration_days}天后"}

evaluator = ModelEvaluator()
evaluator.load_test_set("test_set.json")
# result = evaluator.evaluate_pairwise(baseline, finetuned, num_samples=100)

ab = ABTestDesigner()
n = ab.calculate_sample_size(baseline_rate=0.72, expected_lift=0.05)
print(f"需要每组 {n} 个样本才能检测到 5% 的提升")
# significance = ab.check_significance(control_scores, treatment_scores)

A/B测试的最佳实践

随机分流：使用用户ID的哈希值进行流量分配（如hash(user_id) % 100 < 50 → A组），确保两组用户特征分布一致。样本量计算：在开始实验前先计算所需样本量。如果预期提升只有3%-5%，通常需要每组5000-10000个样本才能得到统计显著的结论。样本量不足就贸然下结论是A/B测试最常见的错误。实验时长：至少运行一个完整的业务周期（通常1-2周），覆盖工作日和周末。太短的实验可能被"新奇效应"（用户对新模型感到好奇）或"星期效应"（周末用户行为与工作日不同）干扰。避免窥探：不要在实验中途频繁查看结果并提前终止。这会导致"窥探偏差"（peeking bias），使假阳性率从5%飙升到20%以上。设置实验时长后，到期再看结果。多指标综合判断：不要只看一个指标。如果你的微调模型在满意度上提升了3%但延迟增加了50%，你需要权衡这个trade-off。建议设置一个"护栏指标"（guardrail metrics）——如果核心业务指标（如转化率、留存率）出现显著下降，无论满意度如何都要暂停实验。

从评估到迭代

评估的最终目的不是给模型打分，而是找到下一步的优化方向。通过分析评估结果：哪些场景提升明显？（继续强化）哪些场景反而变差？（需要调整训练数据或策略）哪些场景没有变化？（数据可能不足或方法不对）将评估洞察转化为具体的优化任务，形成"微调→评估→分析→再微调"的持续改进循环。

A/B测试中的常见陷阱

A/B测试看似简单，实则充满陷阱。除了前面提到的窥探偏差和新奇效应，还有：样本污染问题——如果同一个用户可能在实验期间同时接触到A组和B组（比如在不同设备上），实验结论就会受污染。解决方法是使用稳定的用户标识（如账号ID而非设备ID）进行分流。时间效应干扰——如果在A/B测试期间发生了外部事件（如竞争对手推出新产品、行业政策变化），实验组和对照组可能受到同等影响，但难以从实验指标中排除。解决方法是使用A/A测试作为对照组。多重比较问题——如果你同时观察10个指标，即使没有实际效果，也会有一个指标"碰巧"达到统计显著。解决方法是对p值进行Bonferroni校正，或预先注册（preregister）你关心的主要指标。长期效果评估：大多数A/B测试只持续1-2周，只能观察到短期效果。但许多AI改进的效果需要更长时间才能显现——比如个性化能力的提升需要模型与用户多轮交互才能体现。建议对一些重要的模型变更设置"holdout组"——一小部分用户长期使用旧版本模型，用于评估长期效果。同时，建立"反向指标"监控——微调模型可能在短期内提升了某指标，但长期来看导致了用户倦怠或使用频率下降。AI产品的成功最终取决于用户留存和长期价值，而非短期的指标优化。

实验文化与管理

建立健康的实验文化比掌握实验技术更重要。核心原则：实验结论由数据驱动而非意见驱动（"我觉得这个模型更好"不如"数据显示满意度提升3.2%，p=0.02"）；失败的实验也是成功（知道什么方法不work和有work同样有价值）；鼓励提出大胆的假设并用严谨的实验验证；建立实验知识库——记录每次实验的假设、设计、结果和收获，避免重复踩坑。每月进行一次实验回顾（Experiment Review），分享上个月的实验成果和教训，逐步积累团队的实验智慧。