Chatbot Arena 没有豆包？用 AI 辅助开发打造定制化对话评测系统
摘要：开发者在使用 Chatbot Arena 进行对话模型评测时，常遇到无法自定义评测标准（如“豆包”指标）的痛点。本文介绍如何利用开源框架和 AI 辅助开发技术，快速构建支持自定义指标的对话评测系统。通过本文，您将掌握评测系统核心架构设计、指标扩展接口实现，以及如何避免多模型并发评测时的常见问题。
1. 背景：Chatbot Arena 的“尺子”太短
Chatbot Arena 把两个模型丢进擂台，让人类点“谁更好”，最后算 Elo 分。这套流程简单、直观，却有三块短板：
- 指标封闭：只能打“胜率”一个分，想加“豆包度”（自定义业务指标）？改不动。
- 模型受限：官方只接入了几十款主流模型，私有权重、微调 checkpoint 都塞不进去。
- 并发黑洞：所有请求走单线程 Gradio，一压测就排队，日志里全是 502。
一句话：Arena 是“公共裁判”，而我们需要“私人裁判”，尺子想怎么刻就怎么刻。
2. 技术选型：三行框架横评
| 维度 | Flask | Django | FastAPI |
|---|---|---|---|
| 接口书写速度 | 快 | 中 | 飞快（基于类型提示） |
| 原生异步 | 无 | 3.2+ 部分支持 | 全程 async |
| ORM 生态 | 自由选择 | 绑定 Django ORM | 自由选择 |
| 学习曲线 | 低 | 高 | 中 |
| 生态插件 | 多 | 超多 | 快速增长 |
结论：
- 需要同步写法、后台管理开箱即用 → Django
- 需要高并发、低延迟、纯 API → FastAPI
下文代码以FastAPI为骨架，方便后续用 WebSocket 推流打分进度条。
3. 可扩展指标模块设计
3.1 插件化思路
把“打分”抽象成插件，三步走通：
- 定义抽象基类
Metric - 每个指标一个 py 文件，继承后实现
compute() - 启动时动态扫描
plugins/目录，注册到全局字典
3.2 UML 类图（Mermaid）
classDiagram class Metric { <<abstract>> +str name +compute(ref:str, hyp:str)~float } class DuBaoMetric { +compute(ref:str, hyp:str)~float } class BleuMetric { +compute(ref:str, hyp:str)~float } class MetricRegistry { +register(metric: Metric) +get(name: str) Metric } Metric <|-- DuBaoMetric Metric <|-- BleuMetric MetricRegistry o-- Metric4. 核心代码实现
4.1 抽象基类（PEP8 合规，行宽 79）
# metrics/base.py from abc import ABC, abstractmethod class Metric(ABC): """所有指标的元类""" name: str = "" @abstractmethod def compute(self, ref: str, hyp: str) -> float: """返回指标分数，越高越好""" raise NotImplementedError4.2 自定义“豆包”指标
# metrics/dubao.py import re from metrics.base import Metric class DuBaoMetric(Metric): name = "doubao" def compute(self, ref: str, hyp: str) -> float: """ 示例逻辑：答案里每出现一个‘豆’+‘包’关键词组合得 10 分 时间复杂度：O(n+m) n=len(ref) m=len(hyp) """ combo = re.findall(r'豆包', hyp) return min(len(combo) * 10.0, 100.0) # 封顶 1004.3 插件注册工厂
# metrics/registry.py import importlib import pkgutil from typing import Dict from metrics.base import Metric class MetricRegistry: _store: Dict[str, Metric] = {} @classmethod def register(cls, metric: Metric) -> None: cls._store[metric.name] = metric @classmethod def get(cls, name: str) -> Metric: return cls._store[name] def auto_discover(): """扫描同级目录下所有模块""" for module_info in pkgutil.iter_modules(__path__): module = importlib.import_module(f"metrics.{module_info.name}") for attr in dir(module): obj = getattr(module, attr) if isinstance(obj, type) and issubclass(obj, Metric) and obj is not Metric: MetricRegistry.register(obj())4.4 FastAPI 暴露评测端点
# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from metrics.registry import MetricRegistry, auto_discover app = FastAPI() auto_discover() # 启动即注册 class EvaluateReq(BaseModel): model_name: str prompt: str metric: str class EvaluateRsp(BaseModel): score: float @app.post("/evaluate", response_model=EvaluateRsp) async def evaluate(req: EvaluateReq): metric = MetricRegistry.get(req.metric) # 这里简化：同步调用模型生成 + 打分 hyp = await call_hf_model(req.model_name, req.prompt) score = metric.compute(ref="", hyp=hyp) # ref 留空，示例只测 hyp return EvaluateRsp(score=score)4.5 HuggingFace 模型异步调用
# model_client.py from transformers import AutoTokenizer, AutoModelForCausalLM import torch import asyncio _lock = asyncio.Lock() tokenizer = AutoTokenizer.from_pretrained("THUDM/chatglm-6b") model = AutoModelForCausalLM.from_pretrained("THUDM/chatglm-6b", torch_dtype=torch.float16, device_map="auto") async def call_hf_model(model_name: str, prompt: str) -> str: async with _lock: # 显卡资源有限，加锁防并发 loop = asyncio.get_event_loop() inputs = tokenizer(prompt, return_tensors="pt").to(model.device) outputs = await loop.run_in_executor( None, lambda: model.generate(**inputs, max_new_tokens=128) ) return tokenizer.decode(outputs[0], skip_special_tokens=True)5. 生产级加固
5.1 高并发解耦
- Celery + Redis做任务队列，
/evaluate只落任务 ID，返回 202 状态 +Location头，前端轮询或 WebSocket 推送。 - worker 容器独立扩缩，GPU 节点打标签，只跑模型推理，CPU 节点跑打分与入库。
5.2 数据库选型
| 场景 | MongoDB | PostgreSQL |
|---|---|---|
| 结果字段不固定、指标插件随意加 | ✔ 文档型，无 schema 约束 | 需 JSONB |
| 事务、复杂窗口函数 | 弱事务 | ✔ ACID、窗口函数丰富 |
| 横向读扩展 | 原生分片 | 需 Citus 等插件 |
建议：
- 评测记录写多读多 → PostgreSQL，JSONB 存动态指标列；
- 若团队已深度用 Mongo，亦可，但记得开
transaction保证“任务幂等”。
5.3 模型冷启动优化
- 预加载：容器启动即
model.half().cuda()，避免第一请求触发。 - Warm-up 探针：k8s
readinessProbe调一次假推理，确认显存稳定后再接流量。 - ZeRO-Inference：显存不足时，用 DeepSpeed 拆层到内存 + 盘，速度换空间。
5.4 幂等性保障
- 任务 ID 用
model_name+prompt_sha256生成，Redis 置 60 s 过期 NX 锁，防重复提交。 - PostgreSQL 侧建唯一索引
(model_name, prompt_sha256)，冲突抛回旧结果，实现“同一问题只测一次”。
6. 避坑小结
- 指标插件勿在
compute()里做重型 IO，会拖赛调度线程；如需调外部词典，提前预加载。 - 扫描插件目录别用
__import__字符串拼接，容易踩到命名空间重复；推荐importlib.import_module。 - GPU 容器里如果忘了限制
CUDA_VISIBLE_DEVICE，多 worker 会互相 OOM；用nvidia-docker的device字段显式绑定。 - 横向扩容时，Elo 分数表不要放在本地内存，否则不同 Pod 算出的排名分裂；统一写 Redis Hash 或 PG 行锁更新。
7. 留给读者的三个开放问题
- 当“豆包”指标与官方 Elo 分出现倒挂时，你如何向业务方解释哪项更代表真实用户体验？
- 如果允许厂商针对自定义指标刷榜，你会在系统层引入哪些“防作弊”机制？
- 在多语言、多方言场景下，指标插件应如何保证文化公平，而不让某一语料库成为“高分密码”？
把尺子握在自己手里，才能真正量出模型对业务的价值。
希望这篇笔记帮你搭好“私人裁判”的骨架，剩下的规则，由你和你的场景一起写。
想亲手把语音对话也做出可定制的评测？我在 从0打造个人豆包实时通话AI 动手实验里跑通了 ASR→LLM→TTS 全链路，代码全部开源，本地一键
docker-compose up就能跑。周末花两小时，你也能把“豆包度”测到毫秒级延迟里，顺便体验下当裁判的快乐。
