企业官网建设流程全解析

OpenAI Evals框架的扩展实践：自定义评测任务的注册与结果分析

一、Evals的插件架构设计

OpenAI Evals是一个面向LLM评测的开源框架，其核心理念是"评测即代码"——每个评测任务是一个独立的Python模块，通过声明式接口注册到框架中。这种设计使得评测任务可以像代码一样被版本控制、review和共享，避免了传统benchmark"静态数据集+独立评测脚本"模式下的复现难题。

Evals的架构分为三层：(1) 注册层——通过装饰器将评测函数注册到全局注册表；(2) 执行层——oaievalCLI工具加载指定任务、调用模型、收集结果；(3) 报告层——标准化输出格式（JSON/Markdown），支持结果聚合和比较。理解这一架构后，自定义任务的开发就是填充三个接口：数据集定义、评测逻辑和指标计算。

二、自定义任务的注册与实现

Evals通过@evals.registration.register装饰器来注册评测任务。任务的核心实现分为两类：(1) 基于提示模板的简单任务——继承evals.elsuite.basic.match等内置suite；(2) 完全自定义逻辑的任务——继承evals.api.Eval基类。

import evals import evals.api import evals.record from evals.elsuite import utils from typing import Iterator, Dict, Any, List import json # ===== 方式一：使用内置suite快速定义选择题评测 ===== @evals.registration.register("custom_mmlu_variant") class CustomMMLUVariant(evals.elsuite.basic.match.Match): """基于内置Match suite的自定义多选题评测。 通过继承Match类，自动获得： - 精确匹配和模糊匹配逻辑 - 标准化的准确率指标 - 结果日志记录 """ def __init__( self, samples_jsonl: str = "data/custom_mmlu.jsonl", max_samples: int = None, *args, **kwargs ): super().__init__( samples_jsonl=samples_jsonl, max_samples=max_samples, *args, **kwargs ) def pre_process(self, sample: Dict) -> str: """将样本转换为模型输入的prompt。 构造包含题目和选项的完整prompt， 控制few-shot示例的选取。 Args: sample: 单个数据样本，含question、choices、subject字段 Returns: 格式化后的prompt字符串 """ prompt = f"学科：{sample['subject']}\n\n" prompt += f"问题：{sample['question']}\n\n" # 构造选项列表 for i, choice in enumerate(sample['choices']): letter = chr(65 + i) # A, B, C, D prompt += f"{letter}. {choice}\n" prompt += "\n答案：(" return prompt # ===== 方式二：完全自定义评测逻辑 ===== @evals.registration.register("custom_reasoning_eval") class CustomReasoningEval(evals.api.Eval): """完全自定义的推理评测任务。 不依赖内置suite，完整控制： - 数据集构造 - 模型调用方式 - 答案提取和评判 - 指标聚合 """ def __init__( self, samples_jsonl: str, max_samples: int = None, *args, **kwargs ): super().__init__(*args, **kwargs) self.samples_jsonl = samples_jsonl self.max_samples = max_samples def eval_sample(self, sample: Dict, rng) -> None: """评测单个样本。 这是自定义评测的核心方法。 包含完整的"调用模型→提取答案→判断正确性"流程。 Args: sample: 单个评测样本 rng: 随机数生成器（用于few-shot采样等） """ # Step 1: 构造prompt prompt = self._build_prompt(sample) # Step 2: 调用模型获取补全 result = self.completion_fn( prompt=prompt, temperature=0.0, # 评测时使用确定性生成 max_tokens=512, stop=["\n\n"] # 遇到双换行停止 ) model_output = result.get_completions()[0] # Step 3: 从模型输出中提取答案 extracted_answer = self._extract_answer(model_output) # Step 4: 判断正确性 expected = sample.get("ideal", sample.get("answer")) is_correct = self._judge_correctness( extracted_answer, expected ) # Step 5: 记录结果（包含详细元数据） evals.record.record_match( correct=is_correct, expected=expected, picked=extracted_answer, sampled=[model_output] # 模型原始输出 ) # 额外记录推理步骤质量（用于分析） if "reasoning_steps" in sample: evals.record.record_metrics( reasoning_length=len(model_output.split()), expected_steps=len(sample["reasoning_steps"]) ) def _build_prompt(self, sample: Dict) -> str: """构造评测prompt。 可以根据样本类型动态调整prompt格式。 """ task_type = sample.get("type", "general") if task_type == "math": return ( f"请逐步推理以下数学问题。\n" f"问题：{sample['question']}\n" f"请写出完整的推理步骤，最后给出答案。" ) elif task_type == "code": return ( f"编写Python函数解决以下问题：\n" f"{sample['question']}\n" f"```python\n" ) else: return f"问题：{sample['question']}\n答案：" def _extract_answer(self, text: str) -> str: """从模型输出中提取最终答案。 使用多层fallback策略，从最严格到最宽松。 """ import re # 策略1：匹配"答案是X"模式 m = re.search(r'(?:答案是|答案：|answer is)\s*(.+?)(?:[。\.]|$)', text) if m: return m.group(1).strip() # 策略2：匹配最后一行的数值或文本 lines = text.strip().split('\n') last_line = lines[-1] if lines else text # 策略3：返回整段文本（无匹配时） return last_line.strip() def _judge_correctness(self, predicted: str, expected: str) -> bool: """判断预测答案是否正确。 支持精确匹配和数值容差匹配。 """ # 精确匹配 if predicted.strip().lower() == expected.strip().lower(): return True # 数值容差匹配 try: pred_num = float(predicted.replace(',', '')) exp_num = float(expected.replace(',', '')) return abs(pred_num - exp_num) < 1e-4 except (ValueError, TypeError): pass # 包含匹配 if expected.strip().lower() in predicted.strip().lower(): return True return False def eval_few_shot( self, samples: Iterator[Dict], n_samples: int = None ) -> Dict: """执行少样本评测（可选实现）。 使用数据集中的训练样本作为few-shot示例。 """ # 此处为简化实现，实际需要合理的few-shot采样 return self.eval(samples, n_samples) def run(self, recorder: evals.record.Recorder) -> Dict: """执行完整评测流程。 Args: recorder: Evals的事件记录器 Returns: 聚合后的评测指标 """ samples = self.get_samples() if self.max_samples: from itertools import islice samples = islice(samples, self.max_samples) # 逐样本评测 for sample in samples: with recorder.as_default_recorder(sample): self.eval_sample(sample, rng=None) # 返回聚合结果 return recorder.get_metrics() def get_samples(self) -> Iterator[Dict]: """加载评测数据集。 支持JSONL格式，每行一个样本。 """ with open(self.samples_jsonl, 'r', encoding='utf-8') as f: for line in f: if line.strip(): yield json.loads(line)

