有段时间没聊评测工具链了。这次 HN 上出现了一个很有意思的讨论标题“Ask HN: Anyone interested in building a harness-only benchmark?” 很多人第一反应是benchmark 不是用来测模型的吗harness 不就是个跑分脚本吗实际上这个问题背后是一个真实痛点当你的任务不只是在单机跑一条 prompt而是要批量调度、并发排队、断点续跑、导出结构化结果时真正决定效率的不是模型本身而是那一层负责调度的 harness。所谓 harness-only benchmark直译过来就是“只针对 harness 的基准测试”。它不关心模型答对多少题也不比较模型 A 和模型 B 的能力分数而是把模型服务当作一个黑盒只测量调度层的并发能力、超时控制、失败重试、批量任务处理、日志与结果记录。这个概念和现在很多工具链里被频繁使用的 “DeepSeek harness”“Codex harness” 是同一个场景大家已经把模型选型完成了剩下的问题其实是工程化问题——用什么框架来跑批量评测、用什么调度器来接 API、怎么在长任务中断后不丢结果。这篇文章会沿着这个思路展开先拆解 harness-only benchmark 到底测什么再给出一套可以在本地直接跑通的最小参考实现包括 mock 模型服务、并发压测脚本、批量任务文件、断点续跑和结果指标输出。最后聊一下资源占用、常见问题和合规边界。如果你正在选评测框架、做批处理任务或者打算在自己的项目里加一层可观测的调用调度层这篇文章可以直接收藏。1. Harness-only Benchmark 是什么1.1 不是测模型能力是测调度系统传统 benchmark 的思路很直接准备好一组题目让模型回答算准确率。MMLU、GSM8K、BIG-Bench 都是这个套路。测试对象是模型harness 只是“拿着卷子去考模型”的监考老师。只要监考老师不出错分数高低基本只反映模型水平。但 harness-only benchmark 把角色反过来了模型变成“黑盒考生”我们要考的是监考老师本身。具体的测试内容大致包括这几个方面并发调度同时提交 10 个、50 个、100 个请求时harness 能不能稳定把任务分发出去而不是把进程搞崩。超时控制下游接口迟迟不返回时harness 会不会一直卡住还是能按配置超时并记录失败。失败重试请求返回 500、429、连接断开时harness 是否具备重试策略。批量任务数千条 prompt 能否排队执行而不是一次性把所有请求全部塞进内存。断点续跑任务跑到一半进程退出再次启动时能否跳过已完成样本不重复计费、不重复跑。结果记录每一条请求的输入、输出、耗时、状态码、错误信息是否完整落盘。接口兼容性换一个模型服务地址、换一种请求协议harness 需要改多少代码。资源占用调度器本身吃多少 CPU、内存有没有内存泄漏。从工程角度看这几点比模型分数更影响“任务能不能跑完”。因为单条 prompt 调用几乎不会出问题真正出问题的是批量、并发、长时间运行这些场景。1.2 为什么现在这个话题值得关注从搜索热词来看harness、benchmark 相关的讨论热度不低尤其是“DeepSeek harness”这类关键词频繁出现。这类关键词背后对应的实际需求通常是在某个模型能力确定的前提下用户希望用一套稳定的工具链去完成批量推理、结果整理、日志记录和出错恢复。换句话说模型能力已经不太是瓶颈瓶颈变成了“怎么稳定地把几万条任务跑完”。这正好是 harness-only benchmark 的用武之地。它回答的问题是我手里的这套调度脚本、评测框架或 API 封装层在高并发、长任务、异常情况下到底靠不靠谱。2. 核心能力速览能力项说明项目定位面向 harness 调度层的工程性能基准测试与传统 benchmark 的区别不测模型准确率测调度系统的吞吐、稳定性和可观测性主要测试对象并发调度、超时控制、失败重试、批量任务、断点续跑、日志记录推荐环境本地 Python 3.9任意模型服务包括本地部署或自有 API是否支持 CPU支持harness 本身主要是网络调度和文件处理逻辑是否支持 GPU可选取决于被测模型服务harness 本身不需要 GPU启动方式命令行启动 可选的 mock 模型服务是否支持 API通过 HTTP 调用下游模型服务可对接 OpenAI 兼容接口是否支持批量任务支持通过 JSONL 任务文件批量加载任务是否支持断点续跑支持通过记录 CSV 结果跳过已完成任务适合场景评测框架选型、批量推理流程验证、接口压测、CI 稳定性测试风险提示仅限本地测试环境和自有接口禁止对公共 API 做无限制压测3. 适用场景与使用边界3.1 适合谁用第一类用户是做大模型评测的同学。你需要对比不同的评测框架但每次都因为框架本身崩溃、超时、丢结果而重跑这时候先对 harness 做一轮工程基准测试比直接用真实模型做评测更省时间。第二类用户是做批量推理的工程师。你的场景可能是给一批历史对话生成摘要、给一堆报告做分类、或者跑大量 prompt 做数据增强。这些任务对模型能力要求不高但对“能不能稳定跑完几千条任务”要求很高。harness-only benchmark 的核心指标就是你真正要关心的性能指标。第三类用户是做 Agent 工具链集成的人。现在不少工具链会在模型外面包装一层“harness”负责工具调用、权限控制、上下文管理。这层代码越复杂越需要单独验证它的稳定性。把模型换成 mock 服务只测 harness 的调度逻辑能快速暴露并发问题和状态管理 bug。3.2 不适合什么场景如果你想知道“模型 A 还是模型 B 效果好”那不应该做 harness-only benchmark而应该做正常的模型能力评测。两类 benchmark 解决的问题完全不同混淆之后很容易得出没有意义的结论。另外它也解决不了 harness 内部逻辑正确性的问题。比如你的 harness 是否真的把提示词正确组装、是否在 Agent 多轮对话中丢失了上下文这些需要依靠单元测试和功能测试来保障而不是压测。3.3 合规边界这一点很重要。harness-only benchmark 的测试方式通常是高频并发请求如果你把这个脚本直接打向第三方公共 API很容易触发限流、封禁甚至给对方服务带来压力。所有压测动作都应该限制在本地部署的模型服务、自有的 API 网关或已经明确授权的测试环境里。同时如果被测任务里包含用户数据、人脸信息、声音素材或版权文本必须先做脱敏和授权确认。批量任务往往会放大风险一条数据有问题可能只是小问题一万条数据里有问题的记录就会变成事故。4. 评估维度设计要做一次有效的 harness-only benchmark不是随便并发打几个请求就算数。建议按下面的维度来设计测试用例。测试维度测试方式关键指标基础连通性单请求调用成功率、平均延迟并发能力固定 worker 数压测QPS、P95/P99 延迟、成功率超时控制设置较短 timeout 触发超时超时任务是否被正确记录失败重试让 mock 服务按比例返回 500/429重试次数、最终成功率批量任务大量 JSONL 任务文件吞吐、内存占用、是否有任务丢失断点续跑中断后重启测试已完成任务是否被跳过长任务稳定性长时间持续压测内存增长、句柄泄漏、崩溃率接口兼容性切换不同 API 地址和鉴权方式代码改动量、成功率这里有一个容易被忽略的点并发吞吐不是越高越好。很多框架能顶着高并发把 QPS 刷上去但 P99 延迟可能已经高到不可接受或者有一小部分请求直接超时失败。对于批量任务来说更稳定的策略通常是控制并发、保证成功率而不是盲目追求 QPS。5. 本地环境准备在开始写代码之前先确认下面几项操作系统Windows、Linux、macOS 都可以建议在有稳定网络环境的 Linux 服务器上跑。Python 版本3.9 或更高版本。Python 依赖requests用于调用模型服务flask用于启动 mock 服务。磁盘空间这个参考实现本身只需要几 MB但如果你要跑大量任务的日志和结果需要预留足够的磁盘。端口占用参考实现默认使用8000端口启动 mock 服务如果被占用可以换成别的端口。安装依赖pip install requests flask到这里就可以开始搭最小实现。6. 最小参考实现这一节会提供两个文件一个是 mock 模型服务用来模拟带延迟和偶尔失败的下游接口另一个是 harness 压测脚本负责并发调度、结果记录和指标计算。6.1 Mock 模型服务为什么需要 mock 服务因为很多人在本地并没有一个真正可用的模型服务。如果把压测脚本直接对接到某个在线 API既不安全也无法控制延迟和失败率。mock 服务可以灵活模拟真实场景等你完全跑通之后再替换成自己的模型地址。# mock_model_api.py import os import time import random from flask import Flask, request, jsonify app Flask(__name__) FAIL_RATE float(os.getenv(FAIL_RATE, 0)) MIN_DELAY float(os.getenv(MIN_DELAY, 0.05)) MAX_DELAY float(os.getenv(MAX_DELAY, 0.30)) app.route(/health, methods[GET]) def health(): return jsonify({status: ok}) app.route(/v1/completions, methods[POST]) def completions(): # 按比例返回 500模拟下游不稳定 if random.random() FAIL_RATE: return jsonify({error: mock failure}), 500 # 模拟模型推理耗时 time.sleep(random.uniform(MIN_DELAY, MAX_DELAY)) data request.get_json() prompt (data or {}).get(prompt, ) return jsonify({ id: mock-%d % random.randint(1000, 9999), choices: [{text: ok: %s % str(prompt)[:20], index: 0}] }) if __name__ __main__: app.run(host127.0.0.1, port8000)启动方式python mock_model_api.py默认情况下所有请求都会在 0.05 到 0.30 秒之间随机延迟后返回成功。如果你需要模拟下游不稳定的场景可以这样启动FAIL_RATE0.2 python mock_model_api.py这样 20% 的请求会返回 500用来测试 harness 的失败处理和重试逻辑。6.2 Harness 压测脚本下面这个脚本是 harness-only benchmark 的最小参考实现。它支持并发控制、超时设置、CSV 结果记录和断点续跑。# harness_bench.py import argparse import csv import json import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from statistics import mean, median import requests def load_tasks(path, max_tasks): tasks [] with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue tasks.append(json.loads(line)) if max_tasks and len(tasks) max_tasks: break return tasks def load_done(record_path): done set() if not Path(record_path).exists(): return done with open(record_path, r, encodingutf-8) as f: for row in csv.DictReader(f): if row.get(status) ok: done.add(row[task_id]) return done def call_api(url, payload, timeout): start time.perf_counter() try: resp requests.post(url, jsonpayload, timeouttimeout) latency_ms (time.perf_counter() - start) * 1000 return { ok: resp.ok, http_status: resp.status_code, latency_ms: latency_ms, error: if resp.ok else resp.text[:200], } except Exception as exc: latency_ms (time.perf_counter() - start) * 1000 return { ok: False, http_status: 0, latency_ms: latency_ms, error: str(exc), } def calc_metrics(results): total len(results) if total 0: return None ok_items [r for r in results if r[ok]] fail_items [r for r in results if not r[ok]] latencies sorted(r[latency_ms] for r in results) def percentile(p): idx max(0, min(total - 1, int(total * p) - 1)) return round(latencies[idx], 2) elapsed_sum sum(r[latency_ms] for r in results) / 1000.0 return { total: total, success: len(ok_items), fail: len(fail_items), success_rate: round(len(ok_items) / total * 100, 2), qps: round(total / elapsed_sum, 2), avg_ms: round(mean(latencies), 2), p50_ms: round(median(latencies), 2), p95_ms: percentile(0.95), p99_ms: percentile(0.99), } def run(args): tasks load_tasks(args.tasks, args.max_tasks) done load_done(args.record) if args.record else set() if args.overwrite: pending tasks else: pending [t for t in tasks if t.get(id) not in done] if not pending: print(no pending tasks, all done) return if args.record: record_path Path(args.record) need_header not record_path.exists() or record_path.stat().st_size 0 record_fp open(record_path, a, encodingutf-8, newline) record_writer csv.writer(record_fp) if need_header: record_writer.writerow([task_id, status, http_status, latency_ms, error]) else: record_fp None record_writer None results [] start time.perf_counter() with ThreadPoolExecutor(max_workersargs.workers) as executor: future_map { executor.submit(call_api, args.url, task[payload], args.timeout): task for task in pending } for future in as_completed(future_map): task future_map[future] result future.result() results.append(result) if record_writer: record_writer.writerow([ task[id], ok if result[ok] else fail, result[http_status], round(result[latency_ms], 2), result[error], ]) if record_fp: record_fp.close() elapsed time.perf_counter() - start metrics calc_metrics(results) if metrics: print(json.dumps(metrics, ensure_asciiFalse, indent2)) print(wall_time_seconds:, round(elapsed, 2)) def main(): parser argparse.ArgumentParser(descriptionharness-only benchmark minimal implementation) parser.add_argument(--url, requiredTrue, helpmodel API URL) parser.add_argument(--tasks, requiredTrue, helpJSONL tasks file path) parser.add_argument(--workers, typeint, default4, helpconcurrent workers) parser.add_argument(--timeout, typeint, default30, helprequest timeout in seconds) parser.add_argument(--max-tasks, typeint, default0, helpmax tasks to load) parser.add_argument(--record, defaultresults.csv, helpoutput CSV path) parser.add_argument(--overwrite, actionstore_true, helpignore previous records and rerun all) args parser.parse_args() run(args) if __name__ __main__: main()6.3 任务文件与运行命令任务文件使用 JSONL 格式每一行是一个 JSON 对象包含id和payload两个字段。id用于断点续跑去重payload是实际要发送给模型服务的请求体。{id: task-001, payload: {prompt: hello, max_tokens: 32}} {id: task-002, payload: {prompt: hello again, max_tokens: 64}} {id: task-003, payload: {prompt: tell me a story, max_tokens: 128}} {id: task-004, payload: {prompt: write a short email, max_tokens: 256}} {id: task-005, payload: {prompt: summarize the article, max_tokens: 256}}保存为tasks.jsonl然后启动 mock 服务python mock_model_api.py新开一个终端运行 harnesspython harness_bench.py \ --url http://127.0.0.1:8000/v1/completions \ --tasks tasks.jsonl \ --workers 4 \ --timeout 30 \ --record results.csv首次运行会输出类似这样的统计结果{ total: 5, success: 5, fail: 0, success_rate: 100.0, qps: 18.33, avg_ms: 218.0, p50_ms: 220.0, p95_ms: 300.0, p99_ms: 300.0 }不同机器、不同 mock 延迟参数下结果会不同数字不用纠结。重点是从这些指标里能看到成功率是否 100%、P99 延迟有没有异常上涨、QPS 是否满足你的任务量要求。再次运行同一条命令脚本会读取results.csv跳过已经成功的任务输出no pending tasks, all done。这就是断点续跑的基本行为。7. 功能测试与结果验证7.1 测试并发能力把任务文件扩充到 50 条或 100 条然后逐步增加--workerspython harness_bench.py \ --url http://127.0.0.1:8000/v1/completions \ --tasks tasks_100.jsonl \ --workers 1 \ --timeout 30 \ --record results_w1.csvpython harness_bench.py \ --url http://127.0.0.1:8000/v1/completions \ --tasks tasks_100.jsonl \ --workers 8 \ --timeout 30 \ --record results_w8.csv通过对比不同并发下的成功率、QPS 和 P99 延迟可以判断当前 harness 的并发瓶颈。如果并发从 1 调到 8QPS 没有明显提升可能是网络带宽、下游服务或 GIL 限制。如果成功率下降说明并发太高下游服务开始拒绝请求。7.2 测试失败重试用下面的命令启动一个带 20% 失败率的 mock 服务FAIL_RATE0.2 python mock_model_api.py然后运行 harness。你会在 CSV 里看到部分status为fail的记录。这说明脚本成功捕获了失败请求并把响应码和错误信息记录了下来。真实 harness 通常会在这里加入重试逻辑对 500、429、超时类错误进行退避重试重试 N 次仍然失败再放弃。当前最小实现里没有加重试目的是先把失败暴露出来方便你观察。7.3 测试超时控制把--timeout设成非常小的值比如 0.2 秒python harness_bench.py \ --url http://127.0.0.1:8000/v1/completions \ --tasks tasks_100.jsonl \ --workers 8 \ --timeout 0.2 \ --record results_timeout.csvmock 服务的随机延迟范围是 0.05 到 0.30 秒所以会有一部分请求超过 0.2 秒被标记为失败。这个测试能验证 harness 是否真的受超时参数控制而不是在底层无限等待。8. 接口 API 与批量任务扩展8.1 对接真实模型服务这个参考实现不绑定特定模型只要下游服务支持 HTTP POST JSON 请求即可。把--url换成你自己的模型服务地址比如python harness_bench.py \ --url http://127.0.0.1:11434/v1/chat/completions \ --tasks tasks_ollama.jsonl \ --workers 4 \ --timeout 120 \ --record results_ollama.csv如果服务要求鉴权需要在脚本里加上请求头。可以给call_api函数增加一个headers参数或者在调用requests.post时传入headers{Authorization: Bearer YOUR_TOKEN}。真实项目里建议把接口地址、鉴权信息和任务配置拆到独立的配置文件里避免把密钥写死在代码中。8.2 批量任务设计批量任务的工程化远不止并发调用还包括任务队列、进度跟踪、错误分类和结果归档。下面是一个推荐的任务状态流转待执行任务已经写进 JSONL但还没有发给服务。执行中请求已经发出正在等待响应。成功收到 2xx 响应结果落盘。失败请求超时、连接错误或收到 5xx。重试中失败的请求进入重试队列按指数退避重新执行。放弃重试多次仍然失败记录错误后进入失败列表。参考实现里的results.csv相当于一个简化版的任务状态表。生产环境可以用 SQLite 或数据库来维护任务状态这样即使进程崩溃也能通过任务状态恢复执行。9. 资源占用与性能观察9.1 观察 Harness 本身的资源占用harness 本身的资源占用取决于任务类型和并发方式。当前这个参考实现使用的是ThreadPoolExecutor加requests属于同步阻塞模型。它能跑通但并发高时线程之间的上下文切换会消耗 CPU每个线程还会占用一定内存。观察 CPU 和内存可以使用系统自带工具topps -o pid,%cpu,%mem,rss,cmd -p PID如果 harness 跑在 GPU 机器上观察显存可以使用nvidia-smi -l 1需要特别说明的是如果被测模型是本地部署的nvidia-smi里看到的显存占用主要是模型推理占用的harness 本身通常不消耗 GPU。如果你发现 harness 进程也在吃显存那说明它可能把输入输出 batch 到了 GPU 上这属于实现细节需要根据实际代码判断。9.2 如何降低资源占用一个简单有效的办法是降低并发workers让任务排队执行而不是一拥而上。另一个办法是把requests换成异步客户端比如aiohttp或httpx用事件循环替代线程池可以在同样资源占用下支持更高并发。如果任务非常长比如需要几分钟才能完成的请求要注意--timeout不能设置得太短同时要重点观察长时间运行后的文件句柄和连接池是否泄漏。这属于稳定性测试范围通常需要至少跑 30 分钟到数小时才能发现问题。10. 常见问题与排查方法问题现象可能原因排查方式解决方案启动 mock 服务时端口被占用8000 端口已被其他进程使用lsof -i :8000或 netstat -anofindstr :8000运行 harness 时提示ModuleNotFoundError: requests没有安装 requestspip listgrep requests全部任务失败错误信息是Connection refusedmock 服务未启动或 URL 写错先访问http://127.0.0.1:8000/health启动 mock 服务检查 URL 路径部分任务失败状态码 500mock 服务设置了失败率或下游服务异常查看 CSV 中http_status字段降低FAIL_RATE或检查真实服务日志任务一多就卡住线程数过高或下游服务性能不足观察 CPU、内存和下游日志降低--workers增加超时时间加入重试逻辑断点续跑没有跳过已完成任务--overwrite被误加或 CSV 中status不是ok检查 CSV 文件内容去掉--overwrite确保任务 ID 稳定QPS 很低单线程运行、网络延迟高、服务端限制并发检查workers参数观察 P50/P95 延迟逐步提高并发测试不同并发下的 QPS 变化P99 延迟远高于 P50部分请求排队或触发重试分析 CSV 中延迟分布优化下游服务调整并发和超时参数CSV 文件越来越大大量任务执行和日志记录查看磁盘空间按日期归档结果保留最近 N 次运行记录11. 最佳实践与合规提醒11.1 工程化建议先从小规模开始。不要第一次就跑到一千并发先用 10 条任务、1 个并发验证连通性再逐步增加。一次只改一个变量这样出问题时能快速定位。任务文件、模型配置、输出目录最好分开管理。比如bench/ tasks/ tasks_001.jsonl results/ results_001.csv logs/ run_001.log config.yaml批量任务一定要加日志和失败重试。日志里至少要有任务 ID、请求时间、响应时间、状态码、错误信息。这样即使任务失败也能定位到具体是哪个输入导致的问题。11.2 合规与安全边界harness-only benchmark 涉及高频并发请求必须明确测试边界。只能对以下目标做压测自己本地部署的模型服务。自建的 API 网关或测试环境。已经获得明确授权的外部服务。不要用这个脚本去压测公共接口。公共接口通常有速率限制高频请求既不礼貌也可能直接触发封禁甚至造成法律风险。所有测试数据都应该在受控环境中处理尤其是涉及真实用户数据、人脸图像、声音样本或版权文本的任务。11.3 结果复核最后提醒一个容易被忽略的点benchmark 跑出来的指标不代表业务效果。QPS 高不等于任务质量好。如果你用 harness 批量生成了内容一定要在发布或商用前做效果复核。自动化工具只解决“能不能跑完”的问题内容质量仍然需要人工抽查。12. 总结与下一步harness-only benchmark 的核心理念是把模型当作黑盒单独测试调度层的工程能力。它解决的不是“哪个模型更强”的问题而是“这套工具能不能在批量、并发、异常环境下稳定跑完任务”的问题。对于正在做模型评测、批量推理、Agent 工具链集成的团队来说这层测试是很有必要的。建议你先做三件事第一用 mock 服务跑通最小实现确认结果记录和指标输出正常第二在任务文件里加入几百条真实业务样本观察成功率和延迟分布第三人为制造失败比如提高 mock 的失败率验证你的 harness 是否能正确记录并处理错误。最容易踩的坑是在没有可观测性的情况下盲目上并发。先把 CSV 结果、错误信息、延迟指标这些基础能力建立起来再谈优化 QPS 和压测。后续可以继续扩展接入 OpenAI 兼容接口加入重试与指数退避把任务状态同步到数据库甚至可以做成一个独立的命令行工具在 CI 里对不同模型服务做定期稳定性回归。