☰
Python实现大语言模型效果评测:从评测集到指标计算避坑指南
2026/10/2 17:39:55 网站建设 项目流程

简介:一套围绕大语言模型效果评测场景设计的Python源码,面向需要验证模型在主观题、客观题上表现的研究者、算法工程师及高校相关专业学生。项目采用CSV记录批量评测结果,以JSON管理模型配置,配合Python脚本实现评测逻辑,并附带GIF操作演示与说明文档,形成可复用的评估工作流。压缩包共142个文件,大小约26.62MB,其中102个CSV存放测试集指标数据,21个JSON定义模型参数,6个Python文件承担核心评测功能,结构清晰,便于按模块查阅和二次扩展。已有348人学习下载,读者可基于该项目快速搭建自定义评测方案,替换数据与配置后即可适配不同场景下的大语言模型对比分析任务。

1. 为什么大语言模型效果评测不是跑两个指标那么简单

接触过大语言模型落地的人都有这种体验:模型在公开榜上分数好看,一放到自己的业务数据上就原形毕露。效果评测的代码看着直接——给模型出题、对答案、算百分比——但真跑起来,评测集有没有混入脏数据、prompt模板的措辞偏向、并发请求把本地部署的模型服务打挂、生成长度截断把答案腰斩,每一个环节都能让分数悄悄失真。这篇文章把用Python实现的一套LLM效果评测代码拆开讲,从评测集加载、模型推理封装到指标计算和结果落盘,给出能直接改来用的源码结构。适合正在做模型选型、模型迭代回归,或者准备把大语言模型接进内部系统的工程师。

2. 评测框架怎么设计:从评测集到指标计算的完整链路

2.1 评测集的组织与加载:为什么用JSONL而不是数据库或Excel

先说结论。评测集我一般不用数据库,也不用Excel,统一用JSONL文件。格式是每行一条独立的JSON对象,字段固定为id、task_type、prompt、answer。选这个格式有三层考虑:一是追加方便,新样本直接往文件尾部写一行,不需要改表结构;二是Git友好,文件是纯文本,每一行都能参与diff,评测集调整了哪几条一目了然;三是不同任务类型可以在同一个文件里共存,用task_type字段区分就够,省得拆成一堆小文件。

import json from pathlib import Path class EvalDataset: def __init__(self, data_path: str): self.data_path = Path(data_path) self.samples = [] self._load() def _load(self): if not self.data_path.exists(): raise FileNotFoundError(f"评测集文件不存在: {self.data_path}") with open(self.data_path, "r", encoding="utf-8") as f: for idx, line in enumerate(f): line = line.strip() if not line: continue # 跳过空行,容忍文件尾部的多余换行 item = json.loads(line) # 必填字段校验:缺 id / task_type / prompt / answer 的样本直接丢弃 if not all(k in item for k in ("id", "task_type", "prompt", "answer")): print(f"[警告] 第{idx + 1}行缺少必填字段,已跳过") continue item["index"] = idx # 记录原始行号,方便回查 self.samples.append(item) def filter_by_task(self, task_type: str): """按任务类型过滤样本,返回一个新的列表""" return [s for s in self.samples if s["task_type"] == task_type] def __len__(self): return len(self.samples)

这里有两个设计点容易被新手忽略。第一是json.loads可能抛异常,常见的做法是不但要catch,还要打印行号,否则源文件里混进一行坏数据,整个评测直接中断,排查时还得自己数行。第二是“缺字段就跳过”这个策略要谨慎,如果评测集是别人给的,静默跳过会导致你最后统计的样本数和预期对不上,我习惯先跑一遍只加载不打分,把警告数量当作前置检查项。

参数方面,EvalDataset只接收一个data_path。样本总数就是文件行数,filter_by_task用于把同一份评测集拆成多个子任务分别评估。这个结构支撑了后面所有章节的评测流程,评测集本身不关心用什么模型、怎么调模型——数据和推理彻底解耦,这是评测代码设计里最容易踩的耦合问题。

2.2 模型推理封装:统一接口背后是兼容本地部署与云端API

