lm-eval-harness 评测流程 源码精读

01 全链路概览

lm-evaluation-harness 的评测流程是一条完整的流水线，从 Shell 脚本启动开始，经过模型推理、一阶段评分、二阶段后处理，最终汇总为统一的评测报告。整个流程可以分为五个核心阶段：

以下是每个阶段的职责简述：

02 Head+Worker 架构

在 Canoe 平台上，评测任务采用 Head + Worker 双任务架构：Head 节点负责运行评测脚本（CPU 密集），Worker 节点负责运行 vLLM 推理服务（GPU 密集）。两者通过 HTTP API 进行通信。

wait_server.py 与服务发现

当设置了 SERVER_JOB_ID 环境变量时，Head 节点需要等待 Worker 节点的推理服务启动完成。run_evals_ft.sh 中对应的逻辑如下：

# 如果存在SERVER_JOB_ID且NEED_INFERENCE=1，则等待SERVER_JOB_ID的服务启动完成
if [ -n "$SERVER_JOB_ID" ] && [ "$NEED_INFERENCE" -eq 1 ]; then
    eval $(python3 /output/lm-evaluation-harness/scripts/wait_server.py)
fi
echo $SERVER_IP_DIR

wait_server.py 执行以下操作：

• 根据 SERVER_JOB_ID 查询 Canoe 平台 API，等待 Worker 任务状态变为 Running
• 从 SERVER_IP_DIR 目录中读取 *.txt 文件获取 Worker 节点的 IP 地址
• 输出 shell export 语句，设置模型服务的连接地址

NEED_INFERENCE / NEED_SCORE 控制流程

两个环境变量控制是否执行推理阶段和评分阶段：

NEED_INFERENCE=${NEED_INFERENCE:-1}
NEED_SCORE=${NEED_SCORE:-1}

# if NEED_SCORE和NEED_INFERENCE中存在0，说明是两阶段任务，
# 则使用SERVER_JOB_ID作为后缀，否则使用JOBID作为后缀
if [[ "$NEED_SCORE" -eq 0 || "$NEED_INFERENCE" -eq 0 ]]; then
    export OUTPUT_PATH=${OUTPUT_PATH}/${SERVER_JOB_ID}
else
    export OUTPUT_PATH=${OUTPUT_PATH}/${JOBID}
fi

模型连接方式

在 run_evals_ft.sh 中，模型参数被配置为通过 HTTP API 连接 vLLM 服务：

model_args="url=http://127.0.0.1:5002/playground"

当存在 Worker 节点时，127.0.0.1 会被替换为 Worker 的实际 IP 地址。SERVING_PORT 默认为 5002。lm_eval 框架通过 registry.get_model(model) 获取对应的模型类（如 vllm_clients），该类负责将评测请求封装为 HTTP 调用发送给 vLLM 服务。

03 一阶段：lm_eval 推理

run_evals_ft.sh 中的关键环境变量

入口脚本 run_evals_ft.sh 定义了一组关键环境变量，控制评测行为：

MODEL_PATH=${MODEL_PATH:-/data/minimax-dialogue/donghuang/open_llama_7b}
OUTPUT_PATH=${OUTPUT_PATH:-/data/minimax-dialogue/donghuang/dense-7b-eval}
TASKS=${TASKS:-gsm8k_5shot_generation}
MODEL=${MODEL:-ft}
BATCH_SIZE=${BATCH_SIZE:-0}
if [ $BATCH_SIZE -eq 0 ]; then
    BATCH_SIZE="auto:4.0"
fi
LIMIT=${LIMIT:-1000}
INCLUDE_PATH=${INCLUDE_PATH:-/data/minimax-dialogue/built_data/evaluation_data/pretrain_valid/evaluation_harness_tasks}
DTYPE=${DTYPE:-bfloat16}

调用 lm_eval 的实际命令如下：

if [[ "$NEED_INFERENCE" -eq 1 ]]; then
  python3 -u lm-evaluation-harness/lm_eval/__main__.py --model $MODEL \
      --model_args $model_args \
      --tasks $TASKS \
      --output_path $OUTPUT_JSON \
      --device 'cuda' \
      --log_samples \
      --batch_size $BATCH_SIZE \
      --include_path $INCLUDE_PATH \
      --limit $LIMIT \
      --verbosity DEBUG
