一场持续一周的模型选型之争
今年3月，我们产品要上线一个智能客服功能，技术选型卡在了模型选择上。前端同学觉得GPT-4o响应快、生态好；后端同学觉得Claude Sonnet在长文本理解上更强；老板听说DeepSeek V3便宜，想省钱。
三个人各执一词，开会开了三次都没结论。最后CTO拍板："别吵了，跑数据。"
于是我开始设计A/B测试方案。一开始以为很简单，实际操作下来才发现，给AI模型做A/B测试和给普通功能做A/B测试完全是两回事。
AI模型A/B测试的核心难点
传统A/B测试假设系统输出是确定性的——按钮A和按钮B，点击率就是点击率。但AI模型不一样，你同一个问题问三遍，可能得到三个不同的回答。这带来了三个核心挑战：
非确定性输出
LLM的temperature参数决定了输出的随机性。即使temperature设为0，不同模型在处理复杂问题时的输出仍然有差异。所以你不能只测一次就下结论，每个问题至少要跑3-5次取平均值。
评估标准怎么定
传统A/B测试看转化率、点击率这些硬指标。AI模型的输出质量怎么评估？你不能让每个回答都找人工打分，成本太高。需要一套自动化的评估方案。
成本差异巨大
GPT-4o每百万token输入$2.5、输出$10，Claude Sonnet输入$3、输出$15，DeepSeek V3输入$0.27、输出$1.1。同样一批测试请求，不同模型的成本差了将近10倍。测试本身也是一笔开销，得精打细算。
三种测试方法对比
我调研了业内常见的做法，总结出三种测试方法，各有适用场景：
| 方法 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| 并行对比测试 | 同一请求同时发给多个模型 | 结果最公平，可精确对比 | 成本翻倍，只适合小批量 |
| 流量灰度切换 | 按比例分配真实流量到不同模型 | 最接近真实场景 | 需要灰度发布基础设施 |
| 离线评测 | 用标注好的数据集批量跑分 | 成本低，可重复执行 | 和真实场景可能有偏差 |
我的做法是三步走：先用离线评测缩小范围（4个模型缩小到2个），再用并行对比测试精确对比，最后用流量灰度在线验证。
Python A/B测试框架
下面是我写的测试框架代码，支持多模型并行调用、自动评分和结果统计。核心思路是用 concurrent.futures 做并行请求，用LLM-as-a-Judge的方式做自动评分。
import asyncio
import time
import json
import statistics
from dataclasses import dataclass, field
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor
@dataclass
class ModelConfig:
name: str
provider: str # openai / anthropic / deepseek / google
model_id: str
api_key: str
base_url: str = None
input_price_per_1m: float = 0.0 # 每百万token输入价格
output_price_per_1m: float = 0.0 # 每百万token输出价格
@dataclass
class TestResult:
question: str
model_name: str
answer: str
latency_ms: float
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float
quality_score: float = 0.0 # 0-10分
relevance_score: float = 0.0 # 0-10分
class ABTestFramework:
"""AI模型A/B测试框架"""
def __init__(self, models: List[ModelConfig], judge_model: ModelConfig):
self.models = models
self.judge_model = judge_model
def call_model(self, config: ModelConfig, prompt: str) -> dict:
"""调用指定模型并返回结果"""
start = time.time()
if config.provider == "openai":
from openai import OpenAI
client = OpenAI(api_key=config.api_key)
resp = client.chat.completions.create(
model=config.model_id,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1024
)
usage = resp.usage
return {
"answer": resp.choices[0].message.content,
"latency_ms": (time.time() - start) * 1000,
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
}
elif config.provider == "anthropic":
import anthropic
client = anthropic.Anthropic(api_key=config.api_key)
resp = client.messages.create(
model=config.model_id,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
usage = resp.usage
return {
"answer": resp.content[0].text,
"latency_ms": (time.time() - start) * 1000,
"prompt_tokens": usage.input_tokens,
"completion_tokens": usage.output_tokens,
"total_tokens": usage.input_tokens + usage.output_tokens,
}
elif config.provider == "deepseek":
from openai import OpenAI
client = OpenAI(
api_key=config.api_key,
base_url=config.base_url or "https://api.deepseek.com"
)
resp = client.chat.completions.create(
model=config.model_id,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1024
)
usage = resp.usage
return {
"answer": resp.choices[0].message.content,
"latency_ms": (time.time() - start) * 1000,
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
}
def auto_evaluate(self, question: str, answer: str) -> tuple:
"""用Judge模型自动评分（LLM-as-a-Judge）"""
judge_prompt = f"""你是一个专业的AI回答质量评审。
请对以下AI回答进行评分，严格按JSON格式返回。
问题：{question}
回答：{answer}
请从两个维度打分（0-10分）：
1. quality：回答的准确性、完整性和专业性
2. relevance：回答与问题的相关程度
只返回JSON，格式：{{"quality": 8, "relevance": 9}}"""
result = self.call_model(self.judge_model, judge_prompt)
try:
scores = json.loads(result["answer"])
return scores.get("quality", 5), scores.get("relevance", 5)
except json.JSONDecodeError:
return 5, 5
def run_test(self, questions: List[str], runs: int = 3) -> Dict:
"""对每个问题在所有模型上运行多次测试"""
all_results = {model.name: [] for model in self.models}
for q in questions:
for model in self.models:
for run in range(runs):
try:
raw = self.call_model(model, q)
quality, relevance = self.auto_evaluate(q, raw["answer"])
cost = (
raw["prompt_tokens"] / 1_000_000 * model.input_price_per_1m
+ raw["completion_tokens"] / 1_000_000 * model.output_price_per_1m
)
result = TestResult(
question=q,
model_name=model.name,
answer=raw["answer"],
latency_ms=raw["latency_ms"],
prompt_tokens=raw["prompt_tokens"],
completion_tokens=raw["completion_tokens"],
total_tokens=raw["total_tokens"],
cost_usd=cost,
quality_score=quality,
relevance_score=relevance,
)
all_results[model.name].append(result)
except Exception as e:
print(f"[ERROR] {model.name} run {run}: {e}")
return self._summarize(all_results)
def _summarize(self, all_results: Dict) -> Dict:
"""汇总统计结果"""
summary = {}
for model_name, results in all_results.items():
if not results:
continue
summary[model_name] = {
"样本数": len(results),
"平均延迟(ms)": round(statistics.mean(
r.latency_ms for r in results), 1),
"P95延迟(ms)": round(sorted(
r.latency_ms for r in results
)[int(len(results) * 0.95)], 1),
"平均质量分": round(statistics.mean(
r.quality_score for r in results), 2),
"平均相关性分": round(statistics.mean(
r.relevance_score for r in results), 2),
"平均Token数": round(statistics.mean(
r.total_tokens for r in results), 0),
"平均单次成本($)": round(statistics.mean(
r.cost_usd for r in results), 4),
"总成本($)": round(sum(
r.cost_usd for r in results), 4),
}
return summary
# ===== 使用示例 =====
if __name__ == "__main__":
models = [
ModelConfig("GPT-4o", "openai", "gpt-4o",
"sk-xxx", input_price_per_1m=2.5, output_price_per_1m=10),
ModelConfig("Claude Sonnet", "anthropic", "claude-sonnet-4-20250514",
"sk-ant-xxx", input_price_per_1m=3, output_price_per_1m=15),
ModelConfig("DeepSeek V3", "deepseek", "deepseek-chat",
"sk-xxx", base_url="https://api.deepseek.com",
input_price_per_1m=0.27, output_price_per_1m=1.1),
ModelConfig("Gemini Pro", "google", "gemini-1.5-pro",
"xxx", input_price_per_1m=1.25, output_price_per_1m=5),
]
judge = ModelConfig("Judge", "openai", "gpt-4o-mini", "sk-xxx")
framework = ABTestFramework(models, judge)
# 50个测试问题
test_questions = [
"如何优化Python代码的性能？",
"解释Transformer架构的工作原理",
# ... 更多问题
]
results = framework.run_test(test_questions, runs=3)
for model, stats in results.items():
print(f"\n{'='*50}")
print(f"模型: {model}")
for k, v in stats.items():
print(f" {k}: {v}")
这个框架的核心设计有几个要点：
- 多次运行取平均：每个问题在每个模型上跑3次，消除随机性影响
- LLM-as-a-Judge：用GPT-4o-mini做评分裁判，成本低且一致性不错
- 自动计算成本：根据每个模型的定价自动算出单次请求成本
- 多维度统计：延迟、质量、相关性、Token效率、成本全部量化
评估维度详解
光有代码不够，你得知道看哪些指标。我把评估维度分成五类：
| 维度 | 衡量方法 | 权重建议 |
|---|---|---|
| 准确率 | LLM-as-Judge打分 + 人工抽检 | 30% |
| 相关性 | 回答是否切题、是否包含无关内容 | 20% |
| 延迟 | P50/P95延迟，直接影响用户体验 | 20% |
| 成本 | 单次请求平均花费 | 20% |
| Token效率 | 单位Token的信息密度（质量分/Token数） | 10% |
权重的分配取决于你的业务场景。如果是客服场景，延迟权重可以高一些；如果是内容生成，准确率权重应该更高。没有放之四海而皆准的标准，得根据自己的业务来调。
真实测试数据：50个问题的四模型对比
我用上面的框架，选了50个覆盖编程、写作、推理、知识问答四个领域的问题，在四个模型上各跑了3次（共600次API调用），汇总数据如下：
| 指标 | GPT-4o | Claude Sonnet | DeepSeek V3 | Gemini Pro |
|---|---|---|---|---|
| 平均质量分 | 8.4 | 8.2 | 7.8 | 7.5 |
| 平均相关性分 | 8.6 | 8.8 | 8.1 | 7.9 |
| P50延迟(ms) | 820 | 950 | 650 | 1100 |
| P95延迟(ms) | 2100 | 2400 | 1500 | 3200 |
| 平均Token数 | 385 | 420 | 510 | 460 |
| 单次成本($) | $0.028 | $0.039 | $0.003 | $0.018 |
| Token效率(分/千Token) | 21.8 | 19.5 | 15.3 | 16.3 |
几个有趣的发现：
- GPT-4o综合最强，质量分最高且Token效率最好（同样质量下用的Token更少）
- Claude Sonnet相关性最好，回答最切题，几乎不会跑偏，但延迟稍高
- DeepSeek V3性价比碾压，成本只有GPT-4o的1/9，质量差距不到7%
- Gemini Pro在编程任务上表现不错，但综合评分偏低，且延迟波动大
注意：以上数据基于我们的具体业务场景（智能客服），你的场景可能完全不同。编程类问题GPT-4o明显更强，但中文写作类任务Claude Sonnet可能更合适。一定要跑自己的测试数据。
智能路由策略：不同问题用不同模型
测试做完后我们发现一个规律：不同类型的问题，最优模型不一样。与其一刀切选一个模型，不如做智能路由。
我们的路由策略是这样的：
class ModelRouter:
"""基于问题类型的智能模型路由"""
def __init__(self):
self.rules = {
"coding": {
"model": "gpt-4o",
"reason": "编程任务准确率最高，Token效率好",
},
"creative_writing": {
"model": "claude-sonnet",
"reason": "创意写作质量高，中文表达更自然",
},
"simple_qa": {
"model": "deepseek-v3",
"reason": "简单问答性价比最高，成本仅GPT-4o的1/9",
},
"complex_reasoning": {
"model": "gpt-4o",
"reason": "复杂推理任务质量分领先15%+",
},
"default": {
"model": "deepseek-v3",
"reason": "默认走低成本模型，兜底方案",
}
}
def classify(self, question: str) -> str:
"""简单的问题分类器（生产环境建议用嵌入模型）"""
coding_keywords = ["代码", "函数", "bug", "Python", "API",
"SQL", "编程", "开发"]
writing_keywords = ["写一篇", "文案", "文章", "故事",
"创意", "营销"]
reasoning_keywords = ["分析", "推理", "比较", "为什么",
"如何解决"]
q = question.lower()
if any(k in q for k in coding_keywords):
return "coding"
elif any(k in q for k in writing_keywords):
return "creative_writing"
elif any(k in q for k in reasoning_keywords):
return "complex_reasoning"
else:
return "simple_qa"
def route(self, question: str) -> dict:
category = self.classify(question)
rule = self.rules.get(category, self.rules["default"])
return {
"category": category,
"model": rule["model"],
"reason": rule["reason"],
}
# 使用示例
router = ModelRouter()
print(router.route("帮我写一个Python快速排序函数"))
# -> coding, gpt-4o
print(router.route("写一篇关于AI的营销文案"))
# -> creative_writing, claude-sonnet
print(router.route("今天天气怎么样"))
# -> simple_qa, deepseek-v3
这个路由策略上线后效果非常明显：
- 整体成本降了62%（大部分简单问题走DeepSeek V3）
- 用户满意度从82%提升到91%（复杂问题走GPT-4o，质量有保障）
- 平均响应延迟降了28%（DeepSeek V3比GPT-4o快20%+）
生产环境中的分类器建议用嵌入模型（Embedding）做语义匹配，而不是简单的关键词匹配。我们用的是 text-embedding-3-small，每个分类维护5-10个示例问题，计算余弦相似度来做分类，准确率能到95%以上。
从测试到持续优化
A/B测试不是一次性的工作。模型在持续更新——GPT-4o从发布到现在已经迭代了好几个版本，Claude Sonnet也在不断进步。你需要建立一套持续评测的机制：
- 每周跑一次离线评测：用固定的50个问题测试集，跟踪各模型的质量变化趋势
- 每月做一次并行对比：新模型发布时，和现有模型做精确对比
- 维护一个评测数据集：覆盖你业务的核心场景，不断补充新问题
- 关注模型更新公告：OpenAI、Anthropic、DeepSeek每次更新都可能改变排名
模型选型没有"一劳永逸"的答案，但有了数据驱动的评测体系，你至少能在每次模型更新时快速做出最优决策，而不是继续靠直觉吵架。
如果你想对比更多AI API平台的定价、功能和性能数据，可以到 TokenNexus 上看看。我们收录了330+国内外AI API平台，支持按价格、模型类型、地区等多维度筛选对比，帮你快速锁定最优方案。