评测代码里最容易被写死的部分是模型调用。很多人第一版直接把OpenAI的SDK调用写在评测脚本里,跑通之后想换个本地部署的模型对比,就得改一堆代码。正确的做法是包一个LLMClient,让评测主控只依赖chat这一个方法。

import requests class LLMClient: def __init__(self, base_url: str, api_key: str = "EMPTY", model_name: str = "default", timeout: int = 120): self.base_url = base_url.rstrip("/") self.api_key = api_key self.model_name = model_name self.timeout = timeout def chat(self, prompt: str, task_type: str = "MCQ", temperature: float = 0.0, max_tokens: int = 256): # 单选/判断/多选类任务推荐 temperature=0,保证输出可复现 # 生成类任务可以视情况放到 0.3~0.7,但评测分数会有波动 url = f"{self.base_url}/v1/chat/completions" payload = { "model": self.model_name, "messages": [{"role": "user", "content": prompt}], "temperature": temperature, "max_tokens": max_tokens } headers = {"Authorization": f"Bearer {self.api_key}"} try: resp = requests.post(url, json=payload, headers=headers, timeout=self.timeout) resp.raise_for_status() data = resp.json() except requests.exceptions.Timeout: raise TimeoutError(f"请求超时,prompt 前缀: {prompt[:30]}...") except requests.exceptions.HTTPError as e: raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:200]}") from e return data["choices"][0]["message"]["content"].strip()

为什么选OpenAI兼容协议而不是各家原生SDK?因为现在不管是本地部署的vLLM、Ollama,还是各类云端服务,普遍都提供了OpenAI兼容的HTTP接口。评测代码统一走这个协议,换模型就是换base_url和model_name,评测逻辑一行不用动。这也让配置化成为可能:一个JSON配置文件里写五组不同的base_url,就能跑同一份评测集做横向对比。

参数说明要单独说。temperature是评测里最容易被忽视的参数,选择题、判断题这类有唯一答案的任务必须设成0,否则同一个样本跑两遍结果可能不一致,分数没法复现。max_tokens的选择取决于任务类型:单选题给到64都够,开放生成类建议256或更高,但设太高又会拖慢整体评测速度,这个值需要和后面的并发设计一起考虑。timeout设120秒是给长文本生成的余量,如果评测集里都是短问答,可以降到30秒,让失败的样本更早进入重试流程。

2.3 指标计算:准确率之外还要看哪些数

指标计算是评测代码里最需要较真的环节。如果评测集里全是单选或多选题,只看一个准确率就够了;但只要混入简答、代码生成、摘要类任务,单纯字符串比对就会严重低估模型。我通常把指标拆成两层:第一层是全局的exact_match,适合快速看全局水位;第二层是按task_type分组统计,单独看每类任务的表现,这样才能定位模型到底在哪类任务上退化。

import re from collections import defaultdict def normalize_answer(text: str) -> str: # 归一化:去空白、转小写、去掉两端常见标点 text = text.strip().lower() text = re.sub(r"\s+", " ", text) text = text.strip(".,;:!?,。;:!?") return text def compute_metrics(results: list) -> dict: task_stats = defaultdict(lambda: {"total": 0, "hit": 0}) hit_count = 0 for r in results: task = r["task_type"] task_stats[task]["total"] += 1 pred = normalize_answer(r.get("response", "")) gold = normalize_answer(r["gold"]) # 只统计 status 为 ok 的样本,失败的样本计入总数但不计入命中 if r.get("status") == "ok" and pred == gold: hit_count += 1 task_stats[task]["hit"] += 1 total = len(results) metrics = { "total": total, "exact_match": hit_count / total if total else 0.0, "task_breakdown": {} } for task, stats in task_stats.items(): metrics["task_breakdown"][task] = { "total": stats["total"], "accuracy": stats["hit"] / stats["total"] if stats["total"] else 0.0 } return metrics

注意这里有一条细节:compute_metrics接收的是result列表,而每个result里同时携带了response、gold、status三个字段。这个数据结构在设计评测代码时就要定下来,后面所有落盘、汇总、画图都依赖它。normalize_answer做了最小化的归一化处理——只去空白、小写、去标点,不做同义词替换。原因是在评测场景里,过度归一化会掩盖模型的真实表现:一个把“返回”写成“归还”的模型,在字符串比对里算错,但人工看可能算对。要不要引入语义相似度指标,取决于你的业务容忍度,这是个取舍问题。要不要按任务类型给不同权重,也是评测框架里要提前想好的,总分一样的情况下,权重分配决定了你关注的是哪一类能力。

