把 LLM 评测从数周压缩到一天:一条可持续迭代的工程路径

2026-07-15 30 预计阅读时间: 1 分钟
来源: medium.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

LLM 应用真正拖慢迭代的,往往不是模型调用本身,而是准备数据、批量执行、人工复核和汇总结果之间的等待。来源标题给出了一个鲜明结果:评测周期从数周缩短到一天。由于摘要没有披露具体实现,下面不推测原团队的技术栈,而是给出一套可以这样实践的工程方案:让评测集可版本化,让任务并发执行,让失败能够恢复,并把人工判断集中在最有价值的样本上。

先找到“数周”花在哪里

一次完整评测通常由多段串行工作组成:整理案例、运行模型、等待限流、检查输出、邀请领域专家复核,再生成报告。任何一段依赖手工传递,都会把几个小时的计算变成几天的日历时间。

可以先记录每个阶段的两个指标:实际处理时间和等待时间。常见瓶颈包括:

  • 每轮临时收集测试问题,评测集没有固定版本。
  • 模型请求串行运行,或者一次失败导致整批重跑。
  • 所有结果都交给人工检查,没有自动化的第一层筛选。
  • 报告由表格手工拼接,实验参数无法追溯。
  • 平均分掩盖了特定场景的退化,例如长上下文、拒答或结构化输出失败。

优化目标不应只是“跑得更快”。更重要的是缩短从代码变更到可信结论的反馈时间,同时保留输入、模型版本、提示词、评分规则和原始输出。

把评测当成可恢复的数据流水线

一条适合频繁迭代的评测流水线,可以拆成四层:

  1. 固定数据集:每条样本有稳定 ID、输入、预期行为和场景标签,并随代码一起版本管理。
  2. 并发执行器:在服务限流范围内并行请求,对超时和临时错误进行重试。
  3. 自动评分器:优先使用精确匹配、JSON Schema、正则表达式或业务规则;只有开放式质量问题才交给 LLM 裁判。
  4. 差异报告:比较候选版本与基线版本,而不是孤立地看一个总分。

每条样本都应独立落盘。这样任务中断后只需补跑缺失项,也能单独重新评判评分规则发生变化的案例。缓存键至少应包含数据集版本、提示词版本、模型标识和推理参数,避免误用旧结果。

总分也不够。建议按 scenario、语言、输入长度、风险等级等维度分层统计,同时报告通过率、失败数、延迟分位数和单样本成本。一个版本即使平均质量提高,也可能在关键场景发生不可接受的回归。

可以这样实践:一个并发、可续跑的最小评测器

下面的示例只依赖 Python 标准库。它从 JSONL 文件读取样本,并发调用一个兼容常见 Chat Completions 请求格式的 HTTP 接口,根据必需关键词执行确定性评分,并把每条结果立即写入磁盘。

运行前需要修改 LLM_BASE_URLLLM_API_KEYLLM_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 裁判,但不能把裁判分数直接当作真相。

更稳妥的分层方式是:

  • 所有样本先运行低成本规则,快速暴露格式和协议错误。
  • 对开放式回答运行固定版本的裁判提示词,并记录裁判模型版本。
  • 自动比较候选版本和基线版本,把“结果翻转”或低置信度案例送给人工。
  • 定期抽样人工复核,用来测量自动裁判与专家意见的一致性。

人工时间应该投入边界案例和高风险场景,而不是重复确认机器已经能够稳定判断的格式要求。涉及医疗、金融、安全或合规决策时,自动评分只能辅助筛选,不能替代领域专家和正式验证。

一天内完成评测,需要明确取舍

把周期压到一天,不等于每次提交都运行全部案例。更实际的做法是设置分层门禁:拉取请求运行几十到几百条高信号冒烟案例;合并后运行覆盖主要场景的回归集;每天或发布前运行完整评测,并安排人工处理自动系统标出的差异。

落地时可以按这份清单推进:

  • 为样本、提示词、模型配置和评分器建立版本号。
  • 先自动化一个最影响发布速度的评测场景。
  • 保存原始输入输出,保证评分规则可以离线重放。
  • 设置并发上限、超时、指数退避和预算上限。
  • 同时比较质量、延迟、成本和关键场景退化。
  • 维护一小组经过专家确认的黄金样本,用于校准自动评分。
  • 把评测结果绑定到提交或构建产物,确保结论可追溯。

真正有价值的变化不是把一次大型评测“加速”完成,而是建立一条开发者每天都愿意运行、失败后能够定位、结果足以支持发布决策的反馈回路。


相关推荐