LLM 应用真正拖慢迭代的,往往不是模型调用本身,而是准备数据、批量执行、人工复核和汇总结果之间的等待。来源标题给出了一个鲜明结果:评测周期从数周缩短到一天。由于摘要没有披露具体实现,下面不推测原团队的技术栈,而是给出一套可以这样实践的工程方案:让评测集可版本化,让任务并发执行,让失败能够恢复,并把人工判断集中在最有价值的样本上。
先找到“数周”花在哪里
一次完整评测通常由多段串行工作组成:整理案例、运行模型、等待限流、检查输出、邀请领域专家复核,再生成报告。任何一段依赖手工传递,都会把几个小时的计算变成几天的日历时间。
可以先记录每个阶段的两个指标:实际处理时间和等待时间。常见瓶颈包括:
- 每轮临时收集测试问题,评测集没有固定版本。
- 模型请求串行运行,或者一次失败导致整批重跑。
- 所有结果都交给人工检查,没有自动化的第一层筛选。
- 报告由表格手工拼接,实验参数无法追溯。
- 平均分掩盖了特定场景的退化,例如长上下文、拒答或结构化输出失败。
优化目标不应只是“跑得更快”。更重要的是缩短从代码变更到可信结论的反馈时间,同时保留输入、模型版本、提示词、评分规则和原始输出。
把评测当成可恢复的数据流水线
一条适合频繁迭代的评测流水线,可以拆成四层:
- 固定数据集:每条样本有稳定 ID、输入、预期行为和场景标签,并随代码一起版本管理。
- 并发执行器:在服务限流范围内并行请求,对超时和临时错误进行重试。
- 自动评分器:优先使用精确匹配、JSON Schema、正则表达式或业务规则;只有开放式质量问题才交给 LLM 裁判。
- 差异报告:比较候选版本与基线版本,而不是孤立地看一个总分。
每条样本都应独立落盘。这样任务中断后只需补跑缺失项,也能单独重新评判评分规则发生变化的案例。缓存键至少应包含数据集版本、提示词版本、模型标识和推理参数,避免误用旧结果。
总分也不够。建议按 scenario、语言、输入长度、风险等级等维度分层统计,同时报告通过率、失败数、延迟分位数和单样本成本。一个版本即使平均质量提高,也可能在关键场景发生不可接受的回归。
可以这样实践:一个并发、可续跑的最小评测器
下面的示例只依赖 Python 标准库。它从 JSONL 文件读取样本,并发调用一个兼容常见 Chat Completions 请求格式的 HTTP 接口,根据必需关键词执行确定性评分,并把每条结果立即写入磁盘。
运行前需要修改 LLM_BASE_URL、LLM_API_KEY 和 LLM_MODEL。服务端的具体接口格式不在来源摘要中,因此这里明确采用一个可替换的假设接口。
先创建 cases.jsonl:
{"id":"refund-001","scenario":"policy","input":"用户购买 40 天后要求退款,应如何回复?","must_contain":["退款政策","人工客服"]}
{"id":"json-001","scenario":"format","input":"只返回 JSON:提取姓名张三和年龄 28。","must_contain":["张三","28"]}
再创建 evaluate.py:
import json
import os
import threading
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
BASE_URL = os.environ.get("LLM_BASE_URL", "http://localhost:8000/v1")
API_KEY = os.environ.get("LLM_API_KEY", "")
MODEL = os.environ.get("LLM_MODEL", "your-model")
WORKERS = int(os.environ.get("EVAL_WORKERS", "4"))
OUTPUT = Path("results.jsonl")
write_lock = threading.Lock()
def load_jsonl(path):
with open(path, encoding="utf-8") as f:
return [json.loads(line) for line in f if line.strip()]
def completed_ids():
if not OUTPUT.exists():
return set()
return {row["id"] for row in load_jsonl(OUTPUT)}
def call_model(prompt, attempts=3):
payload = json.dumps({
"model": MODEL,
"temperature": 0,
"messages": [{"role": "user", "content": prompt}],
}).encode("utf-8")
headers = {"Content-Type": "application/json"}
if API_KEY:
headers["Authorization"] = f"Bearer {API_KEY}"
for attempt in range(attempts):
try:
request = urllib.request.Request(
f"{BASE_URL}/chat/completions",
data=payload,
headers=headers,
method="POST",
)
started = time.perf_counter()
with urllib.request.urlopen(request, timeout=60) as response:
data = json.load(response)
return data["choices"][0]["message"]["content"], time.perf_counter() - started
except (urllib.error.URLError, TimeoutError):
if attempt == attempts - 1:
raise
time.sleep(2 ** attempt)
def evaluate(case):
output, latency = call_model(case["input"])
missing = [text for text in case.get("must_contain", []) if text not in output]
return {
"id": case["id"],
"scenario": case.get("scenario", "unknown"),
"passed": not missing,
"missing": missing,
"latency_seconds": round(latency, 3),
"output": output,
"model": MODEL,
}
def persist(row):
with write_lock:
with OUTPUT.open("a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
def main():
done = completed_ids()
cases = [case for case in load_jsonl("cases.jsonl") if case["id"] not in done]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
futures = {pool.submit(evaluate, case): case for case in cases}
for future in as_completed(futures):
case = futures[future]
try:
row = future.result()
except Exception as exc:
row = {"id": case["id"], "scenario": case.get("scenario"), "error": str(exc)}
persist(row)
print(json.dumps(row, ensure_ascii=False))
if __name__ == "__main__":
main()
执行命令:
export LLM_BASE_URL="https://your-llm-service.example/v1"
export LLM_API_KEY="replace-me"
export LLM_MODEL="your-model-name"
export EVAL_WORKERS=8
python evaluate.py
这个最小实现展示了三项关键能力:通过线程池压缩网络等待时间,通过稳定 ID 跳过已经完成的样本,通过逐条写入避免进程中断后丢失整批结果。生产环境还应加入速率限制、响应 Schema 校验、成本记录和更严格的错误分类。
自动评分不能取代人工判断
确定性规则适合检查 JSON 是否有效、字段是否齐全、引用是否存在以及工具调用参数是否合法。语气、完整性、事实一致性等开放式问题,可以增加 LLM 裁判,但不能把裁判分数直接当作真相。
更稳妥的分层方式是:
- 所有样本先运行低成本规则,快速暴露格式和协议错误。
- 对开放式回答运行固定版本的裁判提示词,并记录裁判模型版本。
- 自动比较候选版本和基线版本,把“结果翻转”或低置信度案例送给人工。
- 定期抽样人工复核,用来测量自动裁判与专家意见的一致性。
人工时间应该投入边界案例和高风险场景,而不是重复确认机器已经能够稳定判断的格式要求。涉及医疗、金融、安全或合规决策时,自动评分只能辅助筛选,不能替代领域专家和正式验证。
一天内完成评测,需要明确取舍
把周期压到一天,不等于每次提交都运行全部案例。更实际的做法是设置分层门禁:拉取请求运行几十到几百条高信号冒烟案例;合并后运行覆盖主要场景的回归集;每天或发布前运行完整评测,并安排人工处理自动系统标出的差异。
落地时可以按这份清单推进:
- 为样本、提示词、模型配置和评分器建立版本号。
- 先自动化一个最影响发布速度的评测场景。
- 保存原始输入输出,保证评分规则可以离线重放。
- 设置并发上限、超时、指数退避和预算上限。
- 同时比较质量、延迟、成本和关键场景退化。
- 维护一小组经过专家确认的黄金样本,用于校准自动评分。
- 把评测结果绑定到提交或构建产物,确保结论可追溯。
真正有价值的变化不是把一次大型评测“加速”完成,而是建立一条开发者每天都愿意运行、失败后能够定位、结果足以支持发布决策的反馈回路。