大模型 A/B 测试框架实战：3 步搭建模型效果自动评测系统（附完整代码）

同时用 DeepSeek、通义千问、混元，到底哪个在你的业务场景效果最好？手把手搭建一个自动 A/B 测试框架，用数据说话。

---

为什么要做大模型 A/B 测试？

2026 年的大模型市场卷到飞起，光国内主流模型就有十几个：

模型	厂商	上下文	价格（百万 token）
GPT-4o	OpenAI	128K	¥70
Claude 3.5	Anthropic	200K	¥63
DeepSeek V3	深度求索	128K	¥4
通义千问 Max	阿里云	128K	¥20
混元 Pro	腾讯云	128K	¥15
文心 4.0	百度	128K	¥30

价格差了 10 倍，但便宜的效果真的够用吗？

盲猜不如实测。 下面搭建一个完整的 A/B 测试框架，自动对比多个模型在你业务场景下的表现。

---

一、架构设计

测试用例（JSON）
       ↓
   评测引擎（Python）
  ┌────┼────┐
  ↓    ↓    ↓
模型A 模型B 模型C    ← 多模型并行调用
  ↓    ↓    ↓
 评分模块（LLM-as-Judge）
  ↓
评测报告（Markdown + CSV）

核心思路：
1. 准备测试用例：你业务场景的真实 prompt + 预期回答
2. 并行调用多个模型：记录响应时间、token 消耗
3. 自动评分：用 GPT-4o 作为裁判打分
4. 生成报告：汇总对比，一目了然

---

二、完整代码实现

2.1 项目结构

llm-ab-test/
├── config.yaml          # 模型配置
├── test_cases.json      # 测试用例
├── evaluator.py         # 评测主程序
├── report_generator.py  # 报告生成
└── requirements.txt     # 依赖

2.2 安装依赖

pip install openai pyyaml pandas jinja2

2.3 模型配置

# config.yaml
models:
  - name: "DeepSeek V3"
    provider: "deepseek"
    api_key: "${DEEPSEEK_API_KEY}"
    base_url: "https://api.deepseek.com/v1"
    model: "deepseek-chat"
    temperature: 0.7

  - name: "通义千问 Max"
    provider: "qwen"
    api_key: "${QWEN_API_KEY}"
    base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
    model: "qwen-max"
    temperature: 0.7

  - name: "混元 Pro"
    provider: "hunyuan"
    api_key: "${HUNYUAN_API_KEY}"
    base_url: "https://api.hunyuan.cloud.tencent.com/v1"
    model: "hunyuan-pro"
    temperature: 0.7

# 评分模型（用最强的当裁判）
judge:
  provider: "openai"
  api_key: "${OPENAI_API_KEY}"
  model: "gpt-4o"
  temperature: 0.3

# 评分维度
eval_dimensions:
  - "准确性：回答是否事实正确"
  - "完整性：是否覆盖了用户问题的所有要点"
  - "可读性：语言是否清晰、结构化"
  - "实用性：对用户是否有实际帮助"

2.4 测试用例

