上线了新的prompt版本或者换了模型之后，怎么知道效果是变好了还是变差了？靠人工评测太慢，靠直觉太不靠谱。我们团队的做法是：上线前先做AB测试，用真实流量对比新旧版本的效果，数据说话。

这篇文章分享LLM应用AB测试的完整方案，包括流量分配、指标采集、统计显著性检验、以及自动化评估流程。

为什么LLM应用需要专门的AB测试

传统Web应用的AB测试关注点击率、转化率这些硬指标，好衡量。LLM应用不一样，输出是自然语言，“好不好"很难量化。而且LLM的效果受prompt、模型、参数、上下文等多个因素影响，改了任何一个都可能影响最终效果。

需要关注的指标：

维度          指标                计算方式
质量          回答准确率           人工标注或自动评分
质量          格式合规率           正则匹配检查JSON/列表格式
效率          首Token延迟(TTFT)   P50/P95/P99
效率          生成速度(tokens/s)  平均值
成本          每次请求token消耗    输入+输出token数
体验          用户满意度           显式反馈或隐式信号

架构设计

核心是在请求入口加一个流量分配层，把用户随机分到A组或B组：

flowchart TD A[用户请求] --> B[流量分配器] B -->|50%| C[Control组] B -->|50%| D[Treatment组] C --> E[旧模型/旧Prompt] D --> F[新模型/新Prompt] E --> G[结果+指标采集] F --> G G --> H[数据存储] H --> I[统计分析]

流量分配

用用户ID做hash分流，保证同一个用户始终在同一个组：

import hashlib

class ABRouter:
    def __init__(self, experiments: dict):
        """
        experiments: {
            "model_upgrade": {"control": 0.5, "treatment": 0.5},
            "prompt_v2": {"control": 0.7, "treatment": 0.3}
        }
        """
        self.experiments = experiments

    def assign(self, user_id: str, experiment_name: str) -> str:
        """用hash分流，保证同一用户始终在同一组"""
        key = f"{user_id}:{experiment_name}"
        hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
        ratio = (hash_val % 10000) / 10000

        exp = self.experiments[experiment_name]
        if ratio < exp["control"]:
            return "control"
        return "treatment"

    def get_config(self, user_id: str) -> dict:
        """获取用户的所有实验配置"""
        config = {}
        for name in self.experiments:
            group = self.assign(user_id, name)
            config[name] = group
        return config

指标采集

在推理完成后记录关键指标：

import time
import json
from dataclasses import dataclass, asdict

@dataclass
class InferenceMetric:
    request_id: str
    user_id: str
    experiment_group: str
    model: str
    prompt_version: str
    input_tokens: int
    output_tokens: int
    ttft_ms: float
    total_ms: float
    tokens_per_sec: float
    format_valid: bool
    user_feedback: str = ""

class MetricsCollector:
    def __init__(self, storage_path="metrics.jsonl"):
        self.storage_path = storage_path

    def record(self, metric: InferenceMetric):
        with open(self.storage_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(metric), ensure_ascii=False) + "\n")

    def wrap_inference(self, func, user_id, experiment_group, model, prompt_version):
        """包装推理函数，自动采集指标"""
        def wrapper(prompt, **kwargs):
            request_id = f"req-{int(time.time()*1000)}"
            start = time.time()

            result = func(prompt, **kwargs)

            total_ms = (time.time() - start) * 1000
            output_tokens = len(result) // 4  # 粗略估算

            metric = InferenceMetric(
                request_id=request_id,
                user_id=user_id,
                experiment_group=experiment_group,
                model=model,
                prompt_version=prompt_version,
                input_tokens=len(prompt) // 4,
                output_tokens=output_tokens,
                ttft_ms=total_ms * 0.3,  # 粗略估算
                total_ms=total_ms,
                tokens_per_sec=output_tokens / (total_ms / 1000),
                format_valid=self._check_format(result)
            )
            self.record(metric)
            return result
        return wrapper

    def _check_format(self, text):
        """检查输出格式是否合规"""
        try:
            json.loads(text)
            return True
        except:
            return "[" in text or "{" in text

统计显著性检验

收集够数据后，要做统计检验才能下结论。常用的指标用t检验，比例类指标用卡方检验：

import numpy as np
from scipy import stats