fi

__main__.py 的 cli_evaluate() 流程

cli_evaluate() 是整个评测的 Python 入口函数，它完成从参数解析到结果写入的全过程。

1. parse_eval_args() 解析命令行参数

parse_eval_args() 使用 argparse 定义了完整的命令行接口：

def parse_eval_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
    parser.add_argument("--model", "-m", default="hf", help="Name of model e.g. `hf`")
    parser.add_argument("--tasks", "-t", default=None, metavar="task1,task2")
    parser.add_argument("--model_args", "-a", default="")
    parser.add_argument("--batch_size", "-b", type=str, default=1,
                        metavar="auto|auto:N|N")
    parser.add_argument("--output_path", "-o", default=None, type=str,
                        metavar="DIR|DIR/file.json")
    parser.add_argument("--limit", "-L", type=float, default=None)
    parser.add_argument("--include_path", type=str, default=None)
    parser.add_argument("--log_samples", "-s", action="store_true", default=False)
    parser.add_argument("--gen_kwargs", default=None)
    parser.add_argument("--verbosity", "-v", type=str.upper, default="INFO")
    return parser.parse_args()

2. include_path() 注册外部任务

当指定 --include_path 时，cli_evaluate() 调用 include_path() 将外部目录中的 YAML 任务配置注册到框架的全局任务注册表 ALL_TASKS 中：

if args.include_path is not None:
    eval_logger.info(f"Including path: {args.include_path}")
    include_path(args.include_path)

3. 任务名解析

任务名支持多种指定方式：逗号分隔字符串、glob 通配符（如 mmlu_*）、目录路径（直接扫描 YAML 文件）。核心解析逻辑：

if os.path.isdir(args.tasks):
    import glob
    task_names = []
    yaml_path = os.path.join(args.tasks, "*.yaml")
    for yaml_file in glob.glob(yaml_path):
        config = utils.load_yaml_config(yaml_file)
        task_names.append(config)
else:
    tasks_list = [task.strip() for task in args.tasks.split(",")]
    task_names = utils.pattern_match(tasks_list, ALL_TASKS)

utils.pattern_match() 使用 fnmatch 将通配符模式扩展为具体的任务名列表。例如 mmlu_* 会匹配 mmlu_abstract_algebra、mmlu_anatomy 等所有以 mmlu_ 开头的注册任务。

# 命令行参数
--tasks mmlu_*,gsm8k_5shot_generation

# tasks_list 解析为
tasks_list = ["mmlu_*", "gsm8k_5shot_generation"]

# ALL_TASKS 中注册的任务（部分）：
# {"mmlu_abstract_algebra", "mmlu_anatomy", "mmlu_astronomy", ...,
#  "gsm8k_5shot_generation", "hellaswag_local", "arc_challenge_local", ...}

# pattern_match() 使用 fnmatch 展开通配符：
# "mmlu_*" 匹配 → ["mmlu_abstract_algebra", "mmlu_anatomy", "mmlu_astronomy", ...]
# "gsm8k_5shot_generation" 精确匹配 → ["gsm8k_5shot_generation"]

# 最终 task_names：
task_names = [
    "mmlu_abstract_algebra",
    "mmlu_anatomy",
    "mmlu_astronomy",
    ...,                        # 共 57 个 mmlu_* 子任务
    "gsm8k_5shot_generation",   # 精确匹配
]

simple_evaluate() 流程

simple_evaluate() 是连接 CLI 和评测引擎的桥梁。它完成三件核心工作：

1. 创建模型实例

if isinstance(model, str):
    if model_args is None:
        model_args = ""
    lm = lm_eval.api.registry.get_model(model).create_from_arg_string(
        model_args,
        {
            "batch_size": batch_size,
            "max_batch_size": max_batch_size,
            "device": device,
        },
    )

registry.get_model(model) 从全局注册表中查找模型类。例如 model="ft" 对应 ft_lm.ft_LM 类（通过内部协议连接推理服务），model="vllm_norm" 对应 vllm_norm.VLLM 类（通过 OpenAI 兼容 API 连接 vLLM 服务）。create_from_arg_string() 将 url=http://127.0.0.1:5002/playground 这样的字符串解析为构造参数。