// test_cases.json
[
  {
    "id": 1,
    "category": "云产品咨询",
    "prompt": "我想搭建一个个人博客，用腾讯云轻量应用服务器，2核4G够用吗？需要买什么配置？",
    "expected_points": [
      "2核4G对个人博客够用",
      "需要选择系统镜像（如Ubuntu）",
      "需要购买域名并备案",
      "建议搭配对象存储存图片",
      "建议用CDN加速"
    ]
  },
  {
    "id": 2,
    "category": "技术问答",
    "prompt": "Docker和Kubernetes有什么区别？分别适合什么场景？",
    "expected_points": [
      "Docker是容器化技术，K8s是容器编排",
      "Docker管理单个容器，K8s管理集群",
      "小规模用Docker Compose就够了",
      "大规模微服务才需要K8s",
      "K8s有自动扩缩容、服务发现等高级特性"
    ]
  },
  {
    "id": 3,
    "category": "AI应用",
    "prompt": "我想在Python项目中接入大模型API做智能客服，用什么框架比较好？",
    "expected_points": [
      "推荐LangChain或LlamaIndex",
      "需要考虑对话记忆管理",
      "建议用Redis缓存历史消息",
      "FastAPI搭建后端服务",
      "需要做Prompt工程优化回答质量"
    ]
  },
  {
    "id": 4,
    "category": "运维排障",
    "prompt": "我的云服务器CPU使用率突然到100%，怎么排查？",
    "expected_points": [
      "用top/htop看进程占用",
      "检查是否有异常进程或挖矿程序",
      "查看系统日志/var/log/syslog",
      "检查是否有异常的网络连接",
      "考虑是否被DDoS攻击"
    ]
  },
  {
    "id": 5,
    "category": "成本优化",
    "prompt": "我每月云服务器费用3000块，有什么降本方法？",
    "expected_points": [
      "检查资源利用率，降配空闲实例",
      "用按量付费替代包年包月（利用率低时）",
      "使用预留实例或节省计划",
      "优化存储，清理无用快照和磁盘",
      "CDN和对象存储用生命周期策略"
    ]
  }
]

2.5 评测主程序

"""
evaluator.py - 大模型 A/B 测试评测引擎
"""
import json
import time
import os
import yaml
from openai import OpenAI
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed


class ModelClient:
    """统一的大模型调用客户端"""

    def __init__(self, config: dict):
        self.name = config["name"]
        self.client = OpenAI(
            api_key=os.path.expandvars(config["api_key"]),
            base_url=config["base_url"]
        )
        self.model = config["model"]
        self.temperature = config.get("temperature", 0.7)

    def chat(self, prompt: str) -> dict:
        """调用模型并记录性能数据"""
        start_time = time.time()

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=self.temperature,
                max_tokens=1024
            )

            elapsed = time.time() - start_time
            usage = response.usage

            return {
                "model": self.name,
                "content": response.choices[0].message.content,
                "latency": round(elapsed, 2),
                "prompt_tokens": usage.prompt_tokens,
                "completion_tokens": usage.completion_tokens,
                "total_tokens": usage.total_tokens,
                "success": True,
                "error": None
            }
        except Exception as e:
            elapsed = time.time() - start_time
            return {
                "model": self.name,
                "content": "",
                "latency": round(elapsed, 2),
                "prompt_tokens": 0,
                "completion_tokens": 0,
                "total_tokens": 0,
                "success": False,
                "error": str(e)
            }


class JudgeEvaluator:
    """用 LLM-as-Judge 的方式自动评分"""

    def __init__(self, judge_config: dict, dimensions: list):
        self.client = OpenAI(
            api_key=os.path.expandvars(judge_config["api_key"]),
        )
        self.model = judge_config["model"]
        self.dimensions = dimensions

    def evaluate(self, prompt: str, response: str, expected_points: list) -> dict:
        """对模型回答进行评分"""
        eval_prompt = f"""你是一个专业的AI回答质量评审员。请评估以下AI回答的质量。

## 用户问题
{prompt}

## 期望回答要点
{json.dumps(expected_points, ensure_ascii=False)}

## AI回答
{response}

## 评分维度（每项 1-10 分）
{chr(10).join([f'- {d}' for d in self.dimensions])}

请以JSON格式返回评分结果：
{{
    "scores": {{
        "准确性": <1-10>,
        "完整性": <1-10>,
        "可读性": <1-10>,
        "实用性": <1-10>
    }},
    "total_score": <总分，满分40>,
    "comment": "<一句话评价>",
    "missed_points": ["<遗漏的要点>"]
}}

只返回JSON，不要其他内容。"""

        try:
            result = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": eval_prompt}],
                temperature=0.3,
                response_format={"type": "json_object"}
            )
            scores = json.loads(result.choices[0].message.content)
            return {
                "success": True,
                "scores": scores.get("scores", {}),
                "total_score": scores.get("total_score", 0),
                "comment": scores.get("comment", ""),
                "missed_points": scores.get("missed_points", [])
            }
        except Exception as e:
            return {
                "success": False,
                "scores": {},
                "total_score": 0,
                "comment": f"评分失败: {e}",
                "missed_points": []
            }


