之前在做 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_tokens=256, temperature=0.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_url="https://openrouter.ai/api/v1", timeout=120): self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } def list_models(self): resp = requests.get( f"{self.base_url}/models", headers=self.headers, timeout=self.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", headers=self.headers, json=payload, timeout=self.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, rounds=1, concurrency=3): self.client = client self.tasks = tasks self.rounds = rounds self.concurrency = concurrency def run(self, model): results = [] with ThreadPoolExecutor(max_workers=self.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(path=None): if path: with open(path, "r", encoding="utf-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(description="OpenRouter LLM Benchmark Tool") parser.add_argument("--models", nargs="+", required=False, help="要评测的模型列表,例如 openai/gpt-4o-mini anthropic/claude-3.5-sonnet") parser.add_argument("--api-key", default=os.getenv("OPENROUTER_API_KEY"), help="OpenRouter API Key,也可以通过环境变量 OPENROUTER_API_KEY 设置") parser.add_argument("--rounds", type=int, default=1, help="每个任务重复执行次数") parser.add_argument("--concurrency", type=int, default=3, help="并发请求数") parser.add_argument("--timeout", type=int, default=120, help="请求超时时间(秒)") parser.add_argument("--list-models", action="store_true", help="列出当前可用的模型") parser.add_argument("--output", default="benchmark_report.json", help="报告JSON输出路径") parser.add_argument("--task-file", default=None, 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, timeout=args.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, rounds=args.rounds, concurrency=args.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", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) 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_tokens=256, temperature=0.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_url="https://openrouter.ai/api/v1", timeout=120): self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } def list_models(self): resp = requests.get( f"{self.base_url}/models", headers=self.headers, timeout=self.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", headers=self.headers, json=payload, timeout=self.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, rounds=1, concurrency=3): self.client = client self.tasks = tasks self.rounds = rounds self.concurrency = concurrency def run(self, model): results = [] with ThreadPoolExecutor(max_workers=self.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(path=None): if path: with open(path, "r", encoding="utf-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(description="OpenRouter LLM Benchmark Tool") parser.add_argument("--models", nargs="+", required=False, help="要评测的模型列表,例如 openai/gpt-4o-mini anthropic/claude-3.5-sonnet") parser.add_argument("--api-key", default=os.getenv("OPENROUTER_API_KEY"), help="OpenRouter API Key,也可以通过环境变量 OPENROUTER_API_KEY 设置") parser.add_argument("--rounds", type=int, default=1, help="每个任务重复执行次数") parser.add_argument("--concurrency", type=int, default=3, help="并发请求数") parser.add_argument("--timeout", type=int, default=120, help="请求超时时间(秒)") parser.add_argument("--list-models", action="store_true", help="列出当前可用的模型") parser.add_argument("--output", default="benchmark_report.json", help="报告JSON输出路径") parser.add_argument("--task-file", default=None, 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, timeout=args.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, rounds=args.rounds, concurrency=args.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%}