3. 用Python实现评测主控:并发推理、断点续跑与结果落盘

3.1 主控流程编排:评测任务的生命周期

评测主控的本质是把“数据集加载—模型推理—指标计算—落盘”串成一条流水线,但流水线的健壮性决定了评测能不能过夜跑。一个评测集几百上千条样本,模型推理又慢,跑完一轮动辄一两个小时,中途任何一条样本抛异常整批重来是不能接受的。所以主控脚本的第一要务是任务级的隔离:单条失败只标记那一条,不中断整体。

import json import time from concurrent.futures import ThreadPoolExecutor, as_completed def run_evaluation(dataset, llm_client, task_type_filter=None, max_workers=4, output_path="results.jsonl"): # 评测集过滤,不传就全量跑 tasks = dataset.samples if task_type_filter: tasks = [s for s in tasks if s["task_type"] in task_type_filter] print(f"本次评测样本数: {len(tasks)}, 并发数: {max_workers}") results = [] # ThreadPoolExecutor 适合 I/O 密集的推理请求;CPU 密集场景考虑 ProcessPool with ThreadPoolExecutor(max_workers=max_workers) as executor: future_map = { executor.submit(execute_one_sample, llm_client, sample): sample for sample in tasks } for future in as_completed(future_map): sample = future_map[future] try: result = future.result() results.append(result) print(f"[完成] {sample['id']}: {result['status']}") except Exception as e: results.append({ "id": sample["id"], "task_type": sample["task_type"], "prompt": sample["prompt"], "gold": sample["answer"], "response": "", "error": str(e), "status": "failed" }) print(f"[失败] {sample['id']}: {e}") # 统一落盘 with open(output_path, "w", encoding="utf-8") as f: for r in results: f.write(json.dumps(r, ensure_ascii=False) + "\n") print(f"评测完成,结果写入 {output_path}") return results

这里用ThreadPoolExecutor而不是并发数不设上限的裸循环,是有意的设计选择。模型服务并发能力有限,本地部署的模型即便用vLLM,并发过高也会出现排队甚至显存溢出;云端API并发过高直接触发限流。max_workers推荐从4开始调,观察模型服务的响应耗时和资源占用,逐步加到8、16。失败样本的result结构要和成功样本保持一致,这样后面的指标计算代码不用区分两种情况,失败样本的response为空字符串,天然算作答错。

3.2 单条样本的执行与重试:算力约束下先保吞吐还是先保重试

单条样本的执行函数是评测代码里最值得打磨的地方。核心逻辑只有两步:调用LLMClient.chat获取response,组装成统一的result结构。难的是重试策略——网络抖动、服务过载、限流都是常态,不重试,评测会一直报失败;无脑重试,失败的样本拖住线程池,整个任务时间成倍增长。

def execute_one_sample(llm_client, sample, retry_times=3): last_err = None for attempt in range(retry_times): try: response = llm_client.chat( prompt=sample["prompt"], task_type=sample["task_type"] ) return { "id": sample["id"], "task_type": sample["task_type"], "prompt": sample["prompt"], "gold": sample["answer"], "response": response, "status": "ok" } except Exception as e: last_err = e # 指数退避:第1次等2秒,第2次等4秒,给服务留恢复时间 time.sleep(2 * (attempt + 1)) raise last_err

关于重试次数,retry_times=3是折中方案。评测场景里如果是本地部署的模型,网络问题极少,重试主要对抗的是服务并发过载;云端API时重试大概率能救回因为限流失败的样本。这里有点玄学成分:有些请求超时是因为模型推理本身慢,重试只是再排一次队,未必更快。如果评测集很大,我倾向把retry降到2,把省下来的时间拿去买并发数,效果反而更稳。

execute_one_sample和run_evaluation之间有一个隐式约定:execute抛异常,run_evaluation捕获后标记failed;execute正常返回,result里带上status="ok"。这个约定让单条样本的失败不影响整体,同时保留了完整的失败现场——prompt和gold都在result里,后续排查bad case不需要再去翻原始评测集。