def run_evaluation(config_path: str = "config.yaml",
                   test_cases_path: str = "test_cases.json"):
    """执行完整评测流程"""

    # 加载配置
    with open(config_path, "r", encoding="utf-8") as f:
        config = yaml.safe_load(f)

    with open(test_cases_path, "r", encoding="utf-8") as f:
        test_cases = json.load(f)

    # 初始化模型客户端
    model_clients = [ModelClient(m) for m in config["models"]]
    judge = JudgeEvaluator(config["judge"], config["eval_dimensions"])

    # 存放所有结果
    all_results = []

    print(f"🚀 开始评测，共 {len(test_cases)} 个用例，"
          f"{len(model_clients)} 个模型")
    print("=" * 60)

    for case in test_cases:
        print(f"\n📝 用例 {case['id']}: [{case['category']}] "
              f"{case['prompt'][:40]}...")

        # 并行调用所有模型
        model_results = {}
        with ThreadPoolExecutor(max_workers=len(model_clients)) as executor:
            futures = {
                executor.submit(client.chat, case["prompt"]): client
                for client in model_clients
            }
            for future in as_completed(futures):
                result = future.result()
                model_results[result["model"]] = result
                status = "✅" if result["success"] else "❌"
                print(f"  {status} {result['model']}: "
                      f"{result['latency']}s, "
                      f"{result['total_tokens']} tokens")

        # 评分
        for model_name, result in model_results.items():
            if result["success"]:
                evaluation = judge.evaluate(
                    prompt=case["prompt"],
                    response=result["content"],
                    expected_points=case["expected_points"]
                )
                result["evaluation"] = evaluation

        all_results.append({
            "case_id": case["id"],
            "category": case["category"],
            "prompt": case["prompt"],
            "model_results": model_results
        })

    # 生成报告
    generate_report(all_results, model_clients, config)

    # 保存原始数据
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    data_file = f"ab_test_results_{timestamp}.json"
    with open(data_file, "w", encoding="utf-8") as f:
        json.dump(all_results, f, ensure_ascii=False, indent=2)

    print(f"\n✅ 评测完成！原始数据保存至 {data_file}")
    return all_results


