企业官网建设流程全解析

有段时间没聊评测工具链了。这次 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. 核心能力速览

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，不是随便并发打几个请求就算数。建议按下面的维度来设计测试用例。

这里有一个容易被忽略的点：并发吞吐不是越高越好。很多框架能顶着高并发把 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(host="127.0.0.1", port=8000)

启动方式：

python mock_model_api.py

默认情况下，所有请求都会在 0.05 到 0.30 秒之间随机延迟后返回成功。如果你需要模拟下游不稳定的场景，可以这样启动：

FAIL_RATE=0.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", encoding="utf-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", encoding="utf-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, json=payload, timeout=timeout) 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", encoding="utf-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_workers=args.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_ascii=False, indent=2)) print("wall_time_seconds:", round(elapsed, 2)) def main(): parser = argparse.ArgumentParser(description="harness-only benchmark minimal implementation") parser.add_argument("--url", required=True, help="model API URL") parser.add_argument("--tasks", required=True, help="JSONL tasks file path") parser.add_argument("--workers", type=int, default=4, help="concurrent workers") parser.add_argument("--timeout", type=int, default=30, help="request timeout in seconds") parser.add_argument("--max-tasks", type=int, default=0, help="max tasks to load") parser.add_argument("--record", default="results.csv", help="output CSV path") parser.add_argument("--overwrite", action="store_true", help="ignore 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

新开一个终端，运行 harness：

python 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 条，然后逐步增加--workers：

python harness_bench.py \ --url http://127.0.0.1:8000/v1/completions \ --tasks tasks_100.jsonl \ --workers 1 \ --timeout 30 \ --record results_w1.csv

python 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 调到 8，QPS 没有明显提升，可能是网络带宽、下游服务或 GIL 限制。如果成功率下降，说明并发太高，下游服务开始拒绝请求。

7.2 测试失败重试

用下面的命令启动一个带 20% 失败率的 mock 服务：

FAIL_RATE=0.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.csv

mock 服务的随机延迟范围是 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 和内存可以使用系统自带工具：

top

ps -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. 常见问题与排查方法

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 里对不同模型服务做定期稳定性回归。