From 0adce51520eff869946f03765987b4cc3379d067 Mon Sep 17 00:00:00 2001 From: sufubao Date: Wed, 5 Aug 2026 19:57:55 +0800 Subject: [PATCH 1/5] test/benchmark: add --suite mode and fix bugs in bench_agents.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 给 bench_agents.py 加了一键健壮性/性能测试套件,并修了几个会导致结果失真的 bug。 Bug 修复: - 流式 token 字段捕获:原来只读 delta.content / reasoning_content,thinking 模型输出在 reasoning 字段时会整条流 0 token,误报 "no tokens"。改为通过 pydantic model_extra 兜底读取 reasoning / reasoning_content / text 等任意字段。 - 并发重复派发:fire_one 只检查 s.pending 就 create_task,但 pending 在 do_turn 里才置位;catch-up 突发时同一 session 会被重复选中(饱和区最关键处失真)。 改为 fire_one 里先占位再调度。 - rr_idx 污染:do_turn 的 finally 里不该自增发射游标 rr_idx,会与 fire_one 的 游标推进冲突,移除。 - ITL 计算:原仅取 token1→token2 单点采样,改为全部相邻间隔的均值,并补上汇总输出。 新增能力: - --suite:一键跑 smoke -> QPS 扫描(找饱和拐点) -> 过载 -> idle-then-burst, 可选 --full 追加长稳阶段。每阶段落 *.log,结尾输出 markdown 总表 report.md。 - 实时进度 reporter(--report-sec):每行打印 fired/blocked/ok/err/pend/achQPS 及当前 TTFT/TPOT,便于长跑观察。 - 默认端口改为 8000;文件头补完整用法说明(单次 bench + 套件两种模式、关键参数、 输出判读)。 --- test/benchmark/bench_agents.py | 278 ++++++++++++++++++++++++++++----- 1 file changed, 243 insertions(+), 35 deletions(-) diff --git a/test/benchmark/bench_agents.py b/test/benchmark/bench_agents.py index 93458d7ad..7989815a0 100644 --- a/test/benchmark/bench_agents.py +++ b/test/benchmark/bench_agents.py @@ -1,29 +1,69 @@ -# Example: -# python test/benchmark/bench_agents.py \ -# --qps 5 \ -# --num-sessions 32 \ -# --turns-per-session 6 \ -# --isl 4096 \ -# --osl 256 \ -# --duration-sec 60 \ -# --warmup-sec 10 +"""Agent-style multi-turn open-loop benchmark + one-shot robustness/perf suite. + +What it does +------------ +N concurrent agent sessions, each with a distinct long system prompt (so a +cache_aware router spreads them across prefill nodes); turns within a session +re-send the growing history (KV reuse). Turn emission is OPEN-LOOP: one turn is +fired every 1/QPS seconds regardless of whether previous turns finished. If every +session is still busy when a tick fires, the tick is counted as "blocked" -- the +saturation signal. + +Two modes +--------- +1) Single bench (default). Sweep one operating point and print full stats. + + python test/benchmark/bench_agents.py \\ + --base-url http://127.0.0.1:8000/v1 --model qwen35 \\ + --qps 5 --num-sessions 32 --turns-per-session 6 \\ + --isl 2048 --osl 256 --duration-sec 60 --warmup-sec 10 + +2) Suite (--suite). Runs a scripted sequence: smoke -> QPS sweep (find the + saturation knee) -> overload -> idle-then-burst, [+ long-run with --full]. + Prints a compact progress line per phase and a markdown summary table at the + end; each phase's full output goes to /.log and the table to + /report.md. + + python test/benchmark/bench_agents.py --suite # ~7-8 min + python test/benchmark/bench_agents.py --suite --quick # halve durations + python test/benchmark/bench_agents.py --suite --full # + 10-min long-run + python test/benchmark/bench_agents.py --suite --isl 4096 --osl 512 + +Key flags +--------- + --qps turn emission rate (open loop) + --num-sessions concurrent sessions; distinct prefix each (= concurrency cap) + --turns-per-session turns before a session is recycled into a fresh identity + --isl / --osl system-prompt length (tokens) / max output tokens per turn + --duration-sec emission window + --warmup-sec discard metrics before this (also used as the idle period in + the suite's idle_burst phase) + --report-sec live progress print interval (0 disables) + --suite / --quick / --full / --out suite mode controls + +Reading the output +------------------ + - Saturation knee: the QPS step where achieved QPS falls below target, or TTFT + p95 spikes. + - blocked > 0: the session pool filled before the server did -- raise + --num-sessions and rerun that step. + - idle_burst fail>0: requests rejected right after an idle window (health + misclassification / spurious 503). + - error breakdown: top error strings tell you whether failures are timeouts, + connection resets, or empty streams. +""" import argparse import asyncio +import contextlib +import io import math import time +from pathlib import Path from typing import Any, Dict, List, Optional from openai import AsyncOpenAI -# Agent-style open-loop benchmark. -# - N sessions, each with a distinct long system prompt (so cache_aware spreads -# them across P nodes); turns within a session share that prefix (KV reuse). -# - Fixed turn-emission rate (open loop): the scheduler fires one turn every -# 1/QPS seconds. If every session is still waiting on its previous turn, the -# tick is counted as blocked -> the saturation signal. -# - Prefix grows each turn (full history re-sent), mirroring real multi-turn agents. - def percentile(data: List[float], p: float) -> float: if not data: @@ -74,7 +114,7 @@ def make_user_turn(session_seed: int, turn_idx: int) -> str: class Session: - __slots__ = ("sid", "system", "history", "pending", "turns_done", "birth") + __slots__ = ("sid", "system", "history", "pending", "turns_done") def __init__(self, sid: int, system: str): self.sid = sid @@ -82,7 +122,6 @@ def __init__(self, sid: int, system: str): self.history: List[Dict[str, str]] = [] self.pending = False self.turns_done = 0 - self.birth = 0.0 def reset(self, system: str): self.system = system @@ -98,13 +137,13 @@ async def do_turn( osl: int, metrics: List[Dict[str, Any]], warmup_until: float, - rr_idx: List[int], ): user_msg = make_user_turn(sess.sid, sess.turns_done) messages = [{"role": "system", "content": sess.system}] + sess.history + [{"role": "user", "content": user_msg}] - sess.pending = True t0 = time.perf_counter() - first_t = second_t = last_t = None + first_t = last_t = prev_t = None + itl_sum = 0.0 + itl_n = 0 ntok = 0 err = None try: @@ -119,14 +158,23 @@ async def do_turn( if not (chunk.choices and chunk.choices[0].delta): continue delta = chunk.choices[0].delta - text = getattr(delta, "content", None) or getattr(delta, "reasoning_content", None) + extra = getattr(delta, "model_extra", None) or {} + text = ( + getattr(delta, "content", None) + or extra.get("reasoning") + or extra.get("reasoning_content") + or extra.get("text") + or getattr(delta, "reasoning", None) + ) if text: - ntok += 1 now = time.perf_counter() - if ntok == 1: + ntok += 1 + if first_t is None: first_t = now - elif ntok == 2: - second_t = now + else: + itl_sum += now - prev_t + itl_n += 1 + prev_t = now last_t = now end = time.perf_counter() if ntok < 1: @@ -138,7 +186,7 @@ async def do_turn( metrics.append( { "ttft": (first_t - t0) * 1000, - "itl": ((second_t - first_t) * 1000) if (first_t and second_t) else None, + "itl": (itl_sum / itl_n * 1000) if itl_n else None, "tpot": (((last_t - first_t) / (ntok - 1)) * 1000) if (ntok > 1 and first_t and last_t) else 0.0, @@ -154,7 +202,6 @@ async def do_turn( sess.pending = False if err: metrics.append({"error": err, "turn_idx": sess.turns_done, "sid": sess.sid}) - rr_idx[0] += 1 async def run(args): @@ -178,6 +225,35 @@ async def run(args): end_at = start + args.duration_sec next_tick = start + async def reporter(): + # periodic one-liner: progress + running TTFT/TPOT over metrics so far + try: + while True: + await asyncio.sleep(args.report_sec) + now = time.perf_counter() + if now >= end_at: + return + ok_now = [m for m in metrics if "error" not in m] + errs = [m for m in metrics if "error" in m] + elapsed = now - start + done = len(ok_now) + ttft_s = stats([m["ttft"] for m in ok_now]) + tpot_s = stats([m["tpot"] for m in ok_now if m["tpot"] > 0]) + in_flight = sum(1 for s in sessions if s.pending) + last_err = f" last_err={errs[-1]['error'][:120]!r}" if errs else "" + print( + f"[{elapsed:5.1f}s] fired={fired:<5} blocked={blocked:<5} " + f"ok={done:<5} err={len(errs):<3} pend={in_flight:<3} " + f"achQPS={done / elapsed if elapsed > 0 else 0:.2f} " + f"TTFT p50={ttft_s['p50']:.0f} p95={ttft_s['p95']:.0f} max={ttft_s['max']:.0f} " + f"TPOT p50={tpot_s['p50']:.0f} p95={tpot_s['p95']:.0f}" + f"{last_err}" + ) + except asyncio.CancelledError: + pass + + reporter_task = asyncio.create_task(reporter()) if args.report_sec > 0 else None + async def fire_one(): nonlocal fired, blocked # pick next non-pending session (round-robin over ready ones) @@ -196,8 +272,10 @@ async def fire_one(): # retire a finished session into a fresh identity so the pool stays populated if chosen.turns_done >= args.turns_per_session: chosen.reset(system=make_system_prompt(seed=10_000_000 + rr_idx[0] + fired, isl=args.isl)) - chosen.birth = time.perf_counter() - asyncio.create_task(do_turn(client, args.model, chosen, args.osl, metrics, warmup_until, rr_idx)) + # claim before scheduling: fire_one doesn't await, so without this a + # catch-up burst re-picks the same session before do_turn runs. + chosen.pending = True + asyncio.create_task(do_turn(client, args.model, chosen, args.osl, metrics, warmup_until)) # open-loop arrival loop: fire at fixed rate regardless of completions while True: @@ -216,15 +294,19 @@ async def fire_one(): deadline = time.perf_counter() + args.drain_sec while any(s.pending for s in sessions) and time.perf_counter() < deadline: await asyncio.sleep(0.05) + reporter_task and reporter_task.cancel() + if reporter_task: + await reporter_task total_time = time.perf_counter() - start - summarize(metrics, fired, blocked, total_time, args) + return summarize(metrics, fired, blocked, total_time, args) def summarize(metrics, fired, blocked, total_time, args): ok = [m for m in metrics if "error" not in m] fail = [m for m in metrics if "error" in m] ttfts = [m["ttft"] for m in ok] + itls = [m["itl"] for m in ok if m["itl"] is not None] tpots = [m["tpot"] for m in ok if m["tpot"] > 0] lats = [m["lat"] for m in ok] @@ -245,9 +327,17 @@ def line(name, s): print(f" target QPS : {args.qps}") print(f" fired turns : {fired} blocked ticks (saturation): {blocked}") print(f" completed (ok/fail): {len(ok)} / {len(fail)}") + if fail: + from collections import Counter + + c = Counter(m["error"][:160] for m in fail) + print(" error breakdown:") + for msg, n in c.most_common(5): + print(f" [{n:>3}] {msg}") print(f" achieved turn QPS : {len(ok) / total_time:.2f} /s (wall {total_time:.1f}s)") print("-" * 96) line("TTFT", stats(ttfts)) + line("ITL", stats(itls)) line("TPOT", stats(tpots)) line("Total turn latency", stats(lats)) print("-" * 96) @@ -260,11 +350,109 @@ def line(name, s): ) print("=" * 96 + "\n") + from collections import Counter -def main(): + top_err = Counter(m["error"][:120] for m in fail).most_common(1) + return { + "target_qps": args.qps, + "fired": fired, + "blocked": blocked, + "ok": len(ok), + "fail": len(fail), + "achieved_qps": len(ok) / total_time if total_time > 0 else 0.0, + "wall": total_time, + "ttft": stats(ttfts), + "itl": stats(itls), + "tpot": stats(tpots), + "lat": stats(lats), + "top_error": top_err[0][0] if top_err else None, + } + + +def _run_phase(name: str, out_dir, base: dict, **kw) -> dict: + args = make_args(**{**base, **kw, "report_sec": 5}) + print(f"\n>>> {name} qps={args.qps} sessions={args.num_sessions} " + f"isl={args.isl} osl={args.osl} dur={args.duration_sec}s warmup={args.warmup_sec}s", flush=True) + buf = io.StringIO() + t0 = time.perf_counter() + with contextlib.redirect_stdout(buf): + res = asyncio.run(run(args)) + res["phase"] = name + res["suite_wall"] = time.perf_counter() - t0 + Path(out_dir, f"{name}.log").write_text(buf.getvalue()) + ttft, tpot = res["ttft"], res["tpot"] + err = f" err={res['fail']}({res['top_error'][:40]})" if res["fail"] else "" + print(f" done in {res['suite_wall']:4.0f}s | ok={res['ok']:<4} fail={res['fail']:<3} " + f"blocked={res['blocked']:<3} achQPS={res['achieved_qps']:.2f} " + f"TTFT p50={ttft['p50']:.0f} p95={ttft['p95']:.0f} " + f"TPOT p50={tpot['p50']:.0f} p95={tpot['p95']:.0f}{err}", flush=True) + return res + + +def _md_table(rows: list) -> str: + head = ("| phase | targetQPS | fired | ok | fail | blocked | achQPS | " + "TTFT p50 | TTFT p95 | TPOT p50 | TPOT p95 | top error |\n" + "|---|---|---|---|---|---|---|---|---|---|---|---|") + body = [] + for r in rows: + t, p = r["ttft"], r["tpot"] + body.append( + f"| {r['phase']} | {r['target_qps']} | {r['fired']} | {r['ok']} | {r['fail']} | " + f"{r['blocked']} | {r['achieved_qps']:.2f} | " + f"{t['p50']:.0f} | {t['p95']:.0f} | {p['p50']:.0f} | {p['p95']:.0f} | " + f"{(r['top_error'] or '')[:50]} |" + ) + return head + "\n" + "\n".join(body) + + +def run_suite(args) -> None: + import os + + d = 0.5 if args.quick else 1.0 + out_dir = args.out + os.makedirs(out_dir, exist_ok=True) + base = {"base_url": args.base_url, "model": args.model, "isl": args.isl, "osl": args.osl} + + def dur(s): + return max(10.0, s * d) + + phases = [ + ("smoke", dict(qps=1, num_sessions=2, turns_per_session=3, duration_sec=dur(20), warmup_sec=dur(5))), + ("perf_q2", dict(qps=2, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), + ("perf_q5", dict(qps=5, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), + ("perf_q10", dict(qps=10, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), + ("perf_q20", dict(qps=20, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), + ("overload_q40", dict(qps=40, num_sessions=128, duration_sec=dur(60), warmup_sec=0)), + ("idle_burst", dict(qps=20, num_sessions=64, duration_sec=dur(90), warmup_sec=dur(45))), + ] + if args.full: + phases.append( + ("longrun", dict(qps=8, num_sessions=64, duration_sec=dur(600), warmup_sec=dur(30), report_sec=30)) + ) + + print(f"SUITE -> {args.base_url} model={args.model} isl={args.isl} osl={args.osl} " + f"quick={args.quick} full={args.full} out={out_dir}", flush=True) + t_start = time.perf_counter() + results = [_run_phase(name, out_dir, base, **kw) for name, kw in phases] + total = time.perf_counter() - t_start + + table = _md_table(results) + print("\n" + "=" * 96) + print(f"SUITE DONE in {total:.0f}s ({len(results)} phases)") + print("=" * 96) + print(table) + Path(out_dir, "report.md").write_text( + f"# Bench suite report\n\nTarget: `{args.base_url}` model=`{args.model}` " + f"isl={args.isl} osl={args.osl} quick={args.quick} full={args.full}\n\n" + f"Total wall: {total:.0f}s\n\n{table}\n" + ) + print(f"\nreport -> {Path(out_dir, 'report.md')} (per-phase logs: {out_dir}/*.log)") + + +def build_parser(): p = argparse.ArgumentParser(description="Agent multi-turn open-loop benchmark") p.add_argument("--api-key", default="EMPTY") - p.add_argument("--base-url", default="http://127.0.0.1:16666/v1") + p.add_argument("--base-url", default="http://127.0.0.1:8000/v1") p.add_argument("--model", default="qwen35") p.add_argument("--qps", type=float, default=5.0, help="turn emission rate (open loop)") p.add_argument("--num-sessions", type=int, default=32, help="concurrent sessions; distinct prefix each") @@ -274,7 +462,27 @@ def main(): p.add_argument("--duration-sec", type=float, default=60.0, help="emission window") p.add_argument("--warmup-sec", type=float, default=10.0, help="discard metrics before this") p.add_argument("--drain-sec", type=float, default=120.0, help="max wait for in-flight turns after window") - asyncio.run(run(p.parse_args())) + p.add_argument("--report-sec", type=float, default=5.0, help="live progress print interval (0 disables)") + p.add_argument("--suite", action="store_true", help="run the full robustness+perf suite instead of one bench") + p.add_argument("--quick", action="store_true", help="(suite) halve all durations") + p.add_argument("--full", action="store_true", help="(suite) add a 10-min long-run stability phase") + p.add_argument("--out", default="bench_suite_out", help="(suite) dir for per-phase logs + report.md") + return p + + +def make_args(**overrides): + args = build_parser().parse_args([]) + for k, v in overrides.items(): + setattr(args, k, v) + return args + + +def main(): + args = build_parser().parse_args() + if args.suite: + run_suite(args) + else: + asyncio.run(run(args)) if __name__ == "__main__": From 028ec927366ab2eb6e7cd70b3b3ec59f78bf6bc4 Mon Sep 17 00:00:00 2001 From: sufubao Date: Wed, 5 Aug 2026 20:03:33 +0800 Subject: [PATCH 2/5] test/benchmark: append real generated reply to history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原来 history 里追加的是硬编码桩 "(reply)",下一轮 prefix 几乎不增长, 不像真实多轮 agent。改为累计本轮流式真实输出(content + reasoning)写回 history,使每轮 prefix 真实增长,模拟真实 agent 调用。 --- test/benchmark/bench_agents.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/benchmark/bench_agents.py b/test/benchmark/bench_agents.py index 7989815a0..daf87b1c0 100644 --- a/test/benchmark/bench_agents.py +++ b/test/benchmark/bench_agents.py @@ -145,6 +145,7 @@ async def do_turn( itl_sum = 0.0 itl_n = 0 ntok = 0 + reply_parts: List[str] = [] err = None try: resp = await client.chat.completions.create( @@ -167,6 +168,7 @@ async def do_turn( or getattr(delta, "reasoning", None) ) if text: + reply_parts.append(text) now = time.perf_counter() ntok += 1 if first_t is None: @@ -180,8 +182,9 @@ async def do_turn( if ntok < 1: err = "no tokens" else: - # append assistant reply so the next turn's prefix includes it - sess.history.append({"role": "assistant", "content": "(reply)"}) + # append the real generated reply so the next turn's prefix grows + # like a real multi-turn agent (full history re-sent) + sess.history.append({"role": "assistant", "content": "".join(reply_parts)}) if time.perf_counter() >= warmup_until: metrics.append( { From 1037a64e2c65406a228b5dc5f56d47823d741a23 Mon Sep 17 00:00:00 2001 From: sufubao Date: Thu, 6 Aug 2026 16:04:16 +0800 Subject: [PATCH 3/5] test/benchmark: apply black formatting to bench_agents.py --- test/benchmark/bench_agents.py | 33 ++++++++++++++++++++++----------- 1 file changed, 22 insertions(+), 11 deletions(-) diff --git a/test/benchmark/bench_agents.py b/test/benchmark/bench_agents.py index daf87b1c0..a2b0c2b3c 100644 --- a/test/benchmark/bench_agents.py +++ b/test/benchmark/bench_agents.py @@ -374,8 +374,11 @@ def line(name, s): def _run_phase(name: str, out_dir, base: dict, **kw) -> dict: args = make_args(**{**base, **kw, "report_sec": 5}) - print(f"\n>>> {name} qps={args.qps} sessions={args.num_sessions} " - f"isl={args.isl} osl={args.osl} dur={args.duration_sec}s warmup={args.warmup_sec}s", flush=True) + print( + f"\n>>> {name} qps={args.qps} sessions={args.num_sessions} " + f"isl={args.isl} osl={args.osl} dur={args.duration_sec}s warmup={args.warmup_sec}s", + flush=True, + ) buf = io.StringIO() t0 = time.perf_counter() with contextlib.redirect_stdout(buf): @@ -385,17 +388,22 @@ def _run_phase(name: str, out_dir, base: dict, **kw) -> dict: Path(out_dir, f"{name}.log").write_text(buf.getvalue()) ttft, tpot = res["ttft"], res["tpot"] err = f" err={res['fail']}({res['top_error'][:40]})" if res["fail"] else "" - print(f" done in {res['suite_wall']:4.0f}s | ok={res['ok']:<4} fail={res['fail']:<3} " - f"blocked={res['blocked']:<3} achQPS={res['achieved_qps']:.2f} " - f"TTFT p50={ttft['p50']:.0f} p95={ttft['p95']:.0f} " - f"TPOT p50={tpot['p50']:.0f} p95={tpot['p95']:.0f}{err}", flush=True) + print( + f" done in {res['suite_wall']:4.0f}s | ok={res['ok']:<4} fail={res['fail']:<3} " + f"blocked={res['blocked']:<3} achQPS={res['achieved_qps']:.2f} " + f"TTFT p50={ttft['p50']:.0f} p95={ttft['p95']:.0f} " + f"TPOT p50={tpot['p50']:.0f} p95={tpot['p95']:.0f}{err}", + flush=True, + ) return res def _md_table(rows: list) -> str: - head = ("| phase | targetQPS | fired | ok | fail | blocked | achQPS | " - "TTFT p50 | TTFT p95 | TPOT p50 | TPOT p95 | top error |\n" - "|---|---|---|---|---|---|---|---|---|---|---|---|") + head = ( + "| phase | targetQPS | fired | ok | fail | blocked | achQPS | " + "TTFT p50 | TTFT p95 | TPOT p50 | TPOT p95 | top error |\n" + "|---|---|---|---|---|---|---|---|---|---|---|---|" + ) body = [] for r in rows: t, p = r["ttft"], r["tpot"] @@ -433,8 +441,11 @@ def dur(s): ("longrun", dict(qps=8, num_sessions=64, duration_sec=dur(600), warmup_sec=dur(30), report_sec=30)) ) - print(f"SUITE -> {args.base_url} model={args.model} isl={args.isl} osl={args.osl} " - f"quick={args.quick} full={args.full} out={out_dir}", flush=True) + print( + f"SUITE -> {args.base_url} model={args.model} isl={args.isl} osl={args.osl} " + f"quick={args.quick} full={args.full} out={out_dir}", + flush=True, + ) t_start = time.perf_counter() results = [_run_phase(name, out_dir, base, **kw) for name, kw in phases] total = time.perf_counter() - t_start From 5f2ed7c2c52dcda94c4f2daf091b116e0208b54d Mon Sep 17 00:00:00 2001 From: sufubao Date: Thu, 6 Aug 2026 19:43:46 +0800 Subject: [PATCH 4/5] test/benchmark: turn agent suite into stress test --- test/benchmark/bench_agents.py | 419 ++++++++++++++++++++++++--------- 1 file changed, 311 insertions(+), 108 deletions(-) diff --git a/test/benchmark/bench_agents.py b/test/benchmark/bench_agents.py index a2b0c2b3c..3d197025a 100644 --- a/test/benchmark/bench_agents.py +++ b/test/benchmark/bench_agents.py @@ -1,4 +1,4 @@ -"""Agent-style multi-turn open-loop benchmark + one-shot robustness/perf suite. +"""Agent-style multi-turn open-loop stress test. What it does ------------ @@ -11,23 +11,22 @@ Two modes --------- -1) Single bench (default). Sweep one operating point and print full stats. +1) Single load point (default). Apply a fixed QPS for a fixed duration. - python test/benchmark/bench_agents.py \\ + exp -m "single agent stress load" python test/benchmark/bench_agents.py \\ --base-url http://127.0.0.1:8000/v1 --model qwen35 \\ --qps 5 --num-sessions 32 --turns-per-session 6 \\ --isl 2048 --osl 256 --duration-sec 60 --warmup-sec 10 -2) Suite (--suite). Runs a scripted sequence: smoke -> QPS sweep (find the - saturation knee) -> overload -> idle-then-burst, [+ long-run with --full]. - Prints a compact progress line per phase and a markdown summary table at the - end; each phase's full output goes to /.log and the table to - /report.md. +2) Stress test (--stress). Increase QPS step by step until the target is no + longer sustained, then keep that pressure for a hold phase. Every phase is + logged and a markdown report records where saturation started. - python test/benchmark/bench_agents.py --suite # ~7-8 min - python test/benchmark/bench_agents.py --suite --quick # halve durations - python test/benchmark/bench_agents.py --suite --full # + 10-min long-run - python test/benchmark/bench_agents.py --suite --isl 4096 --osl 512 + exp -m "agent stress ramp" python test/benchmark/bench_agents.py --stress + exp -m "quick agent stress ramp" python test/benchmark/bench_agents.py --stress --quick + exp -m "custom agent stress ramp" python test/benchmark/bench_agents.py --stress \\ + --stress-start-qps 5 --stress-step-qps 5 --stress-max-qps 100 \\ + --stress-step-sec 60 --stress-hold-sec 300 Key flags --------- @@ -36,19 +35,16 @@ --turns-per-session turns before a session is recycled into a fresh identity --isl / --osl system-prompt length (tokens) / max output tokens per turn --duration-sec emission window - --warmup-sec discard metrics before this (also used as the idle period in - the suite's idle_burst phase) + --warmup-sec discard metrics before this in single-load mode --report-sec live progress print interval (0 disables) - --suite / --quick / --full / --out suite mode controls + --stress-* ramp, hold, and saturation thresholds for stress mode Reading the output ------------------ - - Saturation knee: the QPS step where achieved QPS falls below target, or TTFT - p95 spikes. - - blocked > 0: the session pool filled before the server did -- raise - --num-sessions and rerun that step. - - idle_burst fail>0: requests rejected right after an idle window (health - misclassification / spurious 503). + - SATURATED: achieved QPS fell below the configured ratio, requests failed, + or all sessions were busy often enough to block new turns. + - blocked > 0: the session pool is full. Raise --num-sessions if the client, + rather than the server, is limiting the offered load. - error breakdown: top error strings tell you whether failures are timeouts, connection resets, or empty streams. """ @@ -56,8 +52,8 @@ import argparse import asyncio import contextlib -import io import math +import sys import time from pathlib import Path from typing import Any, Dict, List, Optional @@ -166,6 +162,7 @@ async def do_turn( or extra.get("reasoning_content") or extra.get("text") or getattr(delta, "reasoning", None) + or getattr(delta, "reasoning_content", None) ) if text: reply_parts.append(text) @@ -185,26 +182,31 @@ async def do_turn( # append the real generated reply so the next turn's prefix grows # like a real multi-turn agent (full history re-sent) sess.history.append({"role": "assistant", "content": "".join(reply_parts)}) - if time.perf_counter() >= warmup_until: - metrics.append( - { - "ttft": (first_t - t0) * 1000, - "itl": (itl_sum / itl_n * 1000) if itl_n else None, - "tpot": (((last_t - first_t) / (ntok - 1)) * 1000) - if (ntok > 1 and first_t and last_t) - else 0.0, - "lat": (end - t0) * 1000, - "turn_idx": sess.turns_done, - "sid": sess.sid, - } - ) + metrics.append( + { + "ttft": (first_t - t0) * 1000, + "itl": (itl_sum / itl_n * 1000) if itl_n else None, + "tpot": (((last_t - first_t) / (ntok - 1)) * 1000) if (ntok > 1 and first_t and last_t) else 0.0, + "lat": (end - t0) * 1000, + "turn_idx": sess.turns_done, + "sid": sess.sid, + "measured": time.perf_counter() >= warmup_until, + } + ) except Exception as e: # keep the run alive; record failure count only err = str(e) finally: sess.turns_done += 1 sess.pending = False if err: - metrics.append({"error": err, "turn_idx": sess.turns_done, "sid": sess.sid}) + metrics.append( + { + "error": err, + "turn_idx": sess.turns_done, + "sid": sess.sid, + "measured": time.perf_counter() >= warmup_until, + } + ) async def run(args): @@ -220,7 +222,7 @@ async def run(args): warmup_until = time.perf_counter() + args.warmup_sec print( - f"agent bench: qps={args.qps} sessions={args.num_sessions} turns/session={args.turns_per_session} " + f"agent stress load: qps={args.qps} sessions={args.num_sessions} turns/session={args.turns_per_session} " f"isl={args.isl} osl={args.osl} duration={args.duration_sec}s warmup={args.warmup_sec}s" ) @@ -306,12 +308,20 @@ async def fire_one(): def summarize(metrics, fired, blocked, total_time, args): - ok = [m for m in metrics if "error" not in m] - fail = [m for m in metrics if "error" in m] + all_ok = [m for m in metrics if "error" not in m] + all_fail = [m for m in metrics if "error" in m] + ok = [m for m in all_ok if m["measured"]] + fail = [m for m in all_fail if m["measured"]] ttfts = [m["ttft"] for m in ok] itls = [m["itl"] for m in ok if m["itl"] is not None] tpots = [m["tpot"] for m in ok if m["tpot"] > 0] lats = [m["lat"] for m in ok] + completed = len(all_ok) + len(all_fail) + unfinished = max(0, fired - completed) + attempted = fired + blocked + failure_rate = (len(all_fail) + unfinished) / fired if fired else 0.0 + blocked_rate = blocked / attempted if attempted else 0.0 + achieved_qps = len(all_ok) / args.duration_sec if args.duration_sec > 0 else 0.0 # break TTFT down by turn index to show prefix-cache benefit (turn 0 vs later) by_turn: Dict[int, List[float]] = {} @@ -325,19 +335,20 @@ def line(name, s): ) print("\n" + "=" * 96) - print("AGENT OPEN-LOOP BENCHMARK RESULTS") + print("AGENT OPEN-LOOP STRESS LOAD RESULTS") print("=" * 96) print(f" target QPS : {args.qps}") print(f" fired turns : {fired} blocked ticks (saturation): {blocked}") - print(f" completed (ok/fail): {len(ok)} / {len(fail)}") - if fail: + print(f" completed (ok/fail): {len(all_ok)} / {len(all_fail)} unfinished after drain: {unfinished}") + print(f" failure / blocked rate: {failure_rate:.2%} / {blocked_rate:.2%}") + if all_fail: from collections import Counter - c = Counter(m["error"][:160] for m in fail) + c = Counter(m["error"][:160] for m in all_fail) print(" error breakdown:") for msg, n in c.most_common(5): print(f" [{n:>3}] {msg}") - print(f" achieved turn QPS : {len(ok) / total_time:.2f} /s (wall {total_time:.1f}s)") + print(f" achieved turn QPS : {achieved_qps:.2f} /s (wall incl. drain {total_time:.1f}s)") print("-" * 96) line("TTFT", stats(ttfts)) line("ITL", stats(itls)) @@ -355,14 +366,17 @@ def line(name, s): from collections import Counter - top_err = Counter(m["error"][:120] for m in fail).most_common(1) + top_err = Counter(m["error"][:120] for m in all_fail).most_common(1) return { "target_qps": args.qps, "fired": fired, "blocked": blocked, - "ok": len(ok), - "fail": len(fail), - "achieved_qps": len(ok) / total_time if total_time > 0 else 0.0, + "blocked_rate": blocked_rate, + "ok": len(all_ok), + "fail": len(all_fail), + "failure_rate": failure_rate, + "unfinished": unfinished, + "achieved_qps": achieved_qps, "wall": total_time, "ttft": stats(ttfts), "itl": stats(itls), @@ -372,25 +386,40 @@ def line(name, s): } -def _run_phase(name: str, out_dir, base: dict, **kw) -> dict: - args = make_args(**{**base, **kw, "report_sec": 5}) +class _Tee: + """Write phase output to the terminal and its log at the same time.""" + + def __init__(self, *streams): + self.streams = streams + + def write(self, data): + for stream in self.streams: + stream.write(data) + return len(data) + + def flush(self): + for stream in self.streams: + stream.flush() + + +def _run_phase(name: str, out_dir: Path, base: dict, **kw) -> dict: + args = make_args(**{**base, **kw}) print( f"\n>>> {name} qps={args.qps} sessions={args.num_sessions} " f"isl={args.isl} osl={args.osl} dur={args.duration_sec}s warmup={args.warmup_sec}s", flush=True, ) - buf = io.StringIO() t0 = time.perf_counter() - with contextlib.redirect_stdout(buf): - res = asyncio.run(run(args)) + with Path(out_dir, f"{name}.log").open("w", encoding="utf-8") as log_file: + with contextlib.redirect_stdout(_Tee(sys.stdout, log_file)): + res = asyncio.run(run(args)) res["phase"] = name - res["suite_wall"] = time.perf_counter() - t0 - Path(out_dir, f"{name}.log").write_text(buf.getvalue()) + res["phase_wall"] = time.perf_counter() - t0 ttft, tpot = res["ttft"], res["tpot"] err = f" err={res['fail']}({res['top_error'][:40]})" if res["fail"] else "" print( - f" done in {res['suite_wall']:4.0f}s | ok={res['ok']:<4} fail={res['fail']:<3} " - f"blocked={res['blocked']:<3} achQPS={res['achieved_qps']:.2f} " + f" done in {res['phase_wall']:4.0f}s | ok={res['ok']:<4} fail={res['fail']:<3} " + f"unfinished={res['unfinished']:<3} blocked={res['blocked']:<3} achQPS={res['achieved_qps']:.2f} " f"TTFT p50={ttft['p50']:.0f} p95={ttft['p95']:.0f} " f"TPOT p50={tpot['p50']:.0f} p95={tpot['p95']:.0f}{err}", flush=True, @@ -400,87 +429,229 @@ def _run_phase(name: str, out_dir, base: dict, **kw) -> dict: def _md_table(rows: list) -> str: head = ( - "| phase | targetQPS | fired | ok | fail | blocked | achQPS | " - "TTFT p50 | TTFT p95 | TPOT p50 | TPOT p95 | top error |\n" + "| phase | status | targetQPS | fired | ok | fail | unfinished | blocked% | achQPS | " + "TTFT p50 | TTFT p95 | reason |\n" "|---|---|---|---|---|---|---|---|---|---|---|---|" ) body = [] for r in rows: - t, p = r["ttft"], r["tpot"] + reason = "; ".join(r["saturation_reasons"]) or "-" + reason = reason.replace("|", "\\|") body.append( - f"| {r['phase']} | {r['target_qps']} | {r['fired']} | {r['ok']} | {r['fail']} | " - f"{r['blocked']} | {r['achieved_qps']:.2f} | " - f"{t['p50']:.0f} | {t['p95']:.0f} | {p['p50']:.0f} | {p['p95']:.0f} | " - f"{(r['top_error'] or '')[:50]} |" + f"| {r['phase']} | {r['status']} | {r['target_qps']:g} | {r['fired']} | {r['ok']} | " + f"{r['fail']} | {r['unfinished']} | {r['blocked_rate']:.1%} | {r['achieved_qps']:.2f} | " + f"{r['ttft']['p50']:.0f} | {r['ttft']['p95']:.0f} | {reason} |" ) return head + "\n" + "\n".join(body) -def run_suite(args) -> None: - import os - - d = 0.5 if args.quick else 1.0 - out_dir = args.out - os.makedirs(out_dir, exist_ok=True) - base = {"base_url": args.base_url, "model": args.model, "isl": args.isl, "osl": args.osl} - - def dur(s): - return max(10.0, s * d) - - phases = [ - ("smoke", dict(qps=1, num_sessions=2, turns_per_session=3, duration_sec=dur(20), warmup_sec=dur(5))), - ("perf_q2", dict(qps=2, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), - ("perf_q5", dict(qps=5, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), - ("perf_q10", dict(qps=10, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), - ("perf_q20", dict(qps=20, num_sessions=64, duration_sec=dur(60), warmup_sec=dur(15))), - ("overload_q40", dict(qps=40, num_sessions=128, duration_sec=dur(60), warmup_sec=0)), - ("idle_burst", dict(qps=20, num_sessions=64, duration_sec=dur(90), warmup_sec=dur(45))), - ] - if args.full: - phases.append( - ("longrun", dict(qps=8, num_sessions=64, duration_sec=dur(600), warmup_sec=dur(30), report_sec=30)) - ) +def _saturation_reasons(result: dict, args) -> List[str]: + reasons = [] + ratio = result["achieved_qps"] / result["target_qps"] if result["target_qps"] else 0.0 + if ratio < args.stress_min_achieved_ratio: + reasons.append(f"achieved/target {ratio:.1%} < {args.stress_min_achieved_ratio:.1%}") + if result["failure_rate"] > args.stress_max_error_rate: + reasons.append(f"failure rate {result['failure_rate']:.1%} > {args.stress_max_error_rate:.1%}") + if result["blocked_rate"] > args.stress_max_blocked_rate: + reasons.append(f"blocked rate {result['blocked_rate']:.1%} > {args.stress_max_blocked_rate:.1%}") + return reasons + + +def _qps_label(qps: float) -> str: + return f"{qps:g}".replace(".", "_") + + +def _stress_qps_steps(start: float, step: float, maximum: float) -> List[float]: + values = [] + current = start + while current < maximum: + values.append(current) + current += step + if not values or not math.isclose(values[-1], maximum): + values.append(maximum) + return values + + +def run_stress(args) -> None: + scale = 0.25 if args.quick else 1.0 + step_duration = max(5.0, args.stress_step_sec * scale) + hold_duration = max(10.0, args.stress_hold_sec * scale) + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + base = { + "api_key": args.api_key, + "base_url": args.base_url, + "model": args.model, + "num_sessions": args.num_sessions, + "turns_per_session": args.turns_per_session, + "isl": args.isl, + "osl": args.osl, + "warmup_sec": 0, + "drain_sec": args.drain_sec, + "report_sec": args.report_sec, + } print( - f"SUITE -> {args.base_url} model={args.model} isl={args.isl} osl={args.osl} " - f"quick={args.quick} full={args.full} out={out_dir}", + f"STRESS TEST -> {args.base_url} model={args.model} isl={args.isl} osl={args.osl} " + f"qps={args.stress_start_qps:g}..{args.stress_max_qps:g} step={args.stress_step_qps:g} " + f"step_sec={step_duration:g} hold_sec={hold_duration:g} sessions={args.num_sessions} " + f"quick={args.quick} out={out_dir}", flush=True, ) t_start = time.perf_counter() - results = [_run_phase(name, out_dir, base, **kw) for name, kw in phases] + results = [] + hold_qps = args.stress_max_qps + first_saturated_qps = None + + for qps in _stress_qps_steps(args.stress_start_qps, args.stress_step_qps, args.stress_max_qps): + result = _run_phase( + f"ramp_qps_{_qps_label(qps)}", + out_dir, + base, + qps=qps, + duration_sec=step_duration, + ) + result["saturation_reasons"] = _saturation_reasons(result, args) + result["status"] = "SATURATED" if result["saturation_reasons"] else "PASS" + results.append(result) + if result["saturation_reasons"]: + first_saturated_qps = qps + hold_qps = qps + print( + f" saturation detected at {qps:g} QPS: {'; '.join(result['saturation_reasons'])}", + flush=True, + ) + break + + hold = _run_phase( + f"hold_qps_{_qps_label(hold_qps)}", + out_dir, + base, + qps=hold_qps, + duration_sec=hold_duration, + ) + hold["saturation_reasons"] = _saturation_reasons(hold, args) + hold["status"] = "SATURATED" if hold["saturation_reasons"] else "PASS" + results.append(hold) total = time.perf_counter() - t_start table = _md_table(results) print("\n" + "=" * 96) - print(f"SUITE DONE in {total:.0f}s ({len(results)} phases)") + saturation = f"first saturation: {first_saturated_qps:g} QPS" if first_saturated_qps else "no ramp saturation" + print(f"STRESS TEST DONE in {total:.0f}s ({len(results)} phases, {saturation})") print("=" * 96) print(table) Path(out_dir, "report.md").write_text( - f"# Bench suite report\n\nTarget: `{args.base_url}` model=`{args.model}` " - f"isl={args.isl} osl={args.osl} quick={args.quick} full={args.full}\n\n" - f"Total wall: {total:.0f}s\n\n{table}\n" + f"# Agent stress test report\n\nTarget: `{args.base_url}` model=`{args.model}` " + f"isl={args.isl} osl={args.osl} sessions={args.num_sessions} quick={args.quick}\n\n" + f"Result: **{saturation}**. Total wall: {total:.0f}s.\n\n{table}\n", + encoding="utf-8", ) print(f"\nreport -> {Path(out_dir, 'report.md')} (per-phase logs: {out_dir}/*.log)") def build_parser(): - p = argparse.ArgumentParser(description="Agent multi-turn open-loop benchmark") + p = argparse.ArgumentParser(description="Agent multi-turn open-loop stress test") p.add_argument("--api-key", default="EMPTY") p.add_argument("--base-url", default="http://127.0.0.1:8000/v1") p.add_argument("--model", default="qwen35") p.add_argument("--qps", type=float, default=5.0, help="turn emission rate (open loop)") - p.add_argument("--num-sessions", type=int, default=32, help="concurrent sessions; distinct prefix each") - p.add_argument("--turns-per-session", type=int, default=6, help="turns before a session is recycled") - p.add_argument("--isl", type=int, default=2048, help="system-prompt length (tokens) per session") + p.add_argument( + "--num-sessions", + type=int, + default=128, + help="concurrent sessions; distinct prefix each", + ) + p.add_argument( + "--turns-per-session", + type=int, + default=6, + help="turns before a session is recycled", + ) + p.add_argument( + "--isl", + type=int, + default=2048, + help="system-prompt length (tokens) per session", + ) p.add_argument("--osl", type=int, default=256, help="max output tokens per turn") p.add_argument("--duration-sec", type=float, default=60.0, help="emission window") p.add_argument("--warmup-sec", type=float, default=10.0, help="discard metrics before this") - p.add_argument("--drain-sec", type=float, default=120.0, help="max wait for in-flight turns after window") - p.add_argument("--report-sec", type=float, default=5.0, help="live progress print interval (0 disables)") - p.add_argument("--suite", action="store_true", help="run the full robustness+perf suite instead of one bench") - p.add_argument("--quick", action="store_true", help="(suite) halve all durations") - p.add_argument("--full", action="store_true", help="(suite) add a 10-min long-run stability phase") - p.add_argument("--out", default="bench_suite_out", help="(suite) dir for per-phase logs + report.md") + p.add_argument( + "--drain-sec", + type=float, + default=120.0, + help="max wait for in-flight turns after window", + ) + p.add_argument( + "--report-sec", + type=float, + default=5.0, + help="live progress print interval (0 disables)", + ) + p.add_argument( + "--stress", + action="store_true", + help="ramp QPS to saturation, then hold the pressure", + ) + p.add_argument( + "--stress-start-qps", + type=float, + default=5.0, + help="first QPS in the stress ramp", + ) + p.add_argument( + "--stress-step-qps", + type=float, + default=5.0, + help="QPS added after each passing ramp phase", + ) + p.add_argument( + "--stress-max-qps", + type=float, + default=40.0, + help="highest QPS offered by the stress ramp", + ) + p.add_argument( + "--stress-step-sec", + type=float, + default=60.0, + help="duration of each ramp phase", + ) + p.add_argument( + "--stress-hold-sec", + type=float, + default=300.0, + help="duration of the final pressure hold", + ) + p.add_argument( + "--stress-max-error-rate", + type=float, + default=0.01, + help="stop ramp above this failed/unfinished rate", + ) + p.add_argument( + "--stress-max-blocked-rate", + type=float, + default=0.01, + help="stop ramp above this blocked-tick rate", + ) + p.add_argument( + "--stress-min-achieved-ratio", + type=float, + default=0.90, + help="stop ramp when achieved QPS / target QPS falls below this ratio", + ) + p.add_argument( + "--quick", + action="store_true", + help="(stress) run phases at one quarter duration", + ) + p.add_argument( + "--out", + default="stress_test_out", + help="(stress) dir for per-phase logs + report.md", + ) return p @@ -491,10 +662,42 @@ def make_args(**overrides): return args +def validate_args(parser, args) -> None: + positive = { + "qps": args.qps, + "num-sessions": args.num_sessions, + "turns-per-session": args.turns_per_session, + "isl": args.isl, + "osl": args.osl, + "duration-sec": args.duration_sec, + } + for name, value in positive.items(): + if value <= 0: + parser.error(f"--{name} must be greater than zero") + if args.warmup_sec < 0 or args.drain_sec < 0 or args.report_sec < 0: + parser.error("--warmup-sec, --drain-sec, and --report-sec cannot be negative") + if args.stress: + if args.stress_start_qps <= 0 or args.stress_step_qps <= 0 or args.stress_max_qps <= 0: + parser.error("stress QPS values must be greater than zero") + if args.stress_start_qps > args.stress_max_qps: + parser.error("--stress-start-qps cannot exceed --stress-max-qps") + if args.stress_step_sec <= 0 or args.stress_hold_sec <= 0: + parser.error("stress phase durations must be greater than zero") + for name in ( + "stress_max_error_rate", + "stress_max_blocked_rate", + "stress_min_achieved_ratio", + ): + if not 0 <= getattr(args, name) <= 1: + parser.error(f"--{name.replace('_', '-')} must be between 0 and 1") + + def main(): - args = build_parser().parse_args() - if args.suite: - run_suite(args) + parser = build_parser() + args = parser.parse_args() + validate_args(parser, args) + if args.stress: + run_stress(args) else: asyncio.run(run(args)) From 202b2bcb58f2aed5fa147030e1cd7f406b6e974b Mon Sep 17 00:00:00 2001 From: sufubao Date: Thu, 6 Aug 2026 19:45:10 +0800 Subject: [PATCH 5/5] test/benchmark: fix stress summary lint --- test/benchmark/bench_agents.py | 1 - 1 file changed, 1 deletion(-) diff --git a/test/benchmark/bench_agents.py b/test/benchmark/bench_agents.py index 3d197025a..80c336d92 100644 --- a/test/benchmark/bench_agents.py +++ b/test/benchmark/bench_agents.py @@ -311,7 +311,6 @@ def summarize(metrics, fired, blocked, total_time, args): all_ok = [m for m in metrics if "error" not in m] all_fail = [m for m in metrics if "error" in m] ok = [m for m in all_ok if m["measured"]] - fail = [m for m in all_fail if m["measured"]] ttfts = [m["ttft"] for m in ok] itls = [m["itl"] for m in ok if m["itl"] is not None] tpots = [m["tpot"] for m in ok if m["tpot"] > 0]