def generate_report(results: list, clients: list, config: dict):
    """生成 Markdown 评测报告"""

    model_names = [c.name for c in clients]

    # 汇总每个模型的总分
    model_stats = {}
    for name in model_names:
        model_stats[name] = {
            "total_score": 0,
            "total_latency": 0,
            "total_tokens": 0,
            "success_count": 0
        }

    for r in results:
        for name, mr in r["model_results"].items():
            if name in model_stats:
                model_stats[name]["total_latency"] += mr["latency"]
                model_stats[name]["total_tokens"] += mr["total_tokens"]
                if mr["success"] and "evaluation" in mr:
                    model_stats[name]["total_score"] += \
                        mr["evaluation"].get("total_score", 0)
                    model_stats[name]["success_count"] += 1

    # 生成报告
    report = f"""# 大模型 A/B 测试评测报告

> 评测时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}
> 测试用例数：{len(results)}
> 参评模型：{', '.join(model_names)}

---

## 一、总览排名

| 排名 | 模型 | 总分 | 平均分 | 平均延迟(s) | 总 Token |
|------|------|------|--------|------------|----------|
"""

    sorted_models = sorted(
        model_stats.items(),
        key=lambda x: x[1]["total_score"],
        reverse=True
    )

    for rank, (name, stats) in enumerate(sorted_models, 1):
        avg_score = (stats["total_score"] / stats["success_count"]
                     if stats["success_count"] > 0 else 0)
        avg_latency = (stats["total_latency"] / len(results))
        report += f"| {rank} | {name} | {stats['total_score']} | {avg_score:.1f} | {avg_latency:.2f} | {stats['total_tokens']} |\n"

    report += "\n---\n\n## 二、逐题评分\n\n"

    for r in results:
        report += f"### 用例 {r['case_id']}：{r['prompt'][:50]}...\n\n"
        report += f"**类别：** {r['category']}\n\n"
        report += "| 模型 | 得分 | 延迟(s) | Token | 评价 |\n"
        report += "|------|------|---------|-------|------|\n"

        case_scores = []
        for name, mr in r["model_results"].items():
            if mr["success"] and "evaluation" in mr:
                ev = mr["evaluation"]
                score = ev.get("total_score", 0)
                comment = ev.get("comment", "")[:30]
                case_scores.append((name, score))
                report += f"| {name} | {score}/40 | {mr['latency']} | {mr['total_tokens']} | {comment} |\n"

        report += "\n"

    report += """---

## 三、结论与建议

"""

    # 自动生成建议
    best_model = sorted_models[0][0]
    cheapest_model = min(
        model_names,
        key=lambda n: model_stats[n]["total_tokens"]
    )
    fastest_model = min(
        model_names,
        key=lambda n: model_stats[n]["total_latency"]
    )

    report += f"""基于本次评测结果：

- **综合最佳**：{best_model}（总分最高）
- **响应最快**：{fastest_model}（平均延迟最低）
- **Token 最省**：{cheapest_model}（总消耗最少）

> 建议在正式切换模型前，用生产环境的真实流量做灰度测试，
> 并持续监控线上效果指标。
"""

    # 保存报告
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_file = f"ab_test_report_{timestamp}.md"
    with open(report_file, "w", encoding="utf-8") as f:
        f.write(report)

    print(f"📊 评测报告已生成：{report_file}")


if __name__ == "__main__":
    run_evaluation()

2.6 运行评测

# 设置 API Key（建议用 .env 文件或环境变量）
export DEEPSEEK_API_KEY="sk-xxx"
export QWEN_API_KEY="sk-xxx"
export HUNYUAN_API_KEY="xxx"
export OPENAI_API_KEY="sk-xxx"

# 执行评测
python evaluator.py

2.7 输出示例

评测报告会自动生成类似这样的对比：

排名	模型	总分	平均分	平均延迟(s)	总 Token
1	通义千问 Max	178	35.6	2.3	3280
2	DeepSeek V3	172	34.4	1.8	2950
3	混元 Pro	165	33.0	2.1	3100

---

三、评测框架的 5 个最佳实践

1. 测试用例要覆盖真实场景。用生产环境的真实 prompt，别自己编
2. 至少 20 个用例。5 个太少，统计意义不够。建议分类抽样 50-100 个
3. 评分模型用最强的。用 GPT-4o 或 Claude 做裁判，别用被评测的模型评自己
4. 反复跑 3 次取平均。大模型有随机性，单次结果可能有偏差
5. 关注成本效益比。不只看分数，要看”每分钱的分数”——性价比才是王道

---

四、进阶：持续评测

生产环境建议配合定时任务，每周自动跑一轮评测。模型会更新，效果会波动，持续评测才能保证你始终在用最优模型。

# 定时评测（配合 cron 或 GitHub Actions）
import schedule
import time

def weekly_eval():
    """每周一早上自动评测"""
    run_evaluation()
    # 自动发送报告到飞书/钉钉

schedule.every().monday.at("09:00").do(weekly_eval)

while True:
    schedule.run_pending()
    time.sleep(60)

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/火山云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「公有云cloud」，围观AI前沿动态~

博客：y1cloud.cn