diff --git a/README.md b/README.md index 2b8eeb5..5efaebc 100644 --- a/README.md +++ b/README.md @@ -161,6 +161,7 @@ rep.save('gsm8k.report.json') | `--limit N` / `--limit-per-task N` | 全局前 N / 每子集前 N（多科目 bench 用后者；可组合取交集） | | `--subset` `--split` `--source` | 覆盖子集 / split / 数据源（可指本地路径离线跑） | | `--concurrency N` | 并发（默认 32；长输出 bench 建议 8-16） | +| `--auto-concurrency` | 自适应并发门：按端点健康状况自动决定并发（健康且供不应求时 +1 爬坡，请求失败 ×0.7 退避，服务端 `/metrics` 可用时按排队信号调节）；当前值显示在进度条 `gate N`。此时 `--concurrency` 是起点不是上限 | | `--resume [PATH]` | 断点续跑；默认 `/ckpt/.jsonl` | | `--env NAME` | agent 环境（`bfcl_mock` 等） | | `--perf` | 采集流式 TTFT / ITL / 重试率入报告 | diff --git a/evalharness/cli.py b/evalharness/cli.py index 7084092..1877e9a 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -400,6 +400,12 @@ def _compose_model_spec(args): if not model: raise SystemExit('error: --api-url requires --model (model name)') model = f'{internal_provider}/{api_url.rstrip("/")}?{model}' + # --auto-concurrency: the per-endpoint adaptive gate decides how many + # requests fly; it lives in the pool layer, so wrap even a single + # endpoint as a one-member pool (round-robin of 1 = identity) + if getattr(args, 'auto_concurrency', False) and api_url \ + and internal_provider == 'openai': + model = f'openai-pool/{api_url.rstrip("/")}?{model.split("?", 1)[1]}' return _model_with_flags(model, args) @@ -618,6 +624,12 @@ def _cmd_eval_run(args) -> int: run_started = _time.time() total_runs = len(args.datasets) model_spec = _compose_model_spec(args) + if getattr(args, 'auto_concurrency', False) and model_spec: + # the gate starts where the user's manual concurrency was, then + # adapts on its own signals (see pool.AdaptiveGate) + from evalharness.model.pool import AdaptiveGate + + AdaptiveGate.INITIAL = float(max(2, getattr(args, 'concurrency', 8))) if not out_dir and model_spec and not args.out: # always persist results: default dir = evalharness-results/-/ import re as _re @@ -744,7 +756,10 @@ def _cmd_eval_run(args) -> int: if _repeats > 1: print(f'\n[repeat {_rep + 1}/{_repeats}]', flush=True) report = asyncio.run(run_eval( - ds, model_spec, concurrency=args.concurrency, + ds, model_spec, + concurrency=max(args.concurrency, 96) + if getattr(args, 'auto_concurrency', False) + else args.concurrency, limit=args.limit, limit_per_task=args.limit_per_task, gen_kwargs=_gen_kw or None, max_input_tokens=_mit, @@ -1067,6 +1082,12 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument('--provider', default='openai-chat', choices=('openai-chat', 'openai-pool'), help='API protocol/provider (default: openai-chat)') + p.add_argument('--auto-concurrency', action='store_true', + help='let the per-endpoint adaptive gate decide concurrency ' + '(ramps while healthy, backs off x0.7 on failures, ' + 'server /metrics when available); current limit shows ' + 'on the progress bar as "gate N". --concurrency ' + 'becomes the starting point, not a cap') p.add_argument('--judge-model', '--judge', dest='judge', default='', help='judge model name with --judge-api-url, or full spec') p.add_argument('--judge-api-url', default='', diff --git a/evalharness/model/pool.py b/evalharness/model/pool.py index 1a5f6d5..6ce27cb 100644 --- a/evalharness/model/pool.py +++ b/evalharness/model/pool.py @@ -154,17 +154,30 @@ class AdaptiveGate: LO = 2 # never go below: progress beats perfection HI = 96 # sane ceiling for one endpoint PROBE_S = 5.0 # metrics probe interval + INITIAL = 8.0 # class-level start point (--auto-concurrency rebinds it) def __init__(self, adapter: ModelAdapter): self.adapter = adapter - self.limit = 8.0 # float for smooth x0.7; compare with int() + self.limit = self.INITIAL # float for smooth x0.7; compare with int() self._inflight = 0 self._cond: Optional[asyncio.Condition] = None self._task: Optional[asyncio.Task] = None self._stopped = False self._loop = None # loop the cond/probe-task are bound to + self._contended = 0 # acquire() waits this probe interval (demand) + self._interval_fails = 0 # failed releases this probe interval self.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0, - 'backoff_queue': 0} + 'backoff_queue': 0, 'ramp_demand': 0} + + def _push_limit(self) -> None: + """Surface the current limit to the progress bar ('gate N').""" + rep = (self.adapter.extra or {}).get('progress_reporter') + fn = getattr(rep, 'set_gate', None) + if fn is not None: + try: + fn(max(1, int(self.limit))) + except Exception: + pass # ---- gate semantics ---- async def acquire(self) -> None: @@ -181,10 +194,12 @@ class AdaptiveGate: self._cond = asyncio.Condition() self._inflight = 0 self._stopped = False + self._push_limit() if self._task is not None: self._task.cancel() # dead task from the closed loop; no-op self._task = loop.create_task(self._probe_loop()) while self._inflight >= max(1, int(self.limit)): + self._contended += 1 # demand exceeded the cap: potential ramp fuel await self._cond.acquire() try: await self._cond.wait() @@ -195,10 +210,12 @@ class AdaptiveGate: def release(self, ok: bool) -> None: self._inflight = max(0, self._inflight - 1) if not ok: # multiplicative decrease -- survival first + self._interval_fails += 1 before = self.limit self.limit = max(self.LO, self.limit * 0.7) if before != self.limit: self.stats['backoff_fail'] += 1 + self._push_limit() self._wake() def _wake(self) -> None: @@ -215,15 +232,37 @@ class AdaptiveGate: self._cond.notify_all() # ---- server-signal probe ---- + def _no_signal_ramp(self) -> None: + """No server signals available (no /metrics, 404/HTTPError, gateway + stripped it, non-sglang backend): fall back to demand-driven AIMD -- + ramp while the cap is the binding constraint (callers had to WAIT on + acquire) and the interval was failure-free. Failures still cut x0.7 + per release, so a drowning backend shrinks the gate immediately.""" + if self._interval_fails == 0 and self._contended > 0 \ + and int(self.limit) < self.HI: + self.limit = min(self.HI, self.limit + 1) + self.stats['ramp_demand'] += 1 + self._push_limit() + self._interval_fails = 0 + self._contended = 0 + self._wake() + async def _probe_once(self) -> None: import urllib.request url = f'{self.adapter.api_base.rstrip("/")}/metrics' - try: + + def _fetch(): with urllib.request.urlopen(url, timeout=4) as resp: - text = resp.read().decode('utf-8', 'ignore') + return resp.read().decode('utf-8', 'ignore') + + try: + # thread: the blocking fetch must never stall the event loop + # (an unreachable host parks urlopen for the full 4s timeout) + text = await asyncio.to_thread(_fetch) except Exception: - return # no metrics (or busy): hold current limit + self._no_signal_ramp() # no metrics: demand-driven fallback + return running = queue = None for line in text.splitlines(): if line.startswith('sglang:num_running_reqs'): @@ -232,17 +271,22 @@ class AdaptiveGate: queue = float(line.rsplit(' ', 1)[-1]) self.stats['probe'] += 1 if queue is None and running is None: + self._no_signal_ramp() return if queue is not None and queue >= 2: # server is queuing OUR excess: gentle additive decrease self.limit = max(self.LO, self.limit - 1) self.stats['backoff_queue'] += 1 + self._push_limit() elif (queue or 0) == 0 and (running is None or running < max(2, int(self.limit))): # underfed: no queue and running below our own cap -> ramp up self.limit = min(self.HI, self.limit + 1) self.stats['ramp'] += 1 + self._push_limit() else: self.stats['hold_queue'] += 1 + self._interval_fails = 0 + self._contended = 0 self._wake() async def _probe_loop(self) -> None: diff --git a/evalharness/progress/rich_terminal.py b/evalharness/progress/rich_terminal.py index ee06ead..199db6e 100644 --- a/evalharness/progress/rich_terminal.py +++ b/evalharness/progress/rich_terminal.py @@ -46,6 +46,7 @@ class RichTerminalProgress: TextColumn("• Completed {task.completed}/{task.total} [dim]{task.fields[new]}[/dim]"), TextColumn("• in-flight [yellow]{task.fields[inflight]}[/yellow] ([yellow]{task.fields[cur]}[/yellow])"), TextColumn("• [red]retries {task.fields[retries]}[/red]"), + TextColumn("[magenta]{task.fields[gate]}[/magenta]"), TextColumn("• [dim]{task.fields[rate]}/s[/dim]"), TextColumn("• [green]{task.fields[elapsed]}[/green]"), TextColumn("• [cyan]eta {task.fields[eta]}[/cyan]"), @@ -102,6 +103,7 @@ class RichTerminalProgress: inflight=0, cur="0s", retries=0, + gate='', elapsed="0s", eta="-", ) @@ -130,13 +132,13 @@ class RichTerminalProgress: self.task_id = self.progress.add_task( desc, total=total, completed=min(completed, total), new=self._new_txt(completed), rate='0.00', inflight=0, - cur='0s', elapsed='0s', eta='-', retries=0) + cur='0s', elapsed='0s', eta='-', retries=0, gate='') else: self.progress.update(self.task_id, description=desc, total=total, completed=min(completed, total), new=self._new_txt(completed), rate='0.00', inflight=0, cur='0s', elapsed='0s', eta='-', - retries=0) + retries=0, gate='') # ALWAYS recreate the heartbeat: the previous one may have died during # pause/resume cycles between benchmarks (stale reference -> silent # death -> frozen clock while the spinner still animates) @@ -200,6 +202,11 @@ class RichTerminalProgress: if self.task_id is not None: self.progress.update(self.task_id, retries=n) + def set_gate(self, n: int): + """Show the adaptive concurrency gate's current limit ('gate 12').""" + if self.task_id is not None: + self.progress.update(self.task_id, gate=f'gate {n}') + def rollback(self): if self.disabled: return