3.3 结果落盘与断点续跑:让评测过程可追溯

结果落盘是评测代码里看着不起眼、实际上最能省事的部分。每条样本的结果都该持久化,而不是只存一个汇总分数。除了汇总表,我还会保留单独的results.jsonl。这两个文件各有用途:jsonl是原始数据,以后想复查某条样本的完整prompt和response,或者换一个指标口径重新统计,都不需要重跑评测;markdown是给人看的,方便贴在文档或提交到代码评审里。

from pathlib import Path def save_results(results: list, output_path: str): with open(output_path, "w", encoding="utf-8") as f: for r in results: f.write(json.dumps(r, ensure_ascii=False) + "\n") def load_existing_ids(result_path: str) -> set: """读取已有结果文件,返回样本id集合,用于断点续跑""" if not Path(result_path).exists(): return set() ids = set() with open(result_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: ids.add(json.loads(line)["id"]) except (json.JSONDecodeError, KeyError): continue return ids def export_markdown_table(results_by_model: dict, output_path="report.md"): """把多个模型的评测结果拼成一张对比表""" # results_by_model 结构: {"model_a": metrics_dict, "model_b": metrics_dict} lines = ["| 模型 | 样本数 | Exact Match | 任务A准确率 | 任务B准确率 |", "|---|---|---|---|---|"] for model_name, metrics in results_by_model.items(): breakdown = metrics["task_breakdown"] def acc(name): return f"{breakdown[name]['accuracy']:.3f}" if name in breakdown else "-" lines.append( f"| {model_name} | {metrics['total']} | {metrics['exact_match']:.3f} " f"| {acc('task_a')} | {acc('task_b')} |" ) with open(output_path, "w", encoding="utf-8") as f: f.write("\n".join(lines)) print(f"对比报告已生成: {output_path}")

断点续跑的实现思路很简单:评测开始前先调load_existing_ids,把已完成的样本id过滤掉,只提交没跑过的样本。这个逻辑我一般放在main函数里,而不是run_evaluation内部,因为run_evaluation要尽量保持纯粹——只负责跑和存。过滤是调用方的责任,这样职责边界清楚。export_markdown_table里的任务列名是写死的,实际使用时可以改成从评测集的任务类型集合动态生成,或者手动传一个列顺序参数。

4. 大语言模型效果评测的5个避坑点:从数据泄漏到Prompt敏感

4.1 评测集被模型“背下来”了:分数虚高的数据泄漏

现象:换用新模型时分数提升明显,但同一批样本反复用几个月后,分数只升不降,仔细核对发现有些答案早就在模型训练语料里出现过。

原因:公开数据集做评测集,模型训练阶段可能已经见过原题。那些下载量排名靠前的公开benchmark,恰恰是数据泄漏的高发区。评测集固定不变,迭代时间长了,问题出在评测集被“污染”,评测代码本身没做错什么。

解决:评测集做去重只是第一步。更稳的做法是留一份“私有评测集”,永远不公开,只存在自己内网。构建方式是拿一个基座模型在内部业务数据上做采样改写,人工校验答案后再入库。分数虚高这个坑最难受的点是它不会报错,只有当你把线上bad case拉出来对比的时候才会发现。评测代码里可以加一个简单的防泄漏检查:每批新样本入库前,用当前评测模型跑一遍并记录分数,如果某条样本的分数异常高到离谱,单独打标复查。

4.2 Prompt模板措辞一变,分数跟着变

现象:同一个选择题换个问法,从“请选择正确答案”改成“以上哪个选项正确”,准确率掉了好几个点。

原因:大语言模型对prompt的敏感度经常超出预期。评测代码里如果prompt模板写得含糊,模型会输出多余解释,导致后面解析答案的逻辑匹配不上。模板的措辞就是对模型的隐性暗示,同一个模型在不同措辞下表现出的能力差异可能大于不同模型之间的差异。

解决:模板固定、解析逻辑固定,两者都作为评测代码的一部分纳入版本管理。另一个习惯是,每条样本的prompt必须显式带上任务类型标识,方便后面按类型细查。解析时宁可多写几种匹配模式,也别只用字符串包含。我见过的一个翻车案例是答案里带着“选项B:xxx”这种前缀,字符串匹配怎么都对不上,后来统一改用正则提取第一个选项字母才解决。Prompt模板的每次改动都要回归一遍评测,不能只改不动。

4.3 并发数调高后结果反而波动:重试救不了过载

现象:并发数从4调到16后,总耗时没降多少,失败样本数却多了,分数也出现了微小波动。

原因:本地部署的模型服务有并发上限,请求堆积后单条响应时间变长,超时触发重试,重试又加重了服务负载,形成恶性循环。这时候重试机制不但没兜底,反而成了帮凶。

解决:控制并发数,而不是依赖重试。评测启动前先做一次小规模的冒烟测试,用20条样本跑一遍,观察平均耗时和错误率,再决定max_workers。另一条经验是给并发池加一个“正在处理的请求数”上限,超了就不再提交新任务,让已有请求先消化掉。这个指标可以从LLMClient的响应耗时间接推断,如果平均耗时突然翻倍,说明已经过载,需要调低并发。

4.4 生成长度截断把答案截没了

现象:多选题的大题选项文本较长,模型的response里只输出了前半段,后半部分被截断,导致字符串比对失败。

原因:max_tokens设得太小。选择题看似只需要一个字母,但模型先输出解释后给答案,或者反过来先给答案后解释,都会占用token额度。评测代码里如果全局只配一个max_tokens值,遇到长文本样本就会出问题。

解决:不能只看评测集的“任务类型”配置max_tokens,还要看样本本身。一个务实的做法是,在评测代码里对单选题的response做后处理——只取第一个字母或第一个匹配选项,降低对生成长度的依赖。同时max_tokens按任务类型配置,而不是全局统一一个值,生成类任务给到512,选择题给到64。我自己踩过这个坑:一次评测里摘要任务的max_tokens给太小,结果一堆样本最后几个字被截断,分数比预期低了十几个点,一开始还以为是模型问题。

4.5 Benchmark分数高但业务场景用不起来

现象:公开benchmark上分数领先的模型,在业务数据评测集上一测,表现不如预期。

原因:公开数据集的任务分布和真实业务场景差异很大,模型可能在编码、推理类任务上强,但在特定领域的术语理解、格式约束上弱。评测代码能计算分数,但算不出“业务适配度”。

解决:回归问题本质,评测代码的最终目的是服务业务。这里要提大语言模型强化学习后的回归测试,一个经常出现的情况是,强化训练后的模型在策略评估指标上飙升,但在基础能力评测集上出现回退。评测代码的职责是把这个回退量化出来,让决策者看到分数之外的趋势,而不是只看单一benchmark。所以我维护评测集时有一个原则:公开评测集跑出来的分数只作为参考,真正做决策要看的是一份自建的“业务基线集”,哪怕只有一两百条样本。

5. 把评测代码接进模型迭代流程:回归测试与业务评测集

评测代码写完之后,真正让它产生复利的是把它接进模型迭代流程。常见做法是把run_evaluation包成一个命令行入口,参数只有评测集路径、模型配置和输出路径,然后让定时任务每天跑一遍全量评测集,第二天看对比表。这里要提醒的是,评测集不能太大,每天全量跑成本高,回归测试用几百条核心样本就够。

业务评测集的构建我一般这样来:先从线上日志里捞模型输出的bad case,按任务类型归好类,人工补上参考答案,存成之前定义的JSONL格式。这批数据不追求量,追求的是覆盖面——每个业务场景保底五条。量化的标准是:这五条如果模型答对了,我敢把它放到线上流量里试。

最后一个技巧是自动标注回退样本。每次评测结果都和上一轮做diff,把分数下降的样本找出来单独成表。这比看总分有用,总分可能只是波动,但单条样本的回退往往指向具体的问题——某个prompt模板的隐性改动、某个领域的数据权重变化。这个diff脚本大约三十行,跑起来不费劲,但几乎是所有评测代码里被问的最多的功能。

回归测试跑起来之后,评测代码才算是闭环。我自己的习惯是,每次模型更新前先跑旧版本基线,再跑新版本,两次结果diff的差异,比任何一个单独分数都值得看。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询