三、结果收集与多层次分析

Evals的recorder机制自动收集每个样本的评测结果，包括模型输出、预期答案、正确性判断和自定义指标。评测完成后，oaieval生成结构化的JSON报告。

自定义分析的关键是利用recorder记录的多维信息进行切片分析——不是仅看一个总体的准确率数字。

import json import numpy as np from collections import defaultdict from typing import Dict, List def analyze_eval_results(result_file: str) -> Dict: """对Evals结果进行多层次切片分析。 分析维度： 1. 按数据子类别（如数学vs代码vs常识） 2. 按难度级别 3. 按推理步骤数 Args: result_file: Evals输出的JSON结果文件路径 Returns: 包含各维度聚合分析的字典 """ with open(result_file, 'r') as f: results = json.load(f) analysis = {} # 维度一：按类别分析 category_stats = defaultdict(lambda: {"correct": 0, "total": 0}) for sample in results["samples"]: category = sample.get("category", "general") category_stats[category]["total"] += 1 if sample.get("correct", False): category_stats[category]["correct"] += 1 analysis["by_category"] = { cat: { "accuracy": stats["correct"] / stats["total"], "count": stats["total"] } for cat, stats in category_stats.items() } # 维度二：按难度分析（如果有） if any("difficulty" in s for s in results["samples"]): diff_stats = defaultdict(lambda: {"correct": 0, "total": 0}) for sample in results["samples"]: diff = sample.get("difficulty", "unknown") diff_stats[diff]["total"] += 1 if sample.get("correct"): diff_stats[diff]["correct"] += 1 analysis["by_difficulty"] = { d: { "accuracy": s["correct"] / s["total"], "count": s["total"] } for d, s in diff_stats.items() } # 维度三：典型错误分析 errors = [s for s in results["samples"] if not s.get("correct")] analysis["error_summary"] = { "total_errors": len(errors), "error_rate": len(errors) / len(results["samples"]), "sample_errors": errors[:5] # 前5个错误案例 } return analysis

四、Evals与主流评测框架的对比

与lm-eval（见0716/4.md）相比，Evals的特点在于：(1) 更轻量级的任务定义——一个Python文件即可定义一个完整的评测任务；(2) 与OpenAI API的深度集成——completion_fn抽象层使得切换不同模型仅需更改一行配置；(3) 内置的结果记录和可视化支持。劣势在于任务的生态规模尚不如lm-eval丰富，且社区贡献的任务质量参差不齐。

两者的互补使用模式：用Evals快速原型化自定义评测任务（特别是需要复杂评判逻辑的任务），lm-eval用于标准benchmark的复现和对比。

五、总结

OpenAI Evals框架的核心价值在于将"评测任务"从静态数据集+独立脚本的模式升级为可版本化、可组合、可复现的代码模块。自定义任务的开发遵循"数据集定义→评测逻辑→指标聚合"的三步模式。实践中值得注意的两点：(1) 评测逻辑中的答案提取函数（_extract_answer）是决定评测质量的关键——粗糙的正则匹配会将大量"实际正确但格式不符"的输出误判为错误；(2) 利用recorder记录丰富的元数据（推理步骤数、token消耗、中间输出等），这些数据为后续的切片分析和模型诊断提供了远超"总体准确率"的信息量。