之前在做 LLM 模型选型对比时经常要在不同模型之间来回切换 API测速度、看输出质量、统计 token 消耗整个过程非常零散。后来我基于 OpenRouter 的 OpenAI 兼容接口封装了一个轻量级开源评测工具能够用一套代码批量对比任意模型实测下来对日常选型和版本迭代很有帮助。本文就完整拆解这个工具的设计思路、核心代码、运行方式和避坑经验适合正在做模型调研的开发者也适合需要持续关注模型效果的团队参考。1. 为什么需要自己做一个 LLM Benchmark 工具1.1 LLM 选型不是“看参数”就够的当前大语言模型生态中模型数量非常多。每个模型在参数量、上下文长度、推理速度、输出质量、价格成本、指令跟随能力等方面都有差异。只看官方文档和榜单数据很难直接判断某一个模型是否适合自己的业务场景。举一个很常见的场景同样是中文知识问答A 模型回答逻辑严谨但速度偏慢B 模型响应极快却偶发输出格式不稳定的情况。如果只靠人工一条条测试不仅费时而且无法量化对比。这时候就需要一个可重复执行的基准测试工具用固定提示词、统一调用方式、相同并发条件把不同模型的真实表现跑出来。LLM Benchmark 工具的核心价值不是替代权威学术评测体系而是提供一个“业务视角”的快速参考它告诉我们某个模型在当前网络环境、当前请求参数、当前提示词风格下表现到底如何。1.2 OpenRouter 在模型评测中的优势OpenRouter 是一个聚合了多个大模型 API 的平台。开发者只需要申请一个 API Key就能通过统一的 HTTP 接口调用多个第三方模型。对于模型对比场景来说这有几个明显好处统一调用协议不同模型之间的差异被收敛到同一个请求结构里无需为每个模型单独申请 API、单独适配 SDK可以快速在多个模型之间切换方便横向对比返回结果中包含 token 用量信息便于统计成本与效率。因此基于 OpenRouter 做轻量级评测工具可以把“调用差异”这一层完全屏蔽掉让评测逻辑专注于数据采集与结果对比。1.3 轻量级工具与重型评测框架的定位差异业界有很多成熟的评测框架例如 HELM、lm-evaluation-harness、OpenCompass 等。这些框架评测维度全面支持大量数据集和复杂指标但使用门槛也高需要安装较多依赖、需要处理数据集格式、需要理解评测协议有些还需要 GPU 环境来跑生成式评测。而本文实现的轻量级工具定位完全不同对比项重型评测框架轻量级 Benchmark 工具安装成本依赖较多仅依赖 requests运行环境可能需要 GPU普通服务器即可评测范围专业数据集、多维度自定义提示词、基础指标关注点学术对比、论文指标业务可用性、速度、Token 成本上手难度较高相对简单如果你的目标是快速验证“几个模型中哪个更适合我的业务”一套轻量级工具比重型框架更直接。2. 核心概念与评测维度2.1 Benchmark 评测的基本指标一个实用的 LLM 评测工具至少需要关注以下几类指标成功率成功率是衡量请求是否正常返回的重要指标。如果某个模型在多次请求中出现较高比例的报错、超时或空响应说明它在生产环境的稳定性存疑。成功率公式如下成功率 成功请求数 / 总请求数 × 100%延迟延迟可以分为首 Token 延迟和总请求延迟。轻量级工具通常统计“从发出请求到完整响应返回”的总耗时。延迟直接影响用户体验也是模型选型时的重要参考。Token 消耗Token 消耗包括输入 Token 数和输出 Token 数。输入 Token 与提示词长度有关输出 Token 与模型生成内容长度有关。统计 Token 数据可以帮助估算成本也能侧面反映模型是否容易出现冗长输出。输出质量严格意义上的输出质量需要人工评测或模型评测轻量级工具可以选择替代指标例如输出长度、是否为空、是否包含关键字段等。更精细的质量评估可以后续增加关键词匹配或规则校验。2.2 OpenRouter API 的工作原理OpenRouter 的接口设计整体与 OpenAI 兼容。核心请求路径为GET /api/v1/models获取可用模型列表POST /api/v1/chat/completions发起对话补全请求一个典型的对话补全请求体如下{ model: openai/gpt-4o-mini, messages: [ { role: user, content: 请用一句话解释什么是TCP三次握手。 } ], max_tokens: 256, temperature: 0.2 }响应体通常包含模型返回的文本内容以及 usage 信息{ choices: [ { message: { role: assistant, content: TCP三次握手是... } } ], usage: { prompt_tokens: 18, completion_tokens: 45, total_tokens: 63 } }这些字段就是我们评测工具需要收集的核心数据。基于这一认识我们可以在不依赖任何第三方 SDK 的情况下直接用 Python 的 requests 库完成客户端封装。2.3 本文评测工具的方法论为了让评测结果具备可比性工具采用固定提示词、固定请求参数、统一并发控制的方式。每个模型都会执行同一组任务并重复多轮然后汇总以下结果整体成功率平均延迟、中位延迟、95 分位延迟平均输入 Token、平均输出 Token平均输出长度每个任务维度下的分项表现。这套方法不追求学术严谨性但足够支撑日常模型选型。3. 环境准备与项目结构3.1 运行环境说明本项目是一个纯 Python 工具对硬件没有额外要求。示例环境如下Python 3.9 及以上requests 库操作系统macOS / Linux / Windows 均可依赖安装命令pip install requests版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。3.2 项目结构项目采用单文件设计方便快速使用和二次修改。openrouter-benchmark/ ├── openrouter_benchmark.py # 主程序包含全部逻辑 ├── tasks.json # 自定义评测任务可选 ├── benchmark_report.json # 评测结果输出 └── README.md # 使用说明单文件实现的优点是部署简单适合快速上手缺点是代码组织不够模块化。如果后续要扩展更多评测维度可以拆分成client.py、metrics.py、reporter.py等模块。3.3 环境变量与 API Key 安全评测工具通过 OpenRouter API 调用模型需要提前准备 API Key。建议通过环境变量注入避免把密钥硬编码在代码或命令行参数中。export OPENROUTER_API_KEY你的 API Key在 Windows PowerShell 中可用$env:OPENROUTER_API_KEY你的 API Key强调一下API Key 属于敏感凭证不要提交到 Git 仓库。如果代码托管在公开平台务必在.gitignore中忽略包含密钥的文件。4. 完整代码实现下面我们直接实现这个轻量级 Benchmark 工具。代码虽然不长但覆盖了 API 调用、并发控制、指标统计、报告生成等完整流程。4.1 评测任务实体 BenchmarkTask评测任务需要包含任务名称、提示词、max_tokens 和 temperature 四个字段。把这些信息封装成类便于批量定义和统一管理。# 核心评测任务定义 class BenchmarkTask: def __init__(self, name, prompt, max_tokens256, temperature0.2): self.name name self.prompt prompt self.max_tokens max_tokens self.temperature temperature def to_messages(self): return [ { role: user, content: self.prompt } ]其中temperature控制输出的随机性。数学计算类任务建议设为0.0写作创意类任务可以适当调高到0.7左右。4.2 API 客户端封装 OpenRouterClient客户端负责请求发送和响应解析。需要实现两个方法list_models()获取可用模型列表chat()发送对话请求返回结构化结果。# 核心OpenRouter API 客户端 class OpenRouterClient: def __init__(self, api_key, base_urlhttps://openrouter.ai/api/v1, timeout120): self.api_key api_key self.base_url base_url.rstrip(/) self.timeout timeout self.headers { Authorization: fBearer {api_key}, Content-Type: application/json, } def list_models(self): resp requests.get( f{self.base_url}/models, headersself.headers, timeoutself.timeout ) resp.raise_for_status() return resp.json().get(data, []) def chat(self, model, task): payload { model: model, messages: task.to_messages(), max_tokens: task.max_tokens, temperature: task.temperature, } start time.perf_counter() resp requests.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, timeoutself.timeout, ) elapsed time.perf_counter() - start if resp.status_code ! 200: return { success: False, status_code: resp.status_code, error: resp.text[:500] if resp.text else unknown error, latency: elapsed, prompt_tokens: 0, completion_tokens: 0, output_text: , } data resp.json() try: output_text data[choices][0][message][content] or except (KeyError, IndexError): output_text usage data.get(usage, {}) return { success: True, status_code: resp.status_code, latency: elapsed, prompt_tokens: usage.get(prompt_tokens, 0), completion_tokens: usage.get(completion_tokens, 0), output_text: output_text, }这里有几个细节值得说明time.perf_counter()比time.time()更适合测量短时间间隔精度更高非 200 响应也要记录延迟便于分析超时或限流造成的耗时响应体的解析使用try-except避免因个别模型返回结构异常导致整个评测中断。4.3 评测执行器 BenchmarkRunner评测执行器负责并发发起请求、收集结果、计算统计指标。并发控制使用 Python 标准库的ThreadPoolExecutor不引入额外队列依赖。# 核心评测执行与指标统计 class BenchmarkRunner: def __init__(self, client, tasks, rounds1, concurrency3): self.client client self.tasks tasks self.rounds rounds self.concurrency concurrency def run(self, model): results [] with ThreadPoolExecutor(max_workersself.concurrency) as executor: future_map {} for task in self.tasks: for r in range(self.rounds): future executor.submit(self.client.chat, model, task) future_map[future] (task.name, r 1) for future in as_completed(future_map): task_name, round_no future_map[future] try: result future.result() except Exception as exc: result { success: False, status_code: None, error: str(exc), latency: None, prompt_tokens: 0, completion_tokens: 0, output_text: , } result[task] task_name result[round] round_no results.append(result) return results def summarize(self, results): success [r for r in results if r.get(success)] total len(results) latency_values [r[latency] for r in success if r.get(latency) is not None] prompt_values [r[prompt_tokens] for r in success] completion_values [r[completion_tokens] for r in success] out_lengths [len(r[output_text]) for r in success] p95_latency None if latency_values: sorted_latency sorted(latency_values) idx max(0, min(len(sorted_latency) - 1, int(len(sorted_latency) * 0.95))) p95_latency sorted_latency[idx] return { total_requests: total, success_requests: len(success), failed_requests: total - len(success), success_rate: len(success) / total if total else 0, avg_latency: statistics.mean(latency_values) if latency_values else None, median_latency: statistics.median(latency_values) if latency_values else None, p95_latency: p95_latency, avg_prompt_tokens: statistics.mean(prompt_values) if prompt_values else None, avg_completion_tokens: statistics.mean(completion_values) if completion_values else None, avg_output_length: statistics.mean(out_lengths) if out_lengths else None, } def summarize_by_task(self, results): by_task {} for r in results: task_name r.get(task, 未知任务) by_task.setdefault(task_name, []).append(r) task_summary {} for task_name, task_results in by_task.items(): task_summary[task_name] self.summarize(task_results) return task_summary并发数concurrency不宜设置过高避免触发平台限流策略。对于绝大多数场景3 到 5 并发已经足够。4.4 默认评测任务工具内置了一组覆盖多类能力的评测任务包括常识问答、数学计算、代码生成、逻辑推理和中文写作。你可以直接使用也可以通过 JSON 文件自定义。# 默认评测任务 BENCHMARK_TASKS [ { name: 常识问答, prompt: 请用一句话解释什么是TCP三次握手。, max_tokens: 256, temperature: 0.2, }, { name: 数学计算, prompt: 计算12345 * 6789 请直接给出数字结果。, max_tokens: 128, temperature: 0.0, }, { name: 代码生成, prompt: 用Python写一个函数判断一个字符串是否是回文。, max_tokens: 512, temperature: 0.2, }, { name: 逻辑推理, prompt: 有三个盒子一个只装苹果一个只装橘子一个装苹果和橘子。盒子外面贴的标签分别是“苹果”、“橘子”、“苹果和橘子”但每个标签都贴错了。你只能从一个盒子里取一个水果不看盒子里面。请问你该从哪个盒子取水果才能判断出三个盒子各装了什么请给出推理过程。, max_tokens: 512, temperature: 0.2, }, { name: 中文写作, prompt: 写一段80字左右的产品介绍介绍一个开源LLM评测工具。, max_tokens: 512, temperature: 0.7, }, ]自定义任务文件格式如果你需要评测自己的业务提示词可以创建tasks.json文件[ { name: 业务问答, prompt: 请根据以下用户问题给出回答如何重置密码, max_tokens: 512, temperature: 0.2 }, { name: 内容摘要, prompt: 请将下面这段文本压缩成一句话摘要我们的平台支持多种模型接入并提供统一的API接口。, max_tokens: 256, temperature: 0.3 } ]使用--task-file tasks.json参数即可加载。4.5 报告生成评测结果需要以可读形式输出。工具支持两种输出Markdown 表格直接打印到终端方便快速查看JSON 文件保存完整结果便于后续处理或用于 CI。# 核心Markdown 报告渲染 def render_markdown_report(models, overall_results, task_results): lines [] lines.append(# LLM Benchmark 评测报告) lines.append() lines.append(## 总体指标) lines.append() metrics [ (成功率, success_rate, {:.1%}), (成功请求数, success_requests, {}), (失败请求数, failed_requests, {}), (平均延迟(s), avg_latency, {:.2f}), (中位延迟(s), median_latency, {:.2f}), (P95延迟(s), p95_latency, {:.2f}), (平均输入Token, avg_prompt_tokens, {:.1f}), (平均输出Token, avg_completion_tokens, {:.1f}), (平均输出长度, avg_output_length, {:.1f}), ] header | 指标 | | .join(models) | separator | --- | | .join([---] * len(models)) | lines.append(header) lines.append(separator) for label, key, fmt in metrics: values [] for m in models: v overall_results[m].get(key) if v is None: values.append(-) else: values.append(fmt.format(v)) lines.append(f| {label} | | .join(values) |) first_model models[0] task_names task_results[first_model].keys() for task_name in task_names: lines.append() lines.append(f## 任务{task_name}) lines.append() lines.append(header.replace(指标, 指标)) lines.append(separator) for label, key, fmt in metrics: values [] for m in models: v task_results[m].get(task_name, {}).get(key) if v is None: values.append(-) else: values.append(fmt.format(v)) lines.append(f| {label} | | .join(values) |) return \n.join(lines)4.6 主程序入口主程序负责解析命令行参数、加载任务、执行评测并输出报告。# 核心任务加载 def load_tasks(pathNone): if path: with open(path, r, encodingutf-8) as f: data json.load(f) return [BenchmarkTask(**item) for item in data] return [BenchmarkTask(**item) for item in BENCHMARK_TASKS] # 核心命令行入口 def main(): parser argparse.ArgumentParser(descriptionOpenRouter LLM Benchmark Tool) parser.add_argument(--models, nargs, requiredFalse, help要评测的模型列表例如 openai/gpt-4o-mini anthropic/claude-3.5-sonnet) parser.add_argument(--api-key, defaultos.getenv(OPENROUTER_API_KEY), helpOpenRouter API Key也可以通过环境变量 OPENROUTER_API_KEY 设置) parser.add_argument(--rounds, typeint, default1, help每个任务重复执行次数) parser.add_argument(--concurrency, typeint, default3, help并发请求数) parser.add_argument(--timeout, typeint, default120, help请求超时时间秒) parser.add_argument(--list-models, actionstore_true, help列出当前可用的模型) parser.add_argument(--output, defaultbenchmark_report.json, help报告JSON输出路径) parser.add_argument(--task-file, defaultNone, help自定义任务JSON文件) args parser.parse_args() if not args.api_key: print(错误请设置 OPENROUTER_API_KEY 环境变量或通过 --api-key 参数传入 API Key。) sys.exit(1) client OpenRouterClient(args.api_key, timeoutargs.timeout) if args.list_models: models client.list_models() for m in models: print(m.get(id)) return if not args.models: print(错误请通过 --models 指定至少一个模型或先使用 --list-models 查看可用模型。) sys.exit(1) tasks load_tasks(args.task_file) runner BenchmarkRunner(client, tasks, roundsargs.rounds, concurrencyargs.concurrency) overall_results {} task_results {} for model in args.models: print(f开始评测: {model}) results runner.run(model) overall_results[model] runner.summarize(results) task_results[model] runner.summarize_by_task(results) print(f完成评测: {model}成功率 {overall_results[model][success_rate]:.1%}) report { models: args.models, overall: overall_results, by_task: task_results, } with open(args.output, w, encodingutf-8) as f: json.dump(report, f, ensure_asciiFalse, indent2) print(render_markdown_report(args.models, overall_results, task_results)) if __name__ __main__: main()这里argparse是 Python 标准库无需额外安装。--list-models参数可以帮助用户快速查看当前 OpenRouter 平台上实际可用的模型 ID避免写错模型名称。4.7 完整代码汇总为了方便直接复制使用这里是openrouter_benchmark.py的完整代码#!/usr/bin/env python3 openrouter_benchmark.py 轻量级开源 LLM Benchmark 工具在 OpenRouter 上对比任意模型。 import argparse import json import os import statistics import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed import requests class BenchmarkTask: def __init__(self, name, prompt, max_tokens256, temperature0.2): self.name name self.prompt prompt self.max_tokens max_tokens self.temperature temperature def to_messages(self): return [ { role: user, content: self.prompt } ] class OpenRouterClient: def __init__(self, api_key, base_urlhttps://openrouter.ai/api/v1, timeout120): self.api_key api_key self.base_url base_url.rstrip(/) self.timeout timeout self.headers { Authorization: fBearer {api_key}, Content-Type: application/json, } def list_models(self): resp requests.get( f{self.base_url}/models, headersself.headers, timeoutself.timeout ) resp.raise_for_status() return resp.json().get(data, []) def chat(self, model, task): payload { model: model, messages: task.to_messages(), max_tokens: task.max_tokens, temperature: task.temperature, } start time.perf_counter() resp requests.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, timeoutself.timeout, ) elapsed time.perf_counter() - start if resp.status_code ! 200: return { success: False, status_code: resp.status_code, error: resp.text[:500] if resp.text else unknown error, latency: elapsed, prompt_tokens: 0, completion_tokens: 0, output_text: , } data resp.json() try: output_text data[choices][0][message][content] or except (KeyError, IndexError): output_text usage data.get(usage, {}) return { success: True, status_code: resp.status_code, latency: elapsed, prompt_tokens: usage.get(prompt_tokens, 0), completion_tokens: usage.get(completion_tokens, 0), output_text: output_text, } class BenchmarkRunner: def __init__(self, client, tasks, rounds1, concurrency3): self.client client self.tasks tasks self.rounds rounds self.concurrency concurrency def run(self, model): results [] with ThreadPoolExecutor(max_workersself.concurrency) as executor: future_map {} for task in self.tasks: for r in range(self.rounds): future executor.submit(self.client.chat, model, task) future_map[future] (task.name, r 1) for future in as_completed(future_map): task_name, round_no future_map[future] try: result future.result() except Exception as exc: result { success: False, status_code: None, error: str(exc), latency: None, prompt_tokens: 0, completion_tokens: 0, output_text: , } result[task] task_name result[round] round_no results.append(result) return results def summarize(self, results): success [r for r in results if r.get(success)] total len(results) latency_values [r[latency] for r in success if r.get(latency) is not None] prompt_values [r[prompt_tokens] for r in success] completion_values [r[completion_tokens] for r in success] out_lengths [len(r[output_text]) for r in success] p95_latency None if latency_values: sorted_latency sorted(latency_values) idx max(0, min(len(sorted_latency) - 1, int(len(sorted_latency) * 0.95))) p95_latency sorted_latency[idx] return { success_rate: len(success) / total if total else 0, total_requests: total, success_requests: len(success), failed_requests: total - len(success), avg_latency: statistics.mean(latency_values) if latency_values else None, median_latency: statistics.median(latency_values) if latency_values else None, p95_latency: p95_latency, avg_prompt_tokens: statistics.mean(prompt_values) if prompt_values else None, avg_completion_tokens: statistics.mean(completion_values) if completion_values else None, avg_output_length: statistics.mean(out_lengths) if out_lengths else None, } def summarize_by_task(self, results): by_task {} for r in results: task_name r.get(task, 未知任务) by_task.setdefault(task_name, []).append(r) task_summary {} for task_name, task_results in by_task.items(): task_summary[task_name] self.summarize(task_results) return task_summary BENCHMARK_TASKS [ { name: 常识问答, prompt: 请用一句话解释什么是TCP三次握手。, max_tokens: 256, temperature: 0.2, }, { name: 数学计算, prompt: 计算12345 * 6789 请直接给出数字结果。, max_tokens: 128, temperature: 0.0, }, { name: 代码生成, prompt: 用Python写一个函数判断一个字符串是否是回文。, max_tokens: 512, temperature: 0.2, }, { name: 逻辑推理, prompt: 有三个盒子一个只装苹果一个只装橘子一个装苹果和橘子。盒子外面贴的标签分别是“苹果”、“橘子”、“苹果和橘子”但每个标签都贴错了。你只能从一个盒子里取一个水果不看盒子里面。请问你该从哪个盒子取水果才能判断出三个盒子各装了什么请给出推理过程。, max_tokens: 512, temperature: 0.2, }, { name: 中文写作, prompt: 写一段80字左右的产品介绍介绍一个开源LLM评测工具。, max_tokens: 512, temperature: 0.7, }, ] def load_tasks(pathNone): if path: with open(path, r, encodingutf-8) as f: data json.load(f) return [BenchmarkTask(**item) for item in data] return [BenchmarkTask(**item) for item in BENCHMARK_TASKS] def render_markdown_report(models, overall_results, task_results): lines [] lines.append(# LLM Benchmark 评测报告) lines.append() lines.append(## 总体指标) lines.append() metrics [ (成功率, success_rate, {:.1%}), (成功请求数, success_requests, {}), (失败请求数, failed_requests, {}), (平均延迟(s), avg_latency, {:.2f}), (中位延迟(s), median_latency, {:.2f}), (P95延迟(s), p95_latency, {:.2f}), (平均输入Token, avg_prompt_tokens, {:.1f}), (平均输出Token, avg_completion_tokens, {:.1f}), (平均输出长度, avg_output_length, {:.1f}), ] header | 指标 | | .join(models) | separator | --- | | .join([---] * len(models)) | lines.append(header) lines.append(separator) for label, key, fmt in metrics: values [] for m in models: v overall_results[m].get(key) if v is None: values.append(-) else: values.append(fmt.format(v)) lines.append(f| {label} | | .join(values) |) first_model models[0] task_names task_results[first_model].keys() for task_name in task_names: lines.append() lines.append(f## 任务{task_name}) lines.append() lines.append(header) lines.append(separator) for label, key, fmt in metrics: values [] for m in models: v task_results[m].get(task_name, {}).get(key) if v is None: values.append(-) else: values.append(fmt.format(v)) lines.append(f| {label} | | .join(values) |) return \n.join(lines) def main(): parser argparse.ArgumentParser(descriptionOpenRouter LLM Benchmark Tool) parser.add_argument(--models, nargs, requiredFalse, help要评测的模型列表例如 openai/gpt-4o-mini anthropic/claude-3.5-sonnet) parser.add_argument(--api-key, defaultos.getenv(OPENROUTER_API_KEY), helpOpenRouter API Key也可以通过环境变量 OPENROUTER_API_KEY 设置) parser.add_argument(--rounds, typeint, default1, help每个任务重复执行次数) parser.add_argument(--concurrency, typeint, default3, help并发请求数) parser.add_argument(--timeout, typeint, default120, help请求超时时间秒) parser.add_argument(--list-models, actionstore_true, help列出当前可用的模型) parser.add_argument(--output, defaultbenchmark_report.json, help报告JSON输出路径) parser.add_argument(--task-file, defaultNone, help自定义任务JSON文件) args parser.parse_args() if not args.api_key: print(错误请设置 OPENROUTER_API_KEY 环境变量或通过 --api-key 参数传入 API Key。) sys.exit(1) client OpenRouterClient(args.api_key, timeoutargs.timeout) if args.list_models: models client.list_models() for m in models: print(m.get(id)) return if not args.models: print(错误请通过 --models 指定至少一个模型或先使用 --list-models 查看可用模型。) sys.exit(1) tasks load_tasks(args.task_file) runner BenchmarkRunner(client, tasks, roundsargs.rounds, concurrencyargs.concurrency) overall_results {} task_results {} for model in args.models: print(f开始评测: {model}) results runner.run(model) overall_results[model] runner.summarize(results) task_results[model] runner.summarize_by_task(results) print(f完成评测: {model}成功率 {overall_results[model][success_rate]:.1%}