2. 解析任务字典并配置

task_dict = lm_eval.tasks.get_task_dict(tasks)
for task_name in task_dict.keys():
    task_obj = task_dict[task_name]
    config = task_obj._config
    if config["output_type"] == "generate_until" and gen_kwargs is not None:
        config["generation_kwargs"].update(gen_kwargs)

get_task_dict() 将任务名列表转换为 {task_name: Task} 字典。对于 generate_until 类型的任务，CLI 传入的 gen_kwargs（如 temperature=0）会覆盖 YAML 配置中的默认生成参数。

3. 调用 evaluate() 执行评测

results = evaluate(
    lm=lm,
    task_dict=task_dict,
    limit=limit,
    bootstrap_iters=bootstrap_iters,
    write_out=write_out,
    log_samples=log_samples,
    path=path,
    kill_server=True,
)

evaluate() 函数详解

evaluate() 是评测的核心引擎，执行以下步骤：

步骤 1：构建 Instances 和请求队列

# 为每个任务构建所有请求
task.build_all_requests(limit=limit, rank=lm.rank, world_size=lm.world_size)

# 按请求类型（reqtype）聚合所有 Instance
for instance in task.instances:
    reqtype = instance.request_type
    requests[reqtype].append(instance)

每个任务通过 build_all_requests() 生成 Instance 列表。每个 Instance 代表一个待推理的请求，包含 prompt、目标答案等信息。request_type 决定了调用模型的哪个方法（如 generate_until 或 loglikelihood）。

步骤 2：执行模型推理

for reqtype, reqs in requests.items():
    cloned_reqs = [req for req in reqs for _ in range(req.repeats)]
    # 通过模型执行推理
    resps = getattr(lm, reqtype)(cloned_reqs)

    # 将响应分配回各个 Instance
    for x, req in zip(resps, cloned_reqs):
        if isinstance(x, str):
            filtered_think_resp = x.split('</think>')[1] if '</think>' in x else x
            req.resps.append(filtered_think_resp)
            req.think_resps.append(think_resp)
            req.raw_resps.append(x)
        elif isinstance(x, dict):
            req.resps.append(x['content'])
            req.think_resps.append(x['reasoning_content'])
            req.raw_resps.append(x)

步骤 3：一阶段评分

for task_name, task in task_dict.items():
    for key in task.instances[0].filtered_resps.keys():
        for doc_id, doc in tqdm(doc_iterator, total=total):
            requests = list(filter(lambda x: x.doc_id == doc_id, task.instances))
            requests.sort(key=lambda x: x.idx)
            # 调用任务的 process_results 进行评分
            processed_results = task.process_results(
                doc, [req.filtered_resps[key] for req in requests]
            )
            for metric, value in metrics.items():
                vals[(task_name, key, metric)].append(value)

每个任务的 process_results() 方法根据 YAML 配置中的 metric_list 计算得分。例如 exact_match、acc、bleu 等指标都在这里完成计算。

SPLIT_BY_TASK 模式

当环境变量 SPLIT_BY_TASK=1 时，simple_evaluate() 不会一次性运行所有任务，而是逐个任务调用 evaluate()，每个任务有独立的输出目录：

split_by_task = int(os.environ.get("SPLIT_BY_TASK", 0))

if split_by_task:
    results_for_tasks = {}
    for i, task in enumerate(tasks):
        task_dict = lm_eval.tasks.get_task_dict(task)
        # 创建 task 目录
        task_path = path.joinpath(task_name)
        task_path.mkdir(parents=True, exist_ok=True)
        # 逐个任务执行 evaluate()
        is_last_task = (i == len(tasks) - 1)
        results = evaluate(
            lm=lm, task_dict=task_dict, limit=limit, path=str(task_path),
            kill_server=is_last_task  # 最后一个任务完成后终止服务
        )
        results_for_tasks[task] = results

结果写入

评测完成后，cli_evaluate() 将结果写入三种格式的文件：

# 1. JSON 格式的完整结果
output_path_file = task_path.joinpath("eval_results.json")
output_path_file.open("w").write(dumped)

# 2. 纯文本格式（方便人工阅读）
with open(output_path_file.with_suffix(".txt"), "w") as f:
    f.write(format_for_copy(results))

