任务注册机制
lm-eval-harness 的任务系统基于自动发现 + 注册表模式:框架在启动时递归扫描任务目录下的所有 YAML 文件,为每个文件动态创建一个 ConfigurableTask 子类,并注册到全局注册表中。
核心注册表
三个全局字典构成任务系统的骨架(定义在 lm_eval/api/registry.py):
| 注册表 | 类型 | 说明 |
|---|---|---|
TASK_REGISTRY | dict[str, type] | 任务名 → ConfigurableTask 子类的映射 |
GROUP_REGISTRY | dict[str, list[str]] | 分组名 → 包含的任务名列表 |
ALL_TASKS | set[str] | 所有已注册的任务名和分组名的集合 |
注册流程
从 --include_path 到注册表的完整链路:
第一轮:注册单个任务"] B --> D["include_task_folder(dir, register_task=False)
第二轮:注册分组"] C --> E["遍历目录下所有 .yaml 文件"] E --> F["utils.load_yaml_config(yaml_path)"] F --> G["register_configurable_task(config)"] G --> H["type() 动态创建子类"] H --> I["register_task(task_name)(SubClass)"] I --> J["TASK_REGISTRY + ALL_TASKS"] G --> K["register_group(group_name)(SubClass)"] K --> L["GROUP_REGISTRY + ALL_TASKS"]
register_configurable_task 源码
这个函数是注册的核心 —— 它使用 Python 的 type() 元编程动态创建类:
def register_configurable_task(config: Dict[str, str]) -> int:
# 动态创建 ConfigurableTask 的子类,CONFIG 属性设置为解析后的 TaskConfig
SubClass = type(
config["task"] + "ConfigurableTask",
(ConfigurableTask,),
{"CONFIG": TaskConfig(**config)},
)
if "task" in config:
task_name = "{}".format(config["task"])
register_task(task_name)(SubClass) # 注册到 TASK_REGISTRY
if "group" in config:
# group 可以是字符串或列表
group_name = [config["group"]] if type(config["group"]) == str else config["group"]
for group in group_name:
register_group(group)(SubClass) # 注册到 GROUP_REGISTRY
return 0
每个 YAML 文件最终变成一个独立的 Python 类。例如 gsm8k_5shot_generation.yaml 会生成一个名为 gsm8k_5shot_generationConfigurableTask 的类,继承自 ConfigurableTask,其 CONFIG 属性包含了 YAML 中的所有配置。
include_task_folder 的两轮扫描
include_path() 对同一个目录调用两次 include_task_folder():
def include_path(task_dir):
include_task_folder(task_dir) # 第一轮:注册单个任务
include_task_folder(task_dir, register_task=False) # 第二轮:注册分组(Benchmarks)
return 0
第一轮处理 task 字段为字符串的配置(单个任务),第二轮处理 task 字段为列表的配置(分组/Benchmark 定义)。这保证了分组引用的任务在注册分组时已经存在。
YAML 字段全解
每个任务的 YAML 文件对应 TaskConfig dataclass 中的字段。以下是所有字段的完整参考:
任务标识
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
task | str | 是 | 任务名,必须全局唯一。用于 --tasks 参数指定 |
group | str | list | 否 | 所属分组。可以是单个字符串或列表,用于将多个任务归类为一个 Benchmark |
task_alias | str | 否 | 任务显示别名 |
数据集配置
| 字段 | 类型 | 说明 |
|---|---|---|
dataset_path | str | 数据集路径。本地 JSON 数据集用 "json" |
dataset_kwargs | dict | 传给 datasets.load_dataset() 的额外参数。通常为 {data_dir: "/path/to/data/"} |
dataset_name | str | 数据集子集名。本地数据通常为 null |
test_split | str | 测试集 split 名(如 "test") |
training_split | str | 训练集 split 名(用于 few-shot 示例来源) |
validation_split | str | 验证集 split 名 |
输入输出配置
| 字段 | 类型 | 说明 |
|---|---|---|
output_type | str | "generate_until" | "multiple_choice" | "loglikelihood" | "loglikelihood_rolling" |
doc_to_text | str | Callable | 构建输入 prompt。支持字段名、Jinja2 模板、!function |
doc_to_target | str | Callable | 构建目标答案。格式同上 |
doc_to_choice | str | list | 仅 multiple_choice:定义选项列表。支持字段名、Jinja2 |
num_fewshot | int | few-shot 示例数量。0 表示 zero-shot |
description | str | 任务描述,会作为 prompt 前缀 |
生成与评分配置
| 字段 | 类型 | 说明 |
|---|---|---|
generation_kwargs | dict | 生成参数:until(停止词列表)、temperature、max_gen_toks、do_sample、top_p |
metric_list | list | 指标列表。每个元素含 metric、aggregation、higher_is_better |
filter_list | list | 输出过滤器列表。对生成文本做后处理(如 regex 提取答案) |
process_docs | Callable | 数据预处理函数,用 !function utils.func_name 引用 |
process_results | Callable | 自定义评分函数,用 !function utils.func_name 引用 |
repeats | int | 每个样本推理次数,默认 1 |
其他配置
| 字段 | 类型 | 说明 |
|---|---|---|
metadata | dict | 元数据。常用 {version: 1.0} |
target_delimiter | str | prompt 与 target 之间的分隔符,默认 " " |
fewshot_delimiter | str | few-shot 示例之间的分隔符,默认 "\n\n" |
should_decontaminate | bool | 是否进行数据去污染 |
include | str | 继承的模板文件名(详见第 07 节) |
如果 output_type 为 generate_until 但未指定 generation_kwargs,框架会自动设置 {until: [fewshot_delimiter], do_sample: false}。此外,环境变量 TEMPERATURE、TOP_P、MAX_GEN_TOKS 可以全局覆盖这些参数。
YAML 字段在管线中的位置
YAML 里的每个字段最终都会在管线的某个环节被框架调用。下图标注了每个字段的精确调用位置,紫色标签 = 对应的 YAML 字段名。
YAML 字段调用位置一览
下表汇总每个关键字段在管线的哪一步被谁调用、输入输出各是什么:
| YAML 字段 | 调用阶段 | 调用者 | 输入 | 输出 |
|---|---|---|---|---|
dataset_pathdataset_kwargs |
数据加载 | datasets.load_dataset() |
路径 + 参数 | HF Dataset 对象 |
process_docs |
文档预处理 | task._process_doc(),数据集加载后立即调用 |
Dataset 对象(整个 split) |
清洗后的 Dataset |
doc_to_text |
Prompt 构建 | fewshot_context() 内部调用 |
单条 doc 字典 |
prompt 字符串(如 "Question: ...?\nAnswer:") |
doc_to_target |
Prompt 构建 + 评分 | ① ContextSampler(拼 fewshot 答案)② construct_requests()(loglikelihood/mc 的 continuation)③ process_results()(对比正确答案) |
单条 doc 字典 |
目标字符串 或 选项索引 int |
doc_to_choice |
请求构建 | construct_requests()(仅 multiple_choice) |
单条 doc 字典 |
选项列表 List[str] |
num_fewshot |
Prompt 构建 | fewshot_context(doc, num_fewshot) |
整数 | 控制从 train split 采样的示例数量 |
generation_kwargs |
请求构建 | construct_requests() → 写入 Instance.arguments |
— | {until, temperature, max_gen_toks, ...} |
filter_list |
后处理 | task.apply_filters(),模型推理完成后 |
instance.resps(模型原始输出列表) |
instance.filtered_resps(提取后的答案) |
process_results |
评分 | evaluate() 循环逐条调用 |
(doc, results),results = filtered_resps 值 |
{"metric_name": score} 字典 |
metric_list |
评分 + 聚合 | ① 未配置 process_results 时提供默认指标② aggregation() 汇总所有样本 |
逐条分数列表 | 最终平均分 + stderr |
阶段五详解:后处理 & 评分的三步流水线
模型推理结束后,每个 Instance 的 resps 已填充。接下来经过三步才得到最终分数:
第一步:apply_filters — 从模型原始输出中提取答案
只对 generate_until 有意义(loglikelihood/multiple_choice 的 resps 已经是数值元组,无需过滤)。
# 代码位置:evaluator.py:570
task.apply_filters()
# apply_filters 内部(filter.py:45-56):
# 遍历 filter_list 中的每个 FilterEnsemble(如 "get-answer")
# FilterEnsemble 是一个 pipeline,包含多个串行的 filter
# ═══════════════════════════════════════════════════
# 具体执行过程(以 GSM8K 为例)
# ═══════════════════════════════════════════════════
# 输入:所有 Instance 的 resps(二维列表,外层=样本数,内层=repeats)
resps = [inst.resps for inst in instances]
# resps = [
# ["She sells 16-3-4=9 eggs. 9*$2=$18. The answer is: 18."], # doc 0
# ["Let me think... 5+3=8. The answer is: 8."], # doc 1
# ["I need to calculate... The answer is: 42."], # doc 2
# ... # 所有样本
# ]
# 注意!filter 是对整个列表批量操作,不是逐条:
for f in filters: # regex → regex → take_first
resps = f.apply(resps, docs)
# 最终写回每个 Instance:
for inst, resp in zip(instances, resps):
inst.filtered_resps["get-answer"] = resp
# inst.filtered_resps = {"get-answer": "18"} ← 干净答案
如果 YAML 中没有 filter_list,框架会自动创建一个默认 pipeline,名为 "none",只包含一个 take_first。效果:instance.filtered_resps["none"] = instance.resps[0](取列表第一个元素)。
第二步:process_results — 逐条样本计算指标
evaluate() 遍历每条 doc,收集该 doc 对应的所有 Instance,取出 filtered_resps,传给 process_results。
# 代码位置:evaluator.py:688-757
for doc_id, doc in doc_iterator:
# 1. 找出属于这条 doc 的所有 Instance,按 idx 排序
requests = [x for x in task.instances if x.doc_id == doc_id]
requests.sort(key=lambda x: x.idx)
# 2. 提取 filtered_resps,拼成 results 列表
results = [req.filtered_resps[key] for req in requests]
#
# ── generate_until:只有 1 个 Instance ──
# results = ["18"] ← 一个元素(filtered 后的答案字符串)
#
# ── multiple_choice:有 N 个 Instance(每选项一个)──
# results = [(-1.23, True, "..."), (-4.56, False, "..."), ...]
# ↑ 每个选项的 (loglikelihood, is_greedy, info_dict)
#
# ── loglikelihood:只有 1 个 Instance ──
# results = [(-3.45, False, "...")]
# 3. 调用 process_results(两种情况)
metrics = task.process_results(doc, results)
# 情况 A — YAML 配了 process_results: !function utils.process_results
# → 直接调用自定义函数,返回 {"em": 0.5}
#
# 情况 B — 没配 process_results(用框架内置逻辑)
# → 根据 output_type + metric_list 自动计算:
# generate_until + exact_match: result[0] vs doc_to_target(doc)
# multiple_choice + acc: argmax(lls) == gold_index
# loglikelihood + perplexity: 直接用 ll 值
# 4. 存入 vals 列表(后续聚合用)
for metric, value in metrics.items():
vals[(task_name, key, metric)].append(value)
# vals[("gsm8k", "get-answer", "exact_match")] = [1.0, 0.0, 1.0, ...]
概念辨析:Task、doc、Instance 三者是什么关系 重要
一句话总结
# Task = 一个 YAML 文件 = 一个评测任务
# doc = 数据集中的一行 = 一道题
# Instance = 一次模型调用请求 = 一道题拆出来的一个或多个请求
# 关系:
# 1 个 Task 包含 N 条 doc(来自 test.jsonl)
# 1 条 doc 产生 1 个或多个 Instance(取决于 output_type)
用具体例子理解
假设你运行 --tasks gsm8k_5shot_generation,arc_challenge_local,数据集各有 500 / 1000 条:
# ═══════════════════════════════════════════════════
# Task 层:每个 YAML 对应一个 Task 对象
# ═══════════════════════════════════════════════════
task_dict = {
"gsm8k_5shot_generation": Task对象A, # ← gsm8k_5shot_generation.yaml
"arc_challenge_local": Task对象B, # ← arc_challenge.yaml
}
# Task 对象里有什么?
Task对象A.config # TaskConfig(YAML 解析后的所有配置)
Task对象A.dataset # HF Dataset(test.jsonl 加载后的数据集)
Task对象A.instances # List[Instance](所有请求,后面详解)
Task对象A.OUTPUT_TYPE # "generate_until"
# ═══════════════════════════════════════════════════
# doc 层:数据集中的每一行就是一个 doc
# ═══════════════════════════════════════════════════
# GSM8K 有 500 条 doc:
doc_0 = {"prompt": "Janet's ducks...", "label": "18"} # doc_id=0
doc_1 = {"prompt": "A store sells...", "label": "42"} # doc_id=1
...
doc_499 = {"prompt": "If x+y=10...", "label": "7"} # doc_id=499
# ARC 有 1000 条 doc:
doc_0 = {"question": "Which factor...", "choices": {...}, "answerKey": "A"}
...
# doc 就是一个普通 Python 字典,字段名取决于你的 test.jsonl
# ═══════════════════════════════════════════════════
# Instance 层:doc → Instance 的映射关系
# ═══════════════════════════════════════════════════
# ── generate_until:1 doc → 1 Instance ──
# GSM8K 500 条 doc → 500 个 Instance
Task对象A.instances = [
Instance(request_type="generate_until", doc=doc_0, idx=0, doc_id=0, ...),
Instance(request_type="generate_until", doc=doc_1, idx=0, doc_id=1, ...),
... # 共 500 个
]
# ── multiple_choice:1 doc → N 个 Instance(N=选项数)──
# ARC 1000 条 doc,每条有 4 个选项 → 4000 个 Instance
Task对象B.instances = [
# doc_0 的 4 个 Instance(每个选项一个 loglikelihood 请求)
Instance(request_type="loglikelihood", doc=doc_0, idx=0, doc_id=0, ...), # 选项 A
Instance(request_type="loglikelihood", doc=doc_0, idx=1, doc_id=0, ...), # 选项 B
Instance(request_type="loglikelihood", doc=doc_0, idx=2, doc_id=0, ...), # 选项 C
Instance(request_type="loglikelihood", doc=doc_0, idx=3, doc_id=0, ...), # 选项 D
# doc_1 的 4 个 Instance
Instance(request_type="loglikelihood", doc=doc_1, idx=0, doc_id=1, ...), # 选项 A
Instance(request_type="loglikelihood", doc=doc_1, idx=1, doc_id=1, ...), # 选项 B
... # 共 4000 个
]
# idx 是选项索引(0~3),doc_id 是样本编号(0~999)
# 评分时用 doc_id 把同一道题的 4 个 Instance 找回来
evaluate() 循环中如何用 doc_id 关联
# evaluator.py 的评分循环(伪代码简化版):
for doc_id, doc in enumerate(task.test_docs()): # 遍历每条 doc
# 用 doc_id 从 4000 个 Instance 中筛出属于这道题的
requests = [x for x in task.instances if x.doc_id == doc_id]
requests.sort(key=lambda x: x.idx) # 按选项索引排序
# generate_until: requests = [Instance_0] → 1 个
# multiple_choice: requests = [Inst_0, Inst_1, Inst_2, Inst_3] → 4 个
# 提取模型输出
results = [req.filtered_resps[key] for req in requests]
# generate_until: results = ["18"]
# multiple_choice: results = [(-1.23,T,".."), (-4.56,F,".."), (-5.78,F,".."), (-6.12,F,"..")]
# 选项A的ll 选项B的ll 选项C的ll 选项D的ll
# 传给 process_results 计算这一道题的分数
metrics = task.process_results(doc, results) # → {"acc": 1.0, "acc_norm": 1.0}
三层关系总结图
# ┌─────────────────────────────────────────────────┐
# │ Task(评测任务) │
# │ gsm8k_5shot_generation.yaml │
# │ output_type: generate_until │
# │ │
# │ ┌─ doc_0 ─────────────────────────────────┐ │
# │ │ {"prompt":"Janet's ducks...","label":"18"} │ │
# │ │ │ │
# │ │ └→ Instance(doc_id=0, idx=0) │ │
# │ │ resps=["The answer is: 18."] │ │
# │ │ filtered_resps={"get-answer":"18"} │ │
# │ └─────────────────────────────────────────┘ │
# │ │
# │ ┌─ doc_1 ─────────────────────────────────┐ │
# │ │ {"prompt":"A store sells...","label":"42"} │ │
# │ │ │ │
# │ │ └→ Instance(doc_id=1, idx=0) │ │
# │ └─────────────────────────────────────────┘ │
# │ ...(共 500 条 doc → 500 个 Instance) │
# └─────────────────────────────────────────────────┘
#
# ┌─────────────────────────────────────────────────┐
# │ Task(评测任务) │
# │ arc_challenge.yaml │
# │ output_type: multiple_choice │
# │ │
# │ ┌─ doc_0 ─────────────────────────────────┐ │
# │ │ {"question":"Which factor...","answerKey":"A"}│
# │ │ │ │
# │ │ ├→ Instance(doc_id=0, idx=0) 选项A │ │
# │ │ ├→ Instance(doc_id=0, idx=1) 选项B │ │
# │ │ ├→ Instance(doc_id=0, idx=2) 选项C │ │
# │ │ └→ Instance(doc_id=0, idx=3) 选项D │ │
# │ └─────────────────────────────────────────┘ │
# │ ...(共 1000 条 doc → 4000 个 Instance) │
# └─────────────────────────────────────────────────┘
第三步:aggregation — 汇总所有样本的指标
所有样本评分完毕后,对每个指标调用 metric_list 中指定的聚合函数。
# 代码位置:evaluator.py:827-853
for (task_name, key, metric), items in vals.items():
# items = [1.0, 0.0, 1.0, 1.0, 0.0, ...] ← 每条样本的分数
# 1. 获取聚合函数(由 metric_list 中的 aggregation 字段指定)
agg_fn = task.aggregation()[metric]
# 常见 agg_fn:
# "mean" → 取平均(最常用)
# "weighted_perplexity" → 加权困惑度
# "bits_per_byte" → bits-per-byte
# 2. 聚合得到最终分数
results[task_name][metric] = agg_fn(items)
# results["gsm8k"]["exact_match"] = mean([1.0, 0.0, ...]) = 0.7234
# 3. 计算 bootstrap 标准误差
stderr_fn = stderr_for_metric(metric=agg_fn, bootstrap_iters=1000)
results[task_name][metric + "_stderr"] = stderr_fn(items)
# results["gsm8k"]["exact_match_stderr"] = 0.0128
metric_list 与 process_results 的关系
| 配置方式 | 谁计算逐条分数 | metric_list 的作用 |
|---|---|---|
只配 metric_list(无 process_results) |
框架内置逻辑 (根据 output_type 自动处理) |
① 定义用哪些指标(exact_match/acc/perplexity...) ② 定义聚合函数(mean/weighted_perplexity...) ③ 定义 higher_is_better |
同时配 process_results和 metric_list |
自定义函数 ( !function utils.process_results) |
① 不参与逐条计算(自定义函数全权负责) ② 仍然定义聚合函数(汇总时用) ③ 自定义函数返回的 key 必须与 metric 名匹配 |
完整数据流追踪:500 条 GSM8K 样本从 resps 到最终分数 example
# ═══ Step 1: apply_filters ═══
# 500 个 Instance,每个的 resps 是模型原始生成文本
# filter pipeline "get-answer": regex → regex → take_first
instance[0].filtered_resps = {"get-answer": "18"}
instance[1].filtered_resps = {"get-answer": "42"}
instance[2].filtered_resps = {"get-answer": "7"}
...
instance[499].filtered_resps = {"get-answer": "156"}
# ═══ Step 2: process_results 逐条计算 ═══
# GSM8K 没有自定义 process_results,框架用 metric_list 中的 exact_match
# 对每条:比较 filtered_resps 与 doc_to_target(doc)
# doc 0: filtered="18", target="18" → exact_match=1.0 ✓
# doc 1: filtered="42", target="35" → exact_match=0.0 ✗
# doc 2: filtered="7", target="7" → exact_match=1.0 ✓
# ...
# vals[("gsm8k","get-answer","exact_match")] = [1.0, 0.0, 1.0, ..., 0.0]
# ═══ Step 3: aggregation ═══
# metric_list 指定 aggregation: mean
# mean([1.0, 0.0, 1.0, ..., 0.0]) = 0.7234
# stderr = bootstrap([1.0, 0.0, ...], iters=1000) = 0.0128
# ═══ 最终输出 ═══
{
"gsm8k_5shot_generation": {
"exact_match,get-answer": 0.7234,
"exact_match_stderr,get-answer": 0.0128
}
}
# 注意 key 格式:"指标名,filter名"
# 如果有多个 filter pipeline,每个都会独立产生一组指标
端到端示例:RULER 任务的三个 !function 分别在哪一步触发 pipeline
RULER(长文本评测)的 YAML 同时使用了三个 !function 引用,是理解"哪个字段在哪一步被调用"的最佳示例:
# ruler_32k_cwe.yaml
task: ruler_32k_cwe
dataset_path: json
dataset_kwargs:
data_dir: /data/.../ruler/ruler_32k_cwe
output_type: generate_until
doc_to_text: input # ← 纯字段名
process_docs: !function utils.process_docs # ← !function ①
doc_to_target: !function utils.doc_to_target # ← !function ②
process_results: !function utils.process_results # ← !function ③
metric_list:
- metric: em
aggregation: mean
higher_is_better: true
generation_kwargs:
until: ["\n\n"]
max_gen_toks: 64
!function ① process_docs — 阶段二 · 数据加载后立即调用
# utils.py
def process_docs(dataset):
def _process(doc):
return {
"input": doc["input"], # 保留长文本 prompt
"outputs": doc["outputs"], # 保留答案列表(可能多个正确答案)
}
return dataset.map(_process)
# 调用时机:load_dataset() 之后,遍历 doc 之前
# 输入:整个 HF Dataset(所有样本)
# 输出:只保留 input 和 outputs 两个字段的 Dataset
# 效果:丢弃原始数据中不需要的字段,标准化字段名
!function ② doc_to_target — 阶段三 · Prompt 构建 + 阶段五 · 评分
# utils.py
def doc_to_target(doc):
return ' '.join(doc['outputs'])
# 调用时机 A(Prompt 构建):fewshot_context 中拼 few-shot 示例时
# 输入:doc = {"input": "...长文...", "outputs": ["Paris", "France"]}
# 输出:"Paris France"(作为 few-shot 示例的 target 文本)
# 调用时机 B(评分):process_results 内部对比时
# RULER 用自定义 process_results,所以此时不直接调用 doc_to_target
# 而是直接读 doc["outputs"]
# 注意:doc_to_target 在管线中可能被调用多次!
# - 每条 fewshot 示例调用一次(拼答案)
# - loglikelihood/mc 的 construct_requests 调用一次(构建 continuation)
# - generate_until 不在 construct_requests 中调用,但评分时可能用到
!function ③ process_results — 阶段五 · 模型推理结束后
# utils.py
def process_results(doc, results):
preds, refs = results, doc["outputs"]
preds = preds[0] # results = ["模型生成的文本"],取第一个
em = sum([1.0 if r.lower() in preds.lower() else 0.0 for r in refs]) / len(refs)
return {"em": em}
# 调用时机:apply_filters() 之后,evaluate() 循环中逐条调用
# 输入:
# doc = {"input": "...长文...", "outputs": ["Paris", "France"]}
# results = ["Paris is the capital"] ← filtered_resps 的值
# 执行:
# refs = ["Paris", "France"]
# "paris" in "paris is the capital" → 1.0 ✓
# "france" in "paris is the capital" → 0.0 ✗
# em = (1.0 + 0.0) / 2 = 0.5
# 输出:{"em": 0.5}
#
# 然后 metric_list 中 aggregation: mean 对所有样本的 em 取平均 → 最终分数
三个 !function 的调用顺序
# 时间线:
# ① process_docs(dataset) ← 数据加载后,最先执行,且只执行一次
# ↓
# (以下对每条 doc 循环)
# ② doc_to_target(doc) ← fewshot_context 中拼 few-shot 示例时调用
# ↓
# doc_to_text(doc) ← fewshot_context 中获取当前 doc 的 prompt
# ↓
# construct_requests() ← 创建 Instance
# ↓
# LM 推理 ← 模型生成文本
# ↓
# apply_filters() ← 无 filter_list 则跳过
# ↓
# ③ process_results(doc, results) ← 对模型输出评分,最后执行
三种 output_type 详解
output_type 决定了评测的数据流 —— 模型如何被调用、如何评分。框架支持四种类型,最常用的是以下三种:
逐选项计算"] M2 --> M3["取最高概率选项"] M3 --> M4["acc / acc_norm"] end subgraph loglikelihood L1["doc_to_text + doc_to_target"] --> L2["LM.loglikelihood()"] L2 --> L3["计算 perplexity"] end
generate_until — 生成型任务
模型接收 prompt,自由生成文本直到遇到停止词。适用于数学题、问答、代码生成等需要模型产出完整答案的场景。
数据流:doc → doc_to_text 生成 prompt → 模型生成文本 → filter_list 提取答案 → metric_list(默认 exact_match)或 process_results 自定义评分
完整示例 — GSM8K 数学评测:
group:
- generation
task: gsm8k_5shot_generation
dataset_path: json
dataset_kwargs:
data_dir: /data/.../gsm8k/gsm8k_5shot_generation
output_type: generate_until
test_split: test
doc_to_text: prompt # 直接使用数据集中的 "prompt" 字段
doc_to_target: label # 直接使用数据集中的 "label" 字段
metric_list:
- metric: exact_match
aggregation: mean
higher_is_better: true
ignore_case: true
regexes_to_ignore:
- ","
- "\\$"
- "(?s).*#### "
- "\n\n"
generation_kwargs:
until:
- "\n\n"
do_sample: false
temperature: 0.00001
num_fewshot: 0
filter_list:
- name: "get-answer"
filter:
- function: "regex"
regex_pattern: "The answer is: (.*)."
- function: "regex"
regex_pattern: "(\\-?[0-9\\.\\,]+)"
- function: "take_first"
metadata:
version: 1.0
另一个示例 — RULER 长文本评测(使用自定义 process_results):
task: ruler_32k_cwe
dataset_path: json
dataset_kwargs:
data_dir: /data/.../ruler/ruler_32k_cwe
output_type: generate_until
test_split: test
doc_to_text: input
process_docs: !function utils.process_docs
doc_to_target: !function utils.doc_to_target
process_results: !function utils.process_results
metric_list:
- metric: em
aggregation: mean
higher_is_better: true
generation_kwargs:
until:
- "\n\n"
do_sample: false
temperature: 0.0001
max_gen_toks: 64
完整管线追踪:GSM8K 一道题从头到尾经过了什么 pipeline
Stage 0 — 原始数据(test.jsonl 中的一行)
{"prompt": "Janet's ducks lay 16 eggs per day. She eats three for breakfast every morning and bakes muffins for her friends every day with four. She sells every duck egg at the farmers' market daily for $2. How much in dollars does she make every day at the farmers' market?\nThe answer is:", "label": "18"}
这就是数据集中的一条原始文档 doc,是一个 Python 字典。
Stage 1 — doc_to_text(doc) → prompt 字符串
YAML 配置 doc_to_text: prompt(纯字段名模式),框架执行:
# 框架内部等价于:
text = doc["prompt"]
# 结果:
"Janet's ducks lay 16 eggs per day. She eats three for breakfast every morning and bakes muffins for her friends every day with four. She sells every duck egg at the farmers' market daily for $2. How much in dollars does she make every day at the farmers' market?\nThe answer is:"
Stage 2 — fewshot_context(doc, num_fewshot=0) → ctx 字典
因为 num_fewshot: 0(zero-shot),fewshot_context 直接把 description(空字符串)和 doc_to_text 的结果拼接:
# fewshot_context 返回值(ctx)—— 注意是字典,不是纯字符串:
{
"text": "Janet's ducks lay 16 eggs per day...The answer is:"
}
# 如果 num_fewshot > 0,text 会在前面拼上 few-shot 示例:
# {"text": "Q: What is 2+2?\nThe answer is: 4\n\nQ: ...\n\n实际问题的 prompt"}
Stage 3 — construct_requests(doc, ctx) → Instance 对象
output_type: generate_until,框架创建一个 Instance:
Instance(
request_type = "generate_until",
doc = {"prompt": "Janet's ducks...", "label": "18"}, # 原始文档保留
arguments = (
{"text": "Janet's ducks lay 16 eggs...The answer is:"}, # ctx
{"until": ["\n\n"], "do_sample": false, "temperature": 0.00001} # generation_kwargs
),
idx = 0,
metadata = ("gsm8k_5shot_generation", 0, 1), # (task_name, doc_id, repeats)
resps = [], # 待填充
filtered_resps = {}, # 待填充
)
Stage 4 — 模型推理 LM.generate_until() → resps
vllm_norm 将 ctx 转为 HTTP payload 发送给 vLLM 服务,返回的生成文本存入 resps:
# 模型生成文本经 thinking token 处理后写入 instance.resps:
instance.resps = [
"She sells 16 - 3 - 4 = 9 duck eggs a day. 9 * $2 = $18. The answer is: 18."
]
# resps 是列表,长度 = repeats(默认 1)
Stage 5 — apply_filters() → filtered_resps
按 filter_list 中配置的 pipeline 顺序执行过滤器:
# 输入(instance.resps 列表):
["She sells 16 - 3 - 4 = 9 duck eggs a day. 9 * $2 = $18. The answer is: 18."]
# Step 1 regex "The answer is: (.*)."
# 捕获组 → ["18"]
# Step 2 regex "(\\-?[0-9\\.\\,]+)"
# 提取纯数字 → ["18"]
# Step 3 take_first
# 列表 → 单值 → "18"
# 最终写入:
instance.filtered_resps["get-answer"] = "18"
Stage 6 — process_results(doc, results) → 指标字典
GSM8K 使用默认的 exact_match,未配置自定义 process_results:
# results = ["18"](filtered_resps 中的值)
# doc_to_target(doc) = doc["label"] = "18"
# exact_match 计算(考虑 ignore_case、regexes_to_ignore):
# prediction = "18" vs reference = "18" → 完全匹配
# 返回:
{"exact_match": 1.0}
Stage 7 — aggregation → 最终分数
# 所有样本的 exact_match 值:[1.0, 0.0, 1.0, 1.0, 0.0, ...]
# aggregation: mean → 取平均
# 最终结果:
{
"gsm8k_5shot_generation": {
"exact_match": 0.7234, # 准确率
"exact_match_stderr": 0.0128 # 标准误差
}
}
multiple_choice — 选择题
框架将 prompt 与每个选项拼接,分别计算 loglikelihood,选择概率最高的选项作为答案。适用于常识推理、阅读理解选择题等。
数据流:doc → doc_to_text 生成 prompt + doc_to_choice 获取选项列表 → 对每个选项计算 loglikelihood → 取 argmax → 与 doc_to_target(正确选项索引)比较
acc 是原始准确率(取 loglikelihood 最高的选项);acc_norm 是长度归一化准确率(loglikelihood 除以选项 token 数),避免模型偏好长选项。
完整示例 — ARC Challenge(Jinja2 模板):
group:
- ai2_arc
task: arc_challenge_local
num_fewshot: 25
dataset_path: json
dataset_kwargs:
data_dir: /data/.../arc/arc_challenge
output_type: multiple_choice
test_split: test
doc_to_text: "Question: {{question}}\nAnswer:"
doc_to_target: "{{choices.label.index(answerKey)}}"
doc_to_choice: "{{choices.text}}"
metric_list:
- metric: acc
aggregation: mean
higher_is_better: true
- metric: acc_norm
aggregation: mean
higher_is_better: true
完整示例 — HellaSwag(process_docs + 字段引用):
task: hellaswag_local
num_fewshot: 10
dataset_path: json
dataset_kwargs:
data_dir: /data/.../hellaswag/hellaswag
output_type: multiple_choice
test_split: test
process_docs: !function utils.process_docs
doc_to_text: "{{query}}"
doc_to_target: "{{label}}"
doc_to_choice: "choices"
metric_list:
- metric: acc
aggregation: mean
higher_is_better: true
- metric: acc_norm
aggregation: mean
higher_is_better: true
完整管线追踪:ARC 一道选择题从头到尾经过了什么 pipeline
Stage 0 — 原始数据(test.jsonl 中的一行)
{
"question": "Which factor will most likely cause a person to develop a fever?",
"choices": {
"text": ["a]viral infection", "an its injury", "a bee sting", "a broken bone"],
"label": ["A", "B", "C", "D"]
},
"answerKey": "A"
}
Stage 1 — doc_to_text / doc_to_target / doc_to_choice 渲染
三个 Jinja2 模板分别渲染:
# doc_to_text: "Question: {{question}}\nAnswer:"
text = "Question: Which factor will most likely cause a person to develop a fever?\nAnswer:"
# doc_to_target: "{{choices.label.index(answerKey)}}"
target = 0 # choices.label = ["A","B","C","D"], answerKey = "A", index = 0
# doc_to_choice: "{{choices.text}}"
choices = ["a viral infection", "an insect injury", "a bee sting", "a broken bone"]
Stage 2 — fewshot_context(doc, num_fewshot=25) → ctx 字典
从 train split 中随机采样 25 条示例,拼接成 few-shot prompt:
# ContextSampler 对每条 few-shot 示例的处理:
# doc_to_target(fewshot_doc) 返回 int(如 0)
# 但 sampler 会自动转为选项文本:
# doc_to_choice(fewshot_doc)[0] → "a viral infection"
# 见 samplers.py:76: str(self.doc_to_choice(doc)[self.doc_to_target(doc)])
# fewshot_context 返回值(ctx):
{
"text": "Question: What is the most common cause of earthquakes?\nAnswer: tectonic plate movement\n\nQuestion: Which best describes a mineral?\nAnswer: a naturally occurring inorganic solid\n\n... (共 25 个示例,target 都是选项文字而非索引数字) ...\n\nQuestion: Which factor will most likely cause a person to develop a fever?\nAnswer:"
}
# 每个 few-shot 示例 = doc_to_text(doc) + target_delimiter + choices[target_index]
# 示例间用 fewshot_delimiter(默认 "\n\n")分隔
Stage 3 — construct_requests(doc, ctx) → Instance 列表
output_type: multiple_choice,框架为每个选项创建一个 Instance(共 4 个):
# 将 ctx 与每个 choice 拼接,创建 4 个 loglikelihood 请求:
[
Instance(
request_type = "loglikelihood",
doc = {原始 doc},
arguments = (
{"text": "...25 fewshot examples...\nQuestion: Which factor...?\nAnswer:"}, # ctx
" a viral infection", # target_delimiter + choice[0]
{...generation_kwargs}
),
idx = 0, # 第 0 个选项
),
Instance(
request_type = "loglikelihood",
arguments = (ctx, " an insect injury", ...),
idx = 1, # 第 1 个选项
),
Instance(
request_type = "loglikelihood",
arguments = (ctx, " a bee sting", ...),
idx = 2, # 第 2 个选项
),
Instance(
request_type = "loglikelihood",
arguments = (ctx, " a broken bone", ...),
idx = 3, # 第 3 个选项
),
]
# 注意:arguments 的第二个元素前有 target_delimiter(默认 " ")
Stage 4 — 模型推理 LM.loglikelihood() → resps
对每个 Instance,模型计算 continuation(选项文本)在给定 context 条件下的 log probability:
# ═══════════════════════════════════════════════════
# loglikelihood 计算原理(以选项 "a viral infection" 为例)
# ═══════════════════════════════════════════════════
# 1. 拼接完整文本:context + continuation
full_text = "Question: ...?\nAnswer:" + " a viral infection"
# 2. 分别 tokenize context 和 continuation
context_tokens = [tok1, tok2, ..., tok_N] # N 个 token
continuation_tokens = [tok_A, tok_B, tok_C] # 3 个 token(" a", " viral", " infection")
all_tokens = context_tokens + continuation_tokens # 拼在一起
# 3. 构建 mask:context 位置=0,continuation 位置=1
mask = [0]*N + [1, 1, 1]
# 4. 发送给 vLLM,设置 echo=True + logprobs=1 + max_tokens=1
# → 模型返回每个 token 位置的 log probability
# 不生成新 token,只是计算已有 token 的概率
# 5. 只累加 continuation 部分的 log prob(mask=1 的位置)
continuation_logprobs = log_p(" a") + log_p(" viral") + log_p(" infection")
# = (-0.31) + (-0.52) + (-0.40)
# = -1.23
# ═══════════════════════════════════════════════════
# is_greedy 判断原理
# ═══════════════════════════════════════════════════
# 对 continuation 的每个 token 位置,检查:
# 该位置概率最高的 token(top_token) == 实际的 continuation token?
# 如果每个位置都匹配 → is_greedy = True
# 位置 N+0: top_token=" a" == 实际 token " a" ✓
# 位置 N+1: top_token=" viral" == 实际 token " viral" ✓
# 位置 N+2: top_token=" infection" == 实际 token " infection" ✓
# → is_greedy = True (模型贪心解码恰好会生成这个选项)
# 如果某个位置 top_token ≠ 实际 token → is_greedy = False
# 例如:位置 N+1 的 top_token 是 " bacterial" 而非 " viral" → False
4 个选项的计算结果:
# 每个 Instance 的 resps = [(continuation_logprobs, is_greedy, info_dict)]
instances[0].resps = [(-1.23, True, "[{...}]")] # " a viral infection" ← logprob 最高
instances[1].resps = [(-4.56, False, "[{...}]")] # " an insect injury"
instances[2].resps = [(-5.78, False, "[{...}]")] # " a bee sting"
instances[3].resps = [(-6.12, False, "[{...}]")] # " a broken bone"
# loglikelihood 是负数,越接近 0 = 模型认为越可能
# is_greedy = True 表示模型贪心解码恰好会生成这段文本
Stage 5 — process_results(doc, results) → 指标字典
框架自动汇总 4 个 Instance 的结果,计算 acc 和 acc_norm:
# results = [(-1.23, True, {}), (-4.56, False, {}), (-5.78, False, {}), (-6.12, False, {})]
lls = [-1.23, -4.56, -5.78, -6.12]
# acc: 取 argmax(lls) = 0
pred = 0
gold = 0 # doc_to_target 返回的正确索引
acc = 1.0 if pred == gold else 0.0 # → 1.0 ✓
# acc_norm: 除以选项 token 长度后再取 argmax
completion_len = [len("a viral infection"), len("an insect injury"), len("a bee sting"), len("a broken bone")]
# = [17, 16, 10, 12]
norm_lls = [ll/length for ll, length in zip(lls, completion_len)]
# = [-0.072, -0.285, -0.578, -0.510]
pred_norm = argmax(norm_lls) = 0
acc_norm = 1.0 if pred_norm == gold else 0.0 # → 1.0 ✓
# 返回:
{"acc": 1.0, "acc_norm": 1.0}
Stage 6 — aggregation → 最终分数
# 所有样本的指标取 mean:
{
"arc_challenge_local": {
"acc": 0.5623,
"acc_stderr": 0.0145,
"acc_norm": 0.5891,
"acc_norm_stderr": 0.0144
}
}
loglikelihood — 困惑度评测
模型计算给定文本(prompt + target)的 log probability,用于评估模型对文本的困惑度。不生成新文本。
数据流:doc → doc_to_text 获取 context → doc_to_target 获取 continuation → 拼接后计算 loglikelihood → perplexity 指标
完整示例 — BBH 困惑度:
task: bbh_5shot_ppl
dataset_path: json
dataset_kwargs:
data_dir: /data/.../bbh
output_type: loglikelihood
test_split: test
doc_to_text: prompt
doc_to_target: text
metric_list:
- metric: word_perplexity
aggregation: weighted_perplexity
higher_is_better: false
- metric: byte_perplexity
aggregation: weighted_perplexity
higher_is_better: false
- metric: bits_per_byte
aggregation: bits_per_byte
higher_is_better: false
output_type 对比总结
| 类型 | 模型调用 | 需要 doc_to_choice | 默认指标 | 典型场景 |
|---|---|---|---|---|
generate_until | LM.generate_until() | 否 | exact_match | 数学题、问答、代码生成 |
multiple_choice | LM.loglikelihood() × N | 是 | acc, acc_norm | 常识推理、阅读理解 |
loglikelihood | LM.loglikelihood() | 否 | perplexity, acc | 困惑度评测 |
Prompt 模板语法
doc_to_text、doc_to_target、doc_to_choice 支持三种写法,灵活程度逐级递增:
方式一:纯字段名
直接使用数据集中的字段名,框架会自动读取 doc["field_name"] 的值:
doc_to_text: prompt # 等价于 doc["prompt"]
doc_to_target: label # 等价于 doc["label"]
适用场景:数据集中已经有现成的 prompt 字段,无需额外处理。
方式二:Jinja2 模板
使用 {{field}} 语法插值,支持 Python 表达式:
# 简单插值
doc_to_text: "Question: {{question}}\nAnswer:"
# 列表索引 —— 找到正确答案在 choices.label 中的索引
doc_to_target: "{{choices.label.index(answerKey)}}"
# 直接引用嵌套字段的列表
doc_to_choice: "{{choices.text}}"
适用场景:需要组合多个字段、做简单格式化,但逻辑不太复杂。
Jinja2 模板中引用的字段名必须与数据集中的实际字段名完全匹配。doc_to_choice 用 Jinja2 时,表达式结果必须是一个列表(如 {{choices.text}} 返回 ["选项A", "选项B", ...])。
方式三:!function 引用
使用 YAML 的自定义标签 !function 引用同目录下 utils.py 中的 Python 函数:
doc_to_text: !function utils.doc_to_text
doc_to_target: !function utils.doc_to_target
process_docs: !function utils.process_docs
process_results: !function utils.process_results
适用场景:需要复杂逻辑(循环、条件判断、多字段组合格式化等),Jinja2 无法胜任。
三种方式对比
| 方式 | 语法示例 | 灵活度 | 适用场景 |
|---|---|---|---|
| 纯字段名 | prompt | 低 | 数据集已有现成 prompt |
| Jinja2 | "Q: {{question}}\nA:" | 中 | 字段拼接、简单格式化 |
| !function | !function utils.doc_to_text | 高 | 复杂逻辑、循环、条件 |
具体示例:同一条数据走过三种 Prompt 模板方式 示例
# 假设数据集中有这样一条文档:
doc = {
"question": "What is the capital of France?",
"choices": {"text": ["London", "Paris", "Berlin", "Madrid"], "label": ["A", "B", "C", "D"]},
"answerKey": "B",
"prompt": "Question: What is the capital of France?\nA. London\nB. Paris\nC. Berlin\nD. Madrid\nAnswer:"
}
# ═══════════════════════════════════════════════════
# 方式一:纯字段名
# ═══════════════════════════════════════════════════
# YAML: doc_to_text: prompt
# 框架执行: doc["prompt"]
# 输出 →
"Question: What is the capital of France?\nA. London\nB. Paris\nC. Berlin\nD. Madrid\nAnswer:"
# ═══════════════════════════════════════════════════
# 方式二:Jinja2 模板
# ═══════════════════════════════════════════════════
# YAML: doc_to_text: "Question: {{question}}\nAnswer:"
# 框架执行: jinja2.Template("Question: {{question}}\nAnswer:").render(doc)
# 输出 →
"Question: What is the capital of France?\nAnswer:"
# YAML: doc_to_target: "{{choices.label.index(answerKey)}}"
# 框架执行: jinja2 求值 choices.label.index("B") → 1
# 输出 → 1 (即选项 B 的索引)
# YAML: doc_to_choice: "{{choices.text}}"
# 框架执行: jinja2 求值 → ["London", "Paris", "Berlin", "Madrid"]
# ═══════════════════════════════════════════════════
# 方式三:!function 引用
# ═══════════════════════════════════════════════════
# YAML: doc_to_text: !function utils.doc_to_text
# 框架执行: utils.doc_to_text(doc)
# 输出 →(由 Python 函数自由定制)
"Question: What is the capital of France?\nA. London\nB. Paris\nC. Berlin\nD. Madrid\nAnswer: "
utils.py 编写指南
当 YAML 的字段名或 Jinja2 模板无法满足需求时,需要在同目录下创建 utils.py,编写自定义函数。!function utils.func_name 语法会在加载 YAML 时自动解析为对应的函数引用。
process_docs(dataset) — 数据预处理
签名:process_docs(dataset: datasets.Dataset) -> datasets.Dataset
调用时机:数据集加载后、评测开始前。用于清洗、转换、添加字段。
示例 — HellaSwag 文本清洗:
import datasets
import re
def preprocess(text):
text = text.strip()
text = text.replace(" [title]", ". ")
text = re.sub("\\[.*?\\]", "", text)
text = text.replace(" ", " ")
return text
def process_docs(dataset: datasets.Dataset) -> datasets.Dataset:
def _process_doc(doc):
ctx = doc["ctx_a"] + " " + doc["ctx_b"].capitalize()
out_doc = {
"query": preprocess(doc["activity_label"] + ": " + ctx),
"choices": [preprocess(ending) for ending in doc["endings"]],
"gold": int(doc["label"]),
}
return out_doc
return dataset.map(_process_doc)
示例 — DROP 数据预处理:
def process_docs(dataset):
def _process(doc):
return {
"id": doc["query_id"],
"passage": doc["passage"],
"question": doc["question"],
"answers": get_answers(doc), # 解析多种答案格式
}
return dataset.map(_process)
doc_to_text(doc) — 自定义 prompt
签名:doc_to_text(doc: dict) -> str
调用时机:为每个样本构建输入 prompt 时调用。
def doc_to_text(doc):
# 将选项格式化为 "A. xxx\nB. xxx\n..." 的形式
option_text = '\n'.join(
d['label'] + '. ' + d['text']
for d in doc['question']['choices']
)
return '\n'.join([
'Question: ' + doc['question']['stem'],
option_text,
'Answer: '
])
doc_to_target(doc) — 自定义目标答案
签名:doc_to_target(doc: dict) -> str
def doc_to_target(doc):
return ' '.join(doc['outputs'])
process_results(doc, results) — 自定义评分
签名:process_results(doc: dict, results: list) -> dict
调用时机:模型推理完成后,对每个样本计算指标。返回 dict,key 为指标名,value 为分数。
process_results 返回的 key 必须与 metric_list 中的 metric 名称匹配。例如返回 {"em": 0.5, "f1": 0.7},则 metric_list 中需要有 metric: em 和 metric: f1。
示例 — RULER 精确匹配:
def process_results(doc, results):
preds, refs = results, doc["outputs"]
preds = preds[0]
# 计算每个参考答案是否出现在预测中,取平均
em = sum([1.0 if r.lower() in preds.lower() else 0.0 for r in refs]) / len(refs)
return {"em": em}
示例 — DROP EM + F1:
def process_results(doc, results):
preds, golds = results, doc["answers"]
max_em = 0
max_f1 = 0
for gold_answer in golds:
exact_match, f1_score = get_metrics(preds, gold_answer)
if gold_answer[0].strip():
max_em = max(max_em, exact_match)
max_f1 = max(max_f1, f1_score)
return {"em": max_em, "f1": max_f1}
filter_list 输出过滤
filter_list 对 generate_until 的模型输出做后处理,通常用于从自由文本中提取最终答案。过滤器以 pipeline 方式串行执行。
内置 filter 类型
| filter | 说明 | 参数 |
|---|---|---|
regex | 正则表达式匹配,提取第一个捕获组 | regex_pattern |
remove_whitespace | 去除首尾空白字符 | 无 |
take_first | 取列表中的第一个元素(必须作为 pipeline 的最后一步) | 无 |
示例 — GSM8K 答案提取
模型输出可能是 "Let me solve this... The answer is: 42.",需要提取数字 42:
filter_list:
- name: "get-answer"
filter:
- function: "regex"
regex_pattern: "The answer is: (.*)." # 第一步:提取 "The answer is: ..." 后面的内容
- function: "regex"
regex_pattern: "(\\-?[0-9\\.\\,]+)" # 第二步:从中提取数字
- function: "take_first" # 第三步:取第一个匹配结果
具体示例:filter pipeline 逐步数据流追踪 示例
# 模型原始生成输出(generate_until 的返回值)
raw_output = "Let me solve this step by step.\n16 - 3 = 13 eggs eaten\n13 * 2 = 26 dollars at market\nThe answer is: 42."
# Instance.resps = [["Let me solve this step by step.\n16 - 3 = 13...\nThe answer is: 42."]]
# (外层列表对应 repeats,内层是单次输出)
# ═══════════════════════════════════════════════════
# filter pipeline 逐步执行:
# ═══════════════════════════════════════════════════
# Step 1: regex — regex_pattern: "The answer is: (.*)."
# 输入: ["Let me solve this step by step.\n...\nThe answer is: 42."]
# 匹配捕获组 (.*) → "42"
# 输出: ["42"]
# Step 2: regex — regex_pattern: "(\\-?[0-9\\.\\,]+)"
# 输入: ["42"]
# 匹配纯数字 → "42"
# 输出: ["42"]
# Step 3: take_first
# 输入: ["42"] (列表)
# 取第一个元素
# 输出: "42" (字符串)
# 最终:instance.filtered_resps["get-answer"] = "42"
# 与 doc_to_target 的 "42" 进行 exact_match 比较 → 1.0
示例 — BBH CoT 答案提取
CoT 推理的输出通常以 "the answer is X." 结尾:
filter_list:
- name: "get-answer"
filter:
- function: "regex"
regex_pattern: "(?<=the answer is )(.*?)(?=\\.)" # 提取 "the answer is " 后到 "." 前的内容
- function: "take_first"
示例 — 简单空白清理
filter_list:
- name: remove_whitespace
filter:
- function: remove_whitespace
- function: take_first
每个 filter pipeline 的最后一步通常是 take_first。这是因为每个 Instance 可能有 repeats 次输出(列表),take_first 从列表中选取第一个作为最终答案。即使 repeats=1,也需要这一步来从单元素列表中提取值。
YAML 继承 (include)
当多个子任务共享大部分配置时,可以使用 include 字段实现模板继承,避免重复。
工作机制
子任务 YAML 中使用 include: "_template_file_name" 引用同目录下的模板文件。框架先加载模板配置,再用子任务的字段覆盖。
示例 — BBH CoT Few-shot
模板文件(定义所有子任务共享的配置):
group:
- bbh
- bbh_cot_fewshot
dataset_path: lukaemon/bbh
dataset_kwargs:
data_dir: /data/.../bbh/cot_fewshot
output_type: generate_until
test_split: test
doc_to_target: "{{target}}"
metric_list:
- metric: exact_match
aggregation: mean
higher_is_better: true
generation_kwargs:
max_gen_toks: 2048
until:
- "\n\n"
do_sample: false
temperature: 0.0
filter_list:
- name: "get-answer"
filter:
- function: "regex"
regex_pattern: "(?<=the answer is )(.*?)(?=\\.)"
- function: "take_first"
num_fewshot: 0
子任务文件(只覆盖 task、doc_to_text、dataset_kwargs):
include: "_cot_fewshot_template_yaml"
task: bbh_cot_fewshot_boolean_expressions
description: "Evaluate the result of a random Boolean expression.\n\n"
doc_to_text: "Q: not ( ( not not True ) ) is\nA: Let's think step by step.\n...(few-shot 示例)...\n\nQ: {{input}}\nA: Let's think step by step.\n"
dataset_kwargs:
data_dir: /data/.../bbh_fewshot/boolean_expressions/
BBH Benchmark 有 27 个子任务,每个子任务只需要不同的 doc_to_text(few-shot 示例)和 data_dir,其他配置完全相同。使用 include 继承避免了在 27 个文件中重复相同的 generation_kwargs、filter_list、metric_list 等配置。
数据集准备
大部分评测任务使用本地 JSON 数据集,通过 HuggingFace datasets 库的 JSON 加载器读取。
YAML 配置
dataset_path: json # 使用 HF datasets 的 JSON 加载器
dataset_kwargs:
data_dir: /data/.../my_task/ # 数据文件所在目录
dataset_name: null # 不使用子集
test_split: test # 使用 test split
目录结构
data_dir 目录下按 split 名放置 JSON 文件:
JSON 格式要求
每行一个 JSON 对象(JSONL 格式),字段名与 doc_to_text / doc_to_target 中引用的字段对应:
{"prompt": "What is 2+2?", "label": "4"}
{"prompt": "What is 3*5?", "label": "15"}
{"prompt": "What is 10/2?", "label": "5"}
对于 multiple_choice 任务,数据集需要包含选项信息:
{"question": "What is the capital of France?", "choices": {"text": ["London", "Paris", "Berlin", "Madrid"], "label": ["A", "B", "C", "D"]}, "answerKey": "B"}
当 dataset_path: json 时,框架调用 datasets.load_dataset("json", data_dir=...)。HuggingFace 会自动根据文件名识别 split:test.jsonl → test split,train.jsonl → train split。也支持 .json 扩展名。
二阶段后处理
大部分评测任务在一阶段(lm_eval 内部)即可完成评分。但有些任务需要二阶段后处理 —— 在 second_stage_all.sh 中调用额外的评分脚本。
何时需要二阶段
| 需要二阶段 | 不需要二阶段 |
|---|---|
| 代码执行评测(HumanEval, MBPP, BigCodeBench)—— 需要运行生成的代码 | 数学题(GSM8K, DROP)—— 通过 exact_match 或 process_results 评分 |
| SWE-Bench —— 需要 apply patch 并运行测试 | 选择题(ARC, HellaSwag)—— 通过 loglikelihood 对比评分 |
| 需要 GPT-4 裁判评分的任务 | 困惑度评测(BBH ppl)—— 直接计算 perplexity |
| 复杂数学验证(AIME, OlympiadBench)—— 需要符号计算 | 长文本评测(RULER)—— 通过 process_results 评分 |
在 second_stage_all.sh 中添加条目
second_stage_all.sh 使用 shell 模式匹配来检测当前任务是否需要后处理:
# 代码执行类任务
if [[ $TASKS == *"multipl-e"* || $TASKS == *"humaneval_"*"_generation"* ]]; then
python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/multipl-e/score.py
fi
# 数学竞赛类任务
if [[ $TASKS == *"aime24"* ]]; then
python3 -u lm-evaluation-harness/lm_eval/tasks/sft/aime24/score.py
fi
# 添加新任务的二阶段:
if [[ $TASKS == *"my_new_task"* ]]; then
python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/my_new_task/score.py
fi
score.py 的典型结构
二阶段评分脚本通常读取一阶段的 *_log_samples.jsonl 文件,对每个样本执行评分,将结果写入 OUTPUT_PATH:
import json
import os
eval_dir = os.environ.get("OUTPUT_PATH")
# 1. 读取一阶段的模型输出
with open(os.path.join(eval_dir, "my_task_log_samples.jsonl")) as f:
samples = json.load(f)
# 2. 对每个样本执行评分(代码执行、API 调用等)
results = []
for sample in samples:
prediction = sample["resps"][0][0] # 模型输出
score = evaluate(prediction) # 自定义评分逻辑
results.append(score)
# 3. 汇总并写入结果
avg_score = sum(results) / len(results)
with open(os.path.join(eval_dir, "my_task_scores.txt"), "w") as f:
f.write(f"my_task\t{avg_score}\n")
如果你的评测只需要字符串匹配(exact_match, regex 提取, F1 等),在 YAML 的 metric_list 和 process_results 中处理即可,不需要二阶段。只有需要执行代码、调用外部 API、运行测试等操作时,才需要编写 score.py 并添加到 second_stage_all.sh。
完整实战:添加一个新评测
以下是从零开始添加一个评测任务的 step-by-step 指南。我们以一个 generate_until 类型的数学评测为例。
创建一个名为 my_math_eval 的评测任务,模型回答数学问题,用 exact_match 评分。
Step 1:准备数据集
创建数据目录并准备 JSONL 格式的测试数据:
/data/evaluation/datasets/pretrain/my_math_eval/
└── test.jsonl
{"question": "What is 15 * 23?", "answer": "345"}
{"question": "What is sqrt(144)?", "answer": "12"}
{"question": "What is 2^10?", "answer": "1024"}
Step 2:创建任务目录
mkdir -p lm_eval/tasks/pretrain/my_math_eval/
Step 3:编写 YAML 配置
group:
- generation
task: my_math_eval
dataset_path: json
dataset_kwargs:
data_dir: /data/evaluation/datasets/pretrain/my_math_eval
dataset_name: null
output_type: generate_until
test_split: test
num_fewshot: 0
# 输入:使用 Jinja2 模板拼接 prompt
doc_to_text: "Please solve the following math problem. Give only the final numerical answer.\n\nQuestion: {{question}}\nAnswer:"
# 目标答案
doc_to_target: answer
# 生成参数
generation_kwargs:
until:
- "\n\n"
do_sample: false
temperature: 0.0
max_gen_toks: 64
# 输出过滤:提取数字答案
filter_list:
- name: "extract-number"
filter:
- function: "regex"
regex_pattern: "(\\-?[0-9\\.\\,]+)"
- function: "take_first"
# 评分指标
metric_list:
- metric: exact_match
aggregation: mean
higher_is_better: true
ignore_case: true
metadata:
version: 1.0
Step 4(可选):编写 utils.py
如果需要自定义评分逻辑(例如允许数值容差),可以添加 utils.py:
def process_results(doc, results):
"""自定义评分:允许浮点数容差"""
prediction = results[0]
reference = doc["answer"]
try:
pred_num = float(prediction.replace(",", ""))
ref_num = float(reference.replace(",", ""))
is_correct = abs(pred_num - ref_num) < 1e-6
except (ValueError, TypeError):
is_correct = prediction.strip() == reference.strip()
return {"exact_match": float(is_correct)}
如果使用自定义 process_results,需要在 YAML 中添加:
process_results: !function utils.process_results
Step 5:验证注册
# 确认任务已注册
python3 -u lm_eval/__main__.py \
--tasks list \
--include_path lm_eval/tasks/pretrain/ \
--verbosity DEBUG 2>&1 | grep my_math_eval
如果输出中包含 my_math_eval,说明注册成功。
Step 6:运行评测
export TASKS=my_math_eval
export INCLUDE_PATH=/path/to/lm_eval/tasks/pretrain/
export OUTPUT_PATH=/output/eval_results/
export MODEL_DIR=/path/to/model/
export SERVER_IP_DIR=/path/to/server_ips/
python3 -u lm_eval/__main__.py \
--model vllm_norm \
--model_args url=http://127.0.0.1:5002/playground \
--tasks $TASKS \
--output_path $OUTPUT_PATH/eval_results.json \
--log_samples \
--include_path $INCLUDE_PATH \
--verbosity DEBUG
Step 7(可选):添加二阶段后处理
如果你的评测需要执行代码或调用外部 API 评分,在 second_stage_all.sh 中添加:
if [[ $TASKS == *"my_math_eval"* ]]; then
python3 -u lm-evaluation-harness/lm_eval/tasks/pretrain/my_math_eval/score.py
fi
完整清单
| 步骤 | 产出物 | 必需 |
|---|---|---|
| 1. 准备数据集 | data_dir/test.jsonl | 是 |
| 2. 创建任务目录 | lm_eval/tasks/pretrain/my_task/ | 是 |
| 3. 编写 YAML | my_task.yaml | 是 |
| 4. 编写 utils.py | utils.py | 按需 |
| 5. 验证注册 | 确认 --tasks list 输出包含任务名 | 是 |
| 6. 运行评测 | eval_results.json + *_log_samples.jsonl | 是 |
| 7. 二阶段后处理 | score.py + second_stage_all.sh 条目 | 按需 |