class ABAnalyzer:
    def __init__(self, metrics_file="metrics.jsonl"):
        self.data = []
        with open(metrics_file, "r", encoding="utf-8") as f:
            for line in f:
                self.data.append(json.loads(line))

    def t_test(self, metric_name, experiment="model_upgrade"):
        """连续型指标的t检验"""
        control = [d[metric_name] for d in self.data
                   if d["experiment_group"] == "control"]
        treatment = [d[metric_name] for d in self.data
                    if d["experiment_group"] == "treatment"]

        t_stat, p_value = stats.ttest_ind(control, treatment)

        return {
            "metric": metric_name,
            "control_mean": np.mean(control),
            "treatment_mean": np.mean(treatment),
            "diff_pct": (np.mean(treatment) - np.mean(control)) / np.mean(control) * 100,
            "p_value": p_value,
            "significant": p_value < 0.05,
            "control_n": len(control),
            "treatment_n": len(treatment)
        }

    def chi2_test(self, metric_name, experiment="model_upgrade"):
        """比例型指标的卡方检验"""
        control = [d[metric_name] for d in self.data
                   if d["experiment_group"] == "control"]
        treatment = [d[metric_name] for d in self.data
                    if d["experiment_group"] == "treatment"]

        # 构建列联表
        table = np.array([
            [sum(control), len(control) - sum(control)],
            [sum(treatment), len(treatment) - sum(treatment)]
        ])

        chi2, p_value, _, _ = stats.chi2_contingency(table)

        return {
            "metric": metric_name,
            "control_rate": sum(control) / len(control),
            "treatment_rate": sum(treatment) / len(treatment),
            "p_value": p_value,
            "significant": p_value < 0.05
        }

    def report(self):
        """生成完整报告"""
        print("=" * 60)
        print("AB测试报告")
        print("=" * 60)

        # 连续型指标
        for metric in ["total_ms", "tokens_per_sec", "input_tokens"]:
            result = self.t_test(metric)
            sig = "***" if result["significant"] else "n.s."
            print(f"{metric}: {result['control_mean']:.1f} vs {result['treatment_mean']:.1f} "
                  f"({result['diff_pct']:+.1f}%) p={result['p_value']:.4f} {sig}")

        # 比例型指标
        for metric in ["format_valid"]:
            result = self.chi2_test(metric)
            sig = "***" if result["significant"] else "n.s."
            print(f"{metric}: {result['control_rate']:.1%} vs {result['treatment_rate']:.1%} "
                  f"p={result['p_value']:.4f} {sig}")

自动化评估流程

把AB测试做成CI/CD的一部分，每次改prompt或换模型都自动跑一遍：

def run_ab_test(new_prompt, old_prompt, test_cases):
    """自动化AB测试"""
    results = {"control": [], "treatment": []}

    for case in test_cases:
        # Control: 旧prompt
        old_result = inference(case["input"], old_prompt)
        results["control"].append({
            "input": case["input"],
            "output": old_result,
            "expected": case.get("expected"),
            "score": auto_score(old_result, case.get("expected"))
        })

        # Treatment: 新prompt
        new_result = inference(case["input"], new_prompt)
        results["treatment"].append({
            "input": case["input"],
            "output": new_result,
            "expected": case.get("expected"),
            "score": auto_score(new_result, case.get("expected"))
        })

    # 统计检验
    control_scores = [r["score"] for r in results["control"]]
    treatment_scores = [r["score"] for r in results["treatment"]]
    t_stat, p_value = stats.ttest_ind(control_scores, treatment_scores)

    return {
        "control_avg": np.mean(control_scores),
        "treatment_avg": np.mean(treatment_scores),
        "improvement": (np.mean(treatment_scores) - np.mean(control_scores)) / np.mean(control_scores) * 100,
        "p_value": p_value,
        "recommend": "ship" if p_value < 0.05 and np.mean(treatment_scores) > np.mean(control_scores) else "rollback"
    }

踩坑记录

样本量不够

一开始只测了100条就下结论，发现新版本"显著优于"旧版本。上线后真实流量跑了一周，效果反而变差了。原因是100条样本的置信度不够，碰巧那100条都是简单问题。

建议：每个组至少1000条数据，最好覆盖简单、中等、困难三种难度。

分流不均匀

用random.random()做分流，结果实验组和对照组的用户活跃度差异很大。改成用用户ID hash后解决了，同一个用户始终在同一个组。

指标定义不一致

团队里对"回答准确率"的定义不一样：有人觉得完全正确才算，有人觉得关键信息对就行。后来统一了评分标准，用自动化评分函数替代人工判断。

写在最后

LLM应用的AB测试比传统Web应用复杂，因为输出是非结构化的自然语言。但只要抓住"分流一致性、指标可量化、统计显著性"这三个核心，就能做出靠谱的对比实验。

建议先从离线AB测试开始（用固定测试集对比），跑通流程后再做在线AB测试（用真实流量对比）。