# 3. JSONL 格式的逐样本日志
if args.log_samples:
    output_name = f'{task_name}_log_samples'
    filename = task_path.joinpath(f"{output_name}.jsonl")
    filename.write_text(samples_dumped, encoding="utf-8")

// ① eval_results.json（完整结构化结果）
{
  "results": {
    "gsm8k_5shot_generation": {
      "exact_match,get-answer": 0.667,
      "exact_match_stderr,get-answer": 0.047
    }
  },
  "configs": {
    "gsm8k_5shot_generation": {
      "task": "gsm8k_5shot_generation",
      "output_type": "generate_until",
      "num_fewshot": 0
    }
  },
  "versions": {"gsm8k_5shot_generation": "1.0"},
  "n-shot": {"gsm8k_5shot_generation": 0}
}

# ② eval_results.txt（人工可读格式）
|     Tasks      |Version|Filter|n-shot| Metric    |Value |   |Stderr|
|----------------|------:|------|-----:|-----------|-----:|---|-----:|
|gsm8k_5shot_gen.|    1.0|get-answer|    0|exact_match|0.6670|±  |0.0470|

// ③ gsm8k_5shot_generation_log_samples.jsonl（逐样本日志）
{"doc_id": 0, "doc": {"question": "Janet's ducks...", "answer": "#### 18"}, "target": "18", "resps": [["#### 18"]], "filtered_resps": ["18"], "exact_match": 1.0}
{"doc_id": 1, "doc": {"question": "A robe takes...", "answer": "#### 15"}, "target": "15", "resps": [["#### 10"]], "filtered_resps": ["10"], "exact_match": 0.0}

此外，如果当前运行在 Canoe 平台上（CANOE_METRICS_DIST_TYPE == "canoe"），还会调用 report_metric(results) 将结果上报到平台。

04 二阶段：后处理评分

为什么需要二阶段？

并非所有任务都能在 evaluate() 的 process_results() 中完成评分。以下几类任务需要额外的后处理步骤：

• 代码执行类：humaneval、multipl-e、bigcodebench、livecodebench、swebench 等 -- 需要在沙箱中实际运行生成的代码并检验测试用例
• 数学推理类：aime24、olympiadbench、gaokao2024 等 -- 需要使用专门的数学答案提取和等价判定逻辑
• 外部 API 评分类：mtBench、alignBench、arena_hard 等 -- 需要调用 GPT-4 等外部模型进行评分
• 复杂后处理类：livebench、ifeval、superclue_if 等 -- 需要特定领域的评分脚本

second_stage_all.sh 的模式匹配机制

second_stage_all.sh 通过 Bash 的字符串匹配来判断当前运行的任务是否需要二阶段处理：

