hz4th_coder/llm-speed-tester
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2.3.0 多并发测试 + 多测试结果对比

Browse Source 
- 并发数配置：默认单流(1)，预设2/4并发档，支持丝滑添加任意自定义并发数(≥1)，
  多档勾选时同一测试分别跑各并发档并并排对比
- 并发执行：每采样同时发起N个并行流(ThreadPoolExecutor)，整批吞吐聚合
  (多流prompt/output之和/批首字/批耗时)，并记录每流明细streams；
  并发=1 与旧版单流行为一致；每流独立随机前缀避免共享缓存
- 汇总：新增 by_concurrency(按并发分组，含单流均解码) + by_length_concurrency(长度x并发网格) + concurrency_levels
- 详情页：新增「按并发数汇总」表 + 并发对比折线图(X=并发数，预填充左虚线/解码右实线, /concurrency-chart)
- 多测试对比：历史表格勾选(可全选)多个测试 -> /api/compare 返回对比表+柱状图CSV+并发折线图CSV，
  前端弹窗展示指标对比表 + 柱状图(预填充空心/解码实心) + 解码随并发折线图，画图CSV可复制/PNG可下载
- 通用图表代理 POST /api/chart（转发 data-chart-tool 请求体返回 PNG）
- 历史列表增加「并发」列与勾选列；Excel 导出增加并发数列表/按并发分组/采样并发列
- 文档：README/API.md 同步更新
This commit is contained in:
hz4th_coder
2026-09-01 19:33:41 +08:00
parent 2d889c80e0
commit 35ac2a8e40
8 changed files with 797 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tester.py
+165 -37
View File
@@ -57,55 +57,65 @@ class TestRunner(threading.Thread):
# 兼容旧版单值配置
raw_lengths = [int(gen.get("prompt_tokens", 2048))]
lengths = sorted(set(int(x) for x in raw_lengths if int(x) >= 16)) or [2048]
n = max(1, int(gen.get("samples", 2))) # 每个长度采样次数
n = max(1, int(gen.get("samples", 2))) # 每个 (长度×并发) 组合采样次数
max_tokens = max(1, int(gen.get("max_tokens", 128))) # 解码输出长度
avoid_cache = bool(gen.get("avoid_cache"))
warmup = bool(gen.get("warmup", True)) # 测试前空转预热
# 并发数列表（默认单流 [1]；支持 2/4 及自定义，如 [1,2,4,8]）
raw_concs = gen.get("concurrency_levels") or []
if not raw_concs:
raw_concs = [int(gen.get("concurrency", 1))]
concurrency_levels = sorted(set(int(x) for x in raw_concs if int(x) >= 1)) or [1]
self.log("INFO", "═══ 开始速度测试 ═══")
name = gen.get("name") or self.cfg.get("name") or ""
if name:
self.log("INFO", "测试名称(主题): %s" % name)
self.log("INFO", "提供商: %s | 模型: %s" % (lp.PROVIDER_LABELS.get(provider, provider), model))
self.log("INFO", "上下文长度: %s tokens | 生成长度: %d tokens | 每个长度采样: %d 次 | 预热: %s | 避免缓存: %s"
% (" / ".join(str(x) for x in lengths), max_tokens, n,
self.log("INFO", "上下文长度: %s tokens | 生成长度: %d tokens | 并发数: %s | 每个组合采样: %d 次 | 预热: %s | 避免缓存: %s"
% (" / ".join(str(x) for x in lengths), max_tokens,
" / ".join(str(x) for x in concurrency_levels), n,
"开" if warmup else "关", "开" if avoid_cache else "关"))
ratio = self._calibrate()
self.ratio = ratio
self.log("INFO", "校准完成: %.3f tok/字符（%.2f 字符/token）" % (ratio, 1.0 / ratio))
run_seq = 0
for L in lengths:
if self.should_stop():
raise StopRequested()
base_prompt = self._build_prompt(L, ratio)
self.log("INFO", "▸▸ 上下文长度 %d tokens（基准提示词构造完成）" % L)
if warmup:
self._warmup(base_prompt)
for i in range(1, n + 1):
for C in concurrency_levels:
if self.should_stop():
raise StopRequested()
prompt = self._finalize_prompt(base_prompt)
self.log("INFO", "── [%d tok] 采样 %d/%d 开始 ──" % (L, i, n))
try:
m = lp.call_stream(
self.cfg, prompt,
{"max_tokens": max_tokens, "avoid_cache": avoid_cache},
log=lambda lv, msg: self.log(lv, msg),
should_stop=self.should_stop)
m["run_index"] = i
m["context_length"] = L
self.samples.append({"run_index": i, "context_length": L, "ok": True, "metrics": m})
db.add_run(self.test_id, i, m, context_length=L)
self.log("METRIC", self._fmt_metric(L, i, n, m))
except StopRequested:
raise
except ProviderError as e:
# 单次采样失败：记录并继续后续采样，不让整个测试中断
self.last_error = str(e)
self.log("ERROR", "[%d tok] 采样 %d/%d 失败: %s" % (L, i, n, e))
self.samples.append({"run_index": i, "context_length": L, "ok": False, "error": str(e)})
db.add_run(self.test_id, i, {}, str(e), context_length=L)
self.log("INFO", "══ 并发数 %d（同时 %d 个流）══" % (C, C))
if warmup:
self._warmup(base_prompt, C)
for i in range(1, n + 1):
if self.should_stop():
raise StopRequested()
run_seq += 1
self.log("INFO", "── [%d tok · 并发%d] 采样 %d/%d 开始 ──" % (L, C, i, n))
try:
m = self._run_sample(C, base_prompt, max_tokens, avoid_cache)
m["run_index"] = i
m["context_length"] = L
self.samples.append({"run_index": i, "context_length": L,
"concurrency": C, "ok": True, "metrics": m})
db.add_run(self.test_id, run_seq, m, context_length=L)
self.log("METRIC", self._fmt_metric(L, C, i, n, m))
except StopRequested:
raise
except ProviderError as e:
# 单次采样失败：记录并继续后续采样，不让整个测试中断
self.last_error = str(e)
self.log("ERROR", "[%d tok · 并发%d] 采样 %d/%d 失败: %s" % (L, C, i, n, e))
self.samples.append({"run_index": i, "context_length": L,
"concurrency": C, "ok": False, "error": str(e)})
db.add_run(self.test_id, run_seq, {}, str(e), context_length=L)
summary = self._make_summary()
ok_count = summary.get("samples_ok") or 0
@@ -126,20 +136,98 @@ class TestRunner(threading.Thread):
summary.get("avg_prefill_speed") or 0,
summary.get("avg_decode_speed") or 0))
def _warmup(self, base_prompt):
"""空转预热：不计入任何速度统计，用于避免冷启动/首次请求偏慢影响采样"""
self.log("INFO", "预热（空转，不计速度）...")
def _warmup(self, base_prompt, concurrency=1):
"""空转预热：不计入任何速度统计，用于避免冷启动/首次请求偏慢影响采样（按并发数预热）"""
self.log("INFO", "预热（空转，不计速度，并发 %d）..." % concurrency)
try:
lp.call_stream(self.cfg, base_prompt,
{"max_tokens": 8, "avoid_cache": False},
log=lambda lv, msg: self.log(lv, msg),
should_stop=self.should_stop)
self._run_sample(concurrency, base_prompt, 8, False)
self.log("INFO", "预热完成（不纳入统计）")
except StopRequested:
raise
except Exception as e:
self.log("WARN", "预热失败（继续测试）: %s" % e)
# ───────────────────────── 并发采样 ─────────────────────────
def _run_sample(self, concurrency, base_prompt, max_tokens, avoid_cache):
"""
运行一个采样：concurrency 个流同时并发请求（并发=1 即单流）。
返回聚合指标：prompt/output tokens 为 N 流之和，
prefill/decode 速度为“整批吞吐”（tok/s），并附每流明细 streams。
"""
from concurrent.futures import ThreadPoolExecutor
gen_opt = {"max_tokens": max_tokens, "avoid_cache": avoid_cache}
def worker(idx):
prompt = self._finalize_prompt(base_prompt) # 每流独立随机前缀，避免共享缓存
t0 = time.time()
try:
m = lp.call_stream(self.cfg, prompt, gen_opt,
log=lambda lv, msg: self.log(lv, msg),
should_stop=self.should_stop)
m["_wall_start"] = t0
m["_wall_end"] = time.time()
m["_stream_idx"] = idx
return {"ok": True, "metrics": m}
except StopRequested:
raise
except ProviderError as e:
return {"ok": False, "error": str(e)}
except Exception as e:
return {"ok": False, "error": str(e)}
with ThreadPoolExecutor(max_workers=concurrency) as ex:
results = list(ex.map(worker, range(concurrency)))
ok = [r["metrics"] for r in results if r.get("ok")]
streams_detail = [self._clean_stream(m, m.get("_stream_idx", 0)) for m in ok]
if not ok:
errs = [r.get("error") or "未知错误" for r in results if not r.get("ok")]
raise ProviderError("并发 %d 全部失败: %s" % (concurrency, " | ".join(errs[:3])))
total_prompt = sum(m.get("prompt_tokens") or 0 for m in ok)
total_output = sum(m.get("output_tokens") or 0 for m in ok)
total_cached = sum(m.get("cached_tokens") or 0 for m in ok)
total_pchars = sum(m.get("prompt_chars") or 0 for m in ok)
total_ochars = sum(m.get("output_chars") or 0 for m in ok)
batch_start = min(m["_wall_start"] for m in ok)
# 整批首字时刻 = 任一流最早收到第一个 token 的时刻
first_at = min(m["_wall_start"] + (m.get("ttft_ms") or 0) / 1000.0 for m in ok)
batch_end = max(m["_wall_end"] for m in ok)
ttft_ms = max((first_at - batch_start) * 1000.0, 0.1)
decode_ms = max((batch_end - first_at) * 1000.0, 0.1)
total_ms = max((batch_end - batch_start) * 1000.0, 0.1)
prefill = (total_prompt / (ttft_ms / 1000.0)) if total_prompt else None
decode = (total_output / (decode_ms / 1000.0)) if total_output else None
agg = {
"concurrency": concurrency,
"streams_total": concurrency,
"streams_ok": len(ok),
"prompt_tokens": int(total_prompt),
"output_tokens": int(total_output),
"cached_tokens": int(total_cached),
"prompt_chars": int(total_pchars),
"output_chars": int(total_ochars),
"ttft_ms": round(ttft_ms, 1),
"decode_ms": round(decode_ms, 1),
"total_ms": round(total_ms, 1),
"prefill_speed": round(prefill, 1) if prefill else None,
"decode_speed": round(decode, 1) if decode else None,
"avg_stream_prefill": round(prefill / concurrency, 1) if prefill else None,
"avg_stream_decode": round(decode / concurrency, 1) if decode else None,
"streams": streams_detail,
}
return agg
@staticmethod
def _clean_stream(m, idx):
"""去掉内部 _wall 字段，保留每流可展示指标"""
keep = {k: v for k, v in m.items() if not k.startswith("_")}
keep["stream_idx"] = idx
return keep
# ───────────────────────── 工具方法 ─────────────────────────
def _calibrate(self):
@@ -175,10 +263,11 @@ class TestRunner(threading.Thread):
return "[cache-bust %s]\n%s" % (uuid.uuid4().hex, base)
return base
def _fmt_metric(self, L, i, n, m):
return ("[%d tok] 采样 %d/%d 完成 | 提示词 %d tok | 缓存 %d tok | 首字 %s ms | 预填充 %s tok/s"
def _fmt_metric(self, L, C, i, n, m):
return ("[%d tok · 并发%d] 采样 %d/%d 完成 | 流 %d/%d 成功 | 提示词 %d tok | 缓存 %d tok | 首字 %s ms | 预填充 %s tok/s"
" | 输出 %d tok | 解码 %s tok/s | 总耗时 %s ms"
% (L, i, n, m.get("prompt_tokens") or 0, m.get("cached_tokens") or 0,
% (L, C, i, n, m.get("streams_ok") or 0, m.get("streams_total") or C,
m.get("prompt_tokens") or 0, m.get("cached_tokens") or 0,
m.get("ttft_ms"), m.get("prefill_speed"), m.get("output_tokens") or 0,
m.get("decode_speed"), m.get("total_ms")))
@@ -190,6 +279,7 @@ class TestRunner(threading.Thread):
"gen": self.gen,
"samples_total": len(self.samples),
"samples_ok": len(ok),
"concurrency_levels": sorted(set(s.get("concurrency", 1) for s in self.samples)) or [1],
"calibration_chars_per_token": round(1 / self.ratio, 2) if self.ratio else None,
}
if not ok:
@@ -214,6 +304,42 @@ class TestRunner(threading.Thread):
"avg_total_ms": avg(group, "total_ms"),
}
# 按并发数分组汇总（多测试结果并排对比的核心数据）
by_concurrency = {}
for C in sorted(set(s.get("concurrency", 1) for s in ok)):
group = [s["metrics"] for s in ok if s.get("concurrency", 1) == C]
by_concurrency[C] = {
"samples_total": sum(1 for s in self.samples if s.get("concurrency", 1) == C),
"samples_ok": len(group),
"avg_ttft_ms": avg(group, "ttft_ms"),
"avg_prefill_speed": avg(group, "prefill_speed"),
"avg_decode_speed": avg(group, "decode_speed"),
"avg_stream_decode": avg(group, "avg_stream_decode"),
"avg_prompt_tokens": avg(group, "prompt_tokens"),
"avg_output_tokens": avg(group, "output_tokens"),
"avg_total_ms": avg(group, "total_ms"),
}
# 长度 × 并发 全网格（详情/Excel 用）
by_length_concurrency = {}
for L in sorted(set(s["context_length"] for s in ok)):
grid = {}
for C in sorted(set(s.get("concurrency", 1) for s in ok)):
group = [s["metrics"] for s in ok
if s["context_length"] == L and s.get("concurrency", 1) == C]
grid[C] = {
"samples_total": sum(1 for s in self.samples
if s["context_length"] == L and s.get("concurrency", 1) == C),
"samples_ok": len(group),
"avg_ttft_ms": avg(group, "ttft_ms"),
"avg_prefill_speed": avg(group, "prefill_speed"),
"avg_decode_speed": avg(group, "decode_speed"),
"avg_prompt_tokens": avg(group, "prompt_tokens"),
"avg_output_tokens": avg(group, "output_tokens"),
"avg_total_ms": avg(group, "total_ms"),
}
by_length_concurrency[L] = grid
okm = [s["metrics"] for s in ok]
def mn(k):
@@ -227,6 +353,8 @@ class TestRunner(threading.Thread):
summary = dict(base)
summary.update({
"by_length": by_length,
"by_concurrency": by_concurrency,
"by_length_concurrency": by_length_concurrency,
"avg_ttft_ms": avg(okm, "ttft_ms"),
"min_ttft_ms": mn("ttft_ms"),
"max_ttft_ms": mx("ttft_ms"),