# 模式匹配语法：if [[ $TASKS == *"任务关键字"* ]]
if [[ $TASKS == *"multipl-e"* || $TASKS == *"humaneval_"*"_generation"* || $TASKS == *"mbpp_"*"_generation"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/multipl-e/score.py
fi

if [[ $TASKS == *"bigcodebench"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/bigcodebench/bigcodebench_pt_evaluate.py
fi

if [[ $TASKS == *"aime24"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/aime24/score.py
fi

这种设计的优点是非侵入式：新增任务只需在 second_stage_all.sh 中添加一个 if 分支，无需修改框架核心代码。

各类二阶段任务一览

second_stage_all.sh 按任务类别组织，包含以下几大板块：

代码任务

# ======================= 代码任务 开始 =======================
if [[ $TASKS == *"multipl-e"* || $TASKS == *"humaneval_"*"_generation"* || $TASKS == *"mbpp_"*"_generation"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/multipl-e/score.py
fi
if [[ $TASKS == *"bigcodebench"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/bigcodebench/bigcodebench_pt_evaluate.py
fi
if [[ $TASKS == *"livecodebench_test5"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/livecodebench/score_test5.py --task livecodebench_test5_python
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/livecodebench/score_test5.py --task livecodebench_test5_python_2shot
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/livecodebench/score_test5.py --task livecodebench_test5_cpp
fi
if [[ $TASKS == *"swebench_patch_generation"* || $TASKS == *"swebench_lite"* || $TASKS == *"swebench_verified"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/swebench/utils_tooltool.py
fi
# ======================= 代码任务 结束 =======================

数学任务

# ======================= 数学任务 开始 =======================
if [[ $TASKS == *"aime24"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/aime24/score.py
fi
if [[ $TASKS == *"olympiadbench"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/olympiadbench/score.py
fi
if [[ $TASKS == *"gaokao2024"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/gaokao2024/score.py
fi
if [[ $TASKS == *"college_math"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/college_math/score.py
fi
if [[ $TASKS == *"omnimath"* ]]; then
    python3 -u lm-evaluation-harness/lm_eval/tasks/sft/omnimath/score.py
fi
# ======================= 数学任务 结束 =======================

其他类别

除代码和数学外，second_stage_all.sh 还包含以下类别的后处理评分：

不需要二阶段的任务

以下类型的任务在 evaluate() 一阶段中通过 metric_list 配置直接完成评分，无需二阶段：

• 多选/判断类：hellaswag、arc_easy、arc_challenge、piqa、winogrande、openbookqa、boolq -- 使用 loglikelihood 方式，比较各选项的概率
• 简单生成 + 精确匹配：gsm8k、triviaqa、nq_open -- 使用 generate_until 方式，通过 exact_match 或正则提取答案进行匹配
• MMLU 系列：所有 mmlu_* 子任务 -- 使用 loglikelihood 方式，比较 A/B/C/D 的概率

05 指标收集

collect_metrics.py 的工作方式

在一阶段和二阶段评分都完成后，run_evals_ft.sh 的最后一步是调用 collect_metrics.py 汇总所有结果：

# run_evals_ft.sh 最后三行
source lm-evaluation-harness/second_stage_all.sh
python3 lm-evaluation-harness/scripts/collect_metrics.py

collect_metrics.py 的核心逻辑是：扫描 OUTPUT_PATH 目录下的各种结果文件，按固定格式提取指标，拼接为 name\tscore\n 格式的字符串。

eval_dir = os.environ.get("OUTPUT_PATH",
    "/data/minimax-dialogue/users/shennai/eval_results/07/16/cceval/j-aowp1hys2q")
res = ""

# 示例：读取 livebench 结果
try:
    with open(os.path.join(eval_dir, "eval_results.txt"), "r") as f:
        for line in f.readlines()[::-1]:
            if line.startswith("global_average:"):
                res += f"livebench_平均\t{round(float(line.split(':')[1]), 3)}\n"
            elif line.startswith("coding_average:"):
                res += f"livebench_代码\t{round(float(line.split(':')[1]), 3)}\n"
except:
    pass

结果文件类型

collect_metrics.py 会尝试读取多种格式的结果文件：

汇总与上报

所有指标被收集到 res 字符串中后，写入 eval_results_collect.txt 文件，同时构建 eval_results_dict 字典：

eval_results_dict = {}
with open(os.path.join(eval_dir, "eval_results_collect.txt"), "w") as f:
    f.write(res)
    for line in res.split('\n'):
        if line:
            if '\t' in line:
                f.write(line.split('\t')[1] + "\n")
                eval_results_dict[line.split('\t')[0]] = float(line.split('\t')[1])

# eval_results_collect.txt 内容示例
gsm8k_5shot_generation	0.667
hellaswag_local	0.582
arc_challenge_local	0.453
mmlu_平均	0.635
humaneval_python	0.402
livebench_平均	0.321
livebench_代码	0.285
livebench_数学	0.356
aime24	0.133

如果当前运行在 Canoe 平台的预训练自动评测模式下（PRETRAIN_AUTO_EVAL=true），还会通过 Canoe Client SDK 将评测结果上报给平台：

if os.environ.get("PRETRAIN_AUTO_EVAL", "False").lower() == "true":
    client = Client(
        base_url="https://canoe-apiserver.xaminim.com",
        ak="...",
        sk="..."
    )
    client.training_eval.finish_talos_task(
        success=True,
        metrics=eval_results_dict
    )

完整的数据流总结

至此，从 run_evals_ft.sh 启动到 collect_metrics.py 输出最终报告，整个评测流程形成闭环。每个环节的输入输出关系清晰，便于调试和扩展新任务。