From 2ed452a96f727ea8187fda08884131bc25f6bfe5 Mon Sep 17 00:00:00 2001 From: xxxyxun Date: Thu, 29 Oct 2026 21:59:35 +0800 Subject: [PATCH] feat: add priority API key pools and worker key selection --- README.md | 29 ++++- infra/config.example.json | 30 +++-- src/tandem/config.py | 6 +- src/tandem/models/key_pool.py | 211 +++++++++++++++++++++++++++++++ src/tandem/models/slow.py | 149 +++++++++++++++------- src/tandem_eval/__main__.py | 14 +- src/tandem_eval/config_schema.py | 28 +++- src/tandem_eval/rjob_spec.py | 18 ++- tests/eval/test_config.py | 4 +- tests/harness/test_session.py | 5 +- 10 files changed, 418 insertions(+), 76 deletions(-) create mode 100644 src/tandem/models/key_pool.py diff --git a/README.md b/README.md index c259da7..aa4e384 100644 --- a/README.md +++ b/README.md @@ -216,21 +216,38 @@ tandem-eval rjob run --run-id --benchmark osworld2 --tasks \ | `slow_max_tokens` | `16384` | 慢模型单次响应上限 | | `slow_timeout_seconds` | `120` | 慢模型超时 | | `slow_direct_api` | `False` | 直连 API（不经代理） | -| `slow_reasoning_effort` | `"low"` | `off`（归零思考）/ `low` / `medium` / `high` | +| `slow_reasoning_effort` | `None` | 服务端默认，或 `off` / `low` / `medium` / `high` / `max` | | `fast_backend` | `"jev"` | `jev`（`/rank`，需截图）/ `kev`（System One，纯文本） | **换慢模型厂商**：在 `infra/config.json` 的 `harness.slow_model.apis` 里登记多套 API， 启动时用 `--slow-model <名字>` 挑（缺省取 `default`），不改代码。 -> 厂商方言有差异（2026-10-23 实测）：Kimi 认 `enable_thinking=false` 关思考； -> **MiniMax 忽略该字段**（仍返回 `reasoning_content`），只有 `reasoning_effort` 生效 -> （`low`/`medium`/`high`）——想少思考就把 `reasoning_effort` 调低，别指望 `off`。 +集群内使用模板配置时，`api_key` 留空会使用默认号池，也可以手动填写自己的 `api_key`。 + +需要指定号池中的 key 时，在原启动命令后追加： + +```bash +--slow-key-ids 1,2,3,4 +``` + +也可以在所选慢模型条目的 `key_pool` 中配置，命令行参数优先： + +```json +{"key_ids": [1, 2, 3, 4]} +``` + +编号对应凭据列表中的第 1、2、3、4 把 key，不是密钥内容： + +- 不填编号或配置 `[]`：使用默认优先级。 +- 填一个编号：只使用这一把。 +- 填多个编号：这些 key 按同级轮询分配给 worker，正常请求固定使用分配到的 key；故障仅在指定集合内切换。 + +不同实验可指定不重叠的编号集合。这里均分的是 worker，不保证请求量严格相等；编号选择与手填完整 `api_key` 二选一。 慢模型的调用纪律（`models/slow.py`）： -- 思考强度默认 `low`（实测 97 → 22 token）；空正文（全留在思考通道）→ 下一次重试自动升档； +- 思考强度取所选 API 的配置；空正文（全留在思考通道）→ 下一次重试自动升档； - 叙述文字包裹 JSON → 自动剥围栏取首个平衡 `{…}` 抢救； -- 403（TPM 配额限流）→ 指数退避 60/120/240/480s，总计 5 次尝试； - `finish_reason=length` → 明确报错，提示调大 `slow_max_tokens`。 ## 排序后端 diff --git a/infra/config.example.json b/infra/config.example.json index fc7784f..a652093 100644 --- a/infra/config.example.json +++ b/infra/config.example.json @@ -4,22 +4,34 @@ "default": "minimax-m3.1-flash", "apis": { "minimax-m3.1-flash": { - "base_url": "https://api.minimaxi.com/v1", - "api_key": "sk-xxx", + "base_url": "https://api.minimax.cn/v1", + "api_key": "", "model": "MiniMax-M3.1-Flash", + "key_pool": { + "provider": "minimax", + "key_ids": [], + "credentials_file": "/mnt/shared-storage-gpfs2/evobox-share-gpfs2/xuyuanxun/secrets/tandem_api_pool/credentials.json", + "state_file": "/mnt/shared-storage-gpfs2/evobox-share-gpfs2/xuyuanxun/secrets/tandem_api_pool/state.json" + }, "max_tokens": 16384, "timeout_seconds": 180, - "reasoning_effort": "low", - "direct_api": false + "reasoning_effort": "max", + "direct_api": true }, "kimi-k3": { - "base_url": "https://your-kimi-endpoint/v1", - "api_key": "sk-xxx", - "model": "kimi-k3", + "base_url": "https://api.kimi.com/coding/v1", + "api_key": "", + "model": "k3.1-pro", + "key_pool": { + "provider": "kimi", + "key_ids": [], + "credentials_file": "/mnt/shared-storage-gpfs2/evobox-share-gpfs2/xuyuanxun/secrets/tandem_api_pool/credentials.json", + "state_file": "/mnt/shared-storage-gpfs2/evobox-share-gpfs2/xuyuanxun/secrets/tandem_api_pool/state.json" + }, "max_tokens": 16384, "timeout_seconds": 120, - "reasoning_effort": "low", - "direct_api": false + "reasoning_effort": null, + "direct_api": true } } }, diff --git a/src/tandem/config.py b/src/tandem/config.py index 7a43bb6..5786acc 100644 --- a/src/tandem/config.py +++ b/src/tandem/config.py @@ -20,7 +20,7 @@ class SessionConfig: slow_max_tokens: int = 16384 slow_timeout_seconds: float = 120 slow_direct_api: bool = False - slow_reasoning_effort: str = "low" # off=enable_thinking:false | low/medium/high=reasoning_effort + slow_reasoning_effort: str | None = None # None=服务端默认；off=enable_thinking:false fast_backend: str = "jev" # "jev"（qwen-jev-choice /rank）| "kev"（Kev System One） fast_timeout_seconds: float = 300 @@ -35,5 +35,5 @@ def __post_init__(self): raise ValueError("cli_timeout_seconds must be within 1..120 (协议上限)") if self.fast_backend not in {"jev", "kev"}: raise ValueError("fast_backend must be jev or kev") - if self.slow_reasoning_effort not in {"off", "low", "medium", "high"}: - raise ValueError("slow_reasoning_effort must be off/low/medium/high") + if self.slow_reasoning_effort not in {None, "off", "low", "medium", "high", "max"}: + raise ValueError("slow_reasoning_effort must be null/off/low/medium/high/max") diff --git a/src/tandem/models/key_pool.py b/src/tandem/models/key_pool.py new file mode 100644 index 0000000..b8b1a2e --- /dev/null +++ b/src/tandem/models/key_pool.py @@ -0,0 +1,211 @@ +"""慢模型 API key 的优先级选择与跨 worker 冷却；凭据不进入共享状态。""" + +from __future__ import annotations + +import fcntl +import json +import os +import tempfile +import time +from collections.abc import Iterator +from contextlib import contextmanager +from email.utils import parsedate_to_datetime +from pathlib import Path + +POOL_BASE_URLS = { + "minimax": "https://api.minimax.cn/v1", + "kimi": "https://api.kimi.com/coding/v1", +} + + +class PoolExhausted(RuntimeError): + """当前请求已没有可用的 key。""" + + +def validate_key_ids(key_ids) -> None: + if (not isinstance(key_ids, list) + or any(type(number) is not int or number < 1 for number in key_ids) + or len(set(key_ids)) != len(key_ids)): + raise ValueError("key_ids must be a list of distinct positive integers") + + +def classify_api_error(provider: str, status: int, code, message: str) -> tuple[str, float]: + """返回错误类别与 key 冷却秒数；零冷却的错误不切 key。""" + code = str(code) + if status == 401 or (provider == "minimax" and code in {"1004", "2049"}): + return "invalid_key", 3600 + if provider == "minimax": + if code in {"1008", "2053", "2056", "2067"}: + return "quota_exhausted", 300 + if code in {"1002", "1041", "1045", "2045", "2046", "2047", "2062"}: + return "rate_limited", 60 + if code in {"2022", "2057", "2061", "2068"}: + return "model_permission", 3600 + if code in {"1000", "1001", "1024", "1033", "2064", "2066"}: + return "transient", 0 + if provider == "kimi": + text = message.lower() + if status == 429 and any(part in text for part in ( + "usage limit", "credit balance is insufficient", "extra usage balance", + )): + return "quota_exhausted", 300 + if status == 429 and "engine is currently overloaded" in text: + return "transient", 0 + if status == 403 and any(part in text for part in ( + "subscription does not have access", "current plan supports only", + "unable to verify your membership benefits", + )): + return "model_permission", 3600 + if status == 429: + return "rate_limited", 60 + if status >= 500 or status == 408: + return "transient", 0 + return "fatal", 0 + + +def retry_after_seconds(value: str | None, default: float) -> float: + """Retry-After 支持秒数或 HTTP 日期；缺失时用本地冷却策略。""" + if value is None: + return default + try: + return max(0.0, float(value)) + except ValueError: + try: + return max(0.0, parsedate_to_datetime(value).timestamp() - time.time()) + except (TypeError, ValueError): + return default + + +class ApiKeyPool: + def __init__(self, credentials: dict): + base_url = credentials["base_url"].rstrip("/") + self.provider = next( + (name for name, url in POOL_BASE_URLS.items() if base_url == url), "custom" + ) + self.state_file: Path | None = None + self._local_state: dict = {} + self._preferred_id: str | None = None + self._selection_group = "" + options = credentials.get("key_pool") or {} + key_ids = options.get("key_ids", []) + validate_key_ids(key_ids) + if credentials.get("api_key") and key_ids: + raise ValueError("manual api_key cannot be combined with key_pool.key_ids") + # 手填 key 完全独立；无号池时也保留客户端原有的无鉴权 endpoint 用法。 + if credentials.get("api_key") or not options: + self.keys = [{"id": "manual", "priority": 1, "api_key": credentials.get("api_key", "")}] + self._secrets = [row["api_key"] for row in self.keys if row["api_key"]] + return + provider = options["provider"] + if provider not in POOL_BASE_URLS or POOL_BASE_URLS[provider] != base_url: + raise ValueError("key_pool provider does not match its official base_url") + self.keys = json.loads(Path(options["credentials_file"]).read_text(encoding="utf-8"))[provider] + if not isinstance(self.keys, list) or not self.keys: + raise ValueError("key_pool must contain a nonempty credential list") + for row in self.keys: + if not isinstance(row, dict) or not isinstance(row.get("id"), str) or not row["id"]: + raise ValueError("key_pool entries require a nonempty id") + if type(row.get("priority")) is not int or row["priority"] < 1: + raise ValueError("key_pool priorities must be positive integers") + key = row.get("api_key") + if not isinstance(key, str) or not key or key != key.strip(): + raise ValueError("key_pool contains an empty or whitespace-padded api_key") + if len({row["id"] for row in self.keys}) != len(self.keys): + raise ValueError("key_pool contains duplicate ids") + if len({row["api_key"] for row in self.keys}) != len(self.keys): + raise ValueError("key_pool contains duplicate keys") + self._secrets = [row["api_key"] for row in self.keys] + self.state_file = Path(options["state_file"]) + if not self.state_file.is_absolute() or not Path(options["credentials_file"]).is_absolute(): + raise ValueError("key_pool paths must be absolute") + if key_ids: + if max(key_ids) > len(self.keys): + raise ValueError("key_pool.key_ids contains an out-of-range credential number") + self.keys = [self.keys[number - 1] for number in sorted(key_ids)] + self._selection_group = ",".join(str(number) for number in sorted(key_ids)) + with self._state() as state: + cursors = state.setdefault("worker_cursors", {}) + available = self._available(state) or self.keys + cursor = cursors.get(self._selection_group, 0) + self._preferred_id = available[cursor % len(available)]["id"] + cursors[self._selection_group] = (cursor + 1) % len(available) + + @contextmanager + def _state(self) -> Iterator[dict]: + if self.state_file is None: + yield self._local_state + return + parent = self.state_file.parent + parent.mkdir(mode=0o700, parents=True, exist_ok=True) + owner = parent.stat() + descriptor = os.open(str(self.state_file) + ".lock", os.O_CREAT | os.O_RDWR, 0o600) + with os.fdopen(descriptor, "r+") as lock: + # rjob worker 以 root 运行时，私有状态仍归目录所有者管理。 + if os.geteuid() == 0: + os.fchown(lock.fileno(), owner.st_uid, owner.st_gid) + fcntl.flock(lock, fcntl.LOCK_EX) + state = json.loads(self.state_file.read_text()) if self.state_file.exists() else {} + yield state.setdefault(self.provider, {}) + descriptor, temporary = tempfile.mkstemp(prefix=self.state_file.name + ".", dir=parent) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + if os.geteuid() == 0: + os.fchown(stream.fileno(), owner.st_uid, owner.st_gid) + json.dump(state, stream, ensure_ascii=False) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, self.state_file) + finally: + Path(temporary).unlink(missing_ok=True) + + def _available(self, state: dict, excluded: set[str] | None = None) -> list[dict]: + excluded = excluded or set() + now = time.time() + cooldowns = state.get("cooldowns", {}) + return [row for row in self.keys if row["id"] not in excluded + and cooldowns.get(row["id"], {}).get("until", 0) <= now] + + def select(self, excluded: set[str] | None = None) -> dict: + with self._state() as state: + cooldowns = state.get("cooldowns", {}) + available = self._available(state, excluded) + if not available: + next_retry = min( + (cooldowns.get(row["id"], {}).get("until", 0) for row in self.keys), default=0 + ) + raise PoolExhausted( + f"pool_exhausted: provider={self.provider}, next_retry_at={next_retry}" + ) + if self._preferred_id is not None: + preferred = next((row for row in available if row["id"] == self._preferred_id), None) + if preferred is not None: + return preferred + tier = self.keys + cursor_name = "selected:" + self._selection_group + else: + priority = min(row["priority"] for row in available) + tier = [row for row in self.keys if row["priority"] == priority] + cursor_name = str(priority) + cursors = state.setdefault("cursors", {}) + cursor = cursors.get(cursor_name, 0) + indices = ((cursor + offset) % len(tier) for offset in range(len(tier))) + index = next(index for index in indices if tier[index] in available) + cursors[cursor_name] = (index + 1) % len(tier) + return tier[index] + + def cooldown(self, key_id: str, seconds: float, reason: str) -> None: + with self._state() as state: + cooldowns = state.setdefault("cooldowns", {}) + previous = cooldowns.get(key_id, {}).get("until", 0) + cooldowns[key_id] = {"until": max(previous, time.time() + seconds), "reason": reason} + + def redact(self, value): + if isinstance(value, str): + for secret in self._secrets: + value = value.replace(secret, "[REDACTED]") + return value + if isinstance(value, dict): + return {key: self.redact(item) for key, item in value.items()} + if isinstance(value, list): + return [self.redact(item) for item in value] + return value diff --git a/src/tandem/models/slow.py b/src/tandem/models/slow.py index 5bd2089..29ceab1 100644 --- a/src/tandem/models/slow.py +++ b/src/tandem/models/slow.py @@ -11,6 +11,8 @@ import urllib.request from pathlib import Path +from .key_pool import ApiKeyPool, PoolExhausted, classify_api_error, retry_after_seconds + # 空正文（全留在思考通道）或截断时，下一次重试升档思考强度，避免原样重发重复失败 _REASON_EFFORT_LADDER = {"low": "medium", "medium": "high"} @@ -50,12 +52,24 @@ def extract_json_object(text: str): return None +def _api_failure(status: int, raw: dict): + error = raw.get("error") + base = raw.get("base_resp") or {} + if status < 400 and not error and not base.get("status_code"): + return None + if isinstance(error, dict): + return error.get("code") or error.get("type"), str(error.get("message", "")) + return base.get("status_code"), str(error or base.get("status_msg") or json.dumps(raw)) + + class SlowModelClient: def __init__(self, credential_file, config, audit_path: Path, llm_sink=None): if isinstance(credential_file, dict): self.credentials = dict(credential_file) else: self.credentials = json.loads(Path(credential_file).read_text(encoding="utf-8")) + self.key_pool = ApiKeyPool(self.credentials) + self._call_id = 0 self.config = config self.audit_path = audit_path # llm_sink(role=..., request=..., response=..., **meta)：全量请求/响应落盘钩子， @@ -64,25 +78,28 @@ def __init__(self, credential_file, config, audit_path: Path, llm_sink=None): def json_request(self, messages, validate, *, role): credentials = self.credentials + self._call_id += 1 + call_id = self._call_id payload = { "model": credentials.get("model") or "unknown", "messages": messages, "stream": False, "max_tokens": self.config.slow_max_tokens, } - effort = getattr(self.config, "slow_reasoning_effort", "low") + effort = self.config.slow_reasoning_effort if effort == "off": payload["enable_thinking"] = False - else: + elif effort is not None: payload["reasoning_effort"] = effort + def build_request(): return urllib.request.Request( credentials["base_url"].rstrip("/") + "/chat/completions", data=json.dumps(payload).encode("utf-8"), headers={ **( - {"Authorization": "Bearer " + credentials["api_key"]} - if credentials.get("api_key") + {"Authorization": "Bearer " + key["api_key"]} + if key["api_key"] else {} ), "Content-Type": "application/json", @@ -99,14 +116,18 @@ def audit(event, started, attempt, **details): record = { "event": event, "provider": "slow_model", + "api_provider": self.key_pool.provider, "role": role, - "attempt": attempt + 1, + "call_id": call_id, + "key_id": key["id"], + "key_priority": key["priority"], + "attempt": attempt, "time": time.time(), "elapsed_seconds": round(time.perf_counter() - started, 3), "request_bytes": len(request.data), **details, } - text = json.dumps(record, ensure_ascii=False) + text = json.dumps(self.key_pool.redact(record), ensure_ascii=False) print(text, flush=True) with self.audit_path.open("a", encoding="utf-8") as log: log.write(text + "\n") @@ -124,44 +145,87 @@ def public_headers(headers): "x-trace-id", "x-envoy-upstream-service-time", } - return {key: value for key, value in headers.items() if key.lower() in allowed} - - def redact(text): - key = credentials.get("api_key") - return text.replace(key, "[REDACTED]") if key else text + return {name.lower(): value for name, value in headers.items() if name.lower() in allowed} - for attempt in range(5): + excluded: set[str] = set() + key = self.key_pool.select() + attempt = 0 + failures = 0 + while True: + attempt += 1 started = time.perf_counter() request = build_request() # 每次重试重建：升档思考强度需重新序列化 audit("request_start", started, attempt) try: - with opener(request, timeout=self.config.slow_timeout_seconds) as response: - raw = json.load(response) - if self.llm_sink is not None: - self.llm_sink( - role=role, - request={"payload": payload, "messages": messages}, - response=raw, - attempt=attempt + 1, - ) - audit( - "http_response", - started, - attempt, - http_status=response.status, - headers=public_headers(response.headers), - finish_reason=raw["choices"][0].get("finish_reason"), - usage=raw.get("usage"), + try: + with opener(request, timeout=self.config.slow_timeout_seconds) as response: + status = response.status + headers = public_headers(response.headers) + raw = json.load(response) + except urllib.error.HTTPError as error: + status = error.code + headers = public_headers(error.headers) + body = self.key_pool.redact(error.read().decode("utf-8", errors="replace")) + try: + raw = json.loads(body) + except ValueError: + raw = {"error": {"message": body[:2000]}} + if not isinstance(raw, dict): + raise ValueError("slow model response must be a JSON object") + if self.llm_sink is not None: + self.llm_sink( + role=role, + request=self.key_pool.redact({"payload": payload, "messages": messages}), + response=self.key_pool.redact(raw), + attempt=attempt, call_id=call_id, + key_id=key["id"], key_priority=key["priority"], + http_status=status, + elapsed_seconds=round(time.perf_counter() - started, 3), ) - content = raw["choices"][0]["message"]["content"].strip() - finish_reason = raw["choices"][0].get("finish_reason") + failure = _api_failure(status, raw) + if failure is not None: + code, message = failure + category, cooldown = classify_api_error(self.key_pool.provider, status, code, message) + audit("api_error", started, attempt, http_status=status, headers=headers, + error_code=code, category=category, error=message[:2000], usage=raw.get("usage")) + error = RuntimeError(self.key_pool.redact( + f"slow model API error: status={status}, code={code}, message={message[:2000]}" + )) + if cooldown: + cooldown = retry_after_seconds(headers.get("retry-after"), cooldown) + self.key_pool.cooldown(key["id"], cooldown, category) + excluded.add(key["id"]) + audit("key_cooldown", started, attempt, category=category, + cooldown_seconds=cooldown) + try: + key = self.key_pool.select(excluded) + except PoolExhausted as exhausted: + raise exhausted from error + continue + if category != "transient": + raise error + failures += 1 + if failures >= 3: + raise error + delay = retry_after_seconds(headers.get("retry-after"), 30) + audit("retry_wait", started, attempt, retry_wait_seconds=delay) + time.sleep(delay) + continue + choices = raw.get("choices") or [] + if not choices: + raise ValueError("slow model response contains no choices") + choice = choices[0] + finish_reason = choice.get("finish_reason") + content = (choice.get("message", {}).get("content") or "").strip() + audit("http_response", started, attempt, http_status=status, headers=headers, + finish_reason=finish_reason, usage=raw.get("usage")) if finish_reason == "length": completion_tokens = (raw.get("usage") or {}).get("completion_tokens") raise ValueError( f"slow model response truncated by max_tokens (finish_reason=length, " f"completion_tokens={completion_tokens}); raise slow_max_tokens" ) - if not content and raw["choices"][0]["message"].get("reasoning_content"): + if not content and choice.get("message", {}).get("reasoning_content"): raise ValueError( "slow model returned empty content (all completion tokens stayed in the " "reasoning channel); escalating reasoning_effort for the next attempt" @@ -172,33 +236,20 @@ def redact(text): if not validate(value): raise ValueError("slow model response is missing required fields") return value - except urllib.error.HTTPError as error: - body = redact(error.read().decode("utf-8", errors="replace")) - audit( - "http_error", - started, - attempt, - http_status=error.code, - headers=public_headers(error.headers), - response=body[:2000], - ) - if attempt == (4 if error.code == 403 else 2): - raise - # 403 多为 TPM/风控限流：指数退避（60/120/240/480s）等配额恢复 - time.sleep(60 * (2 ** attempt) if error.code == 403 else 30) except (urllib.error.URLError, TimeoutError, ValueError) as error: audit( "transport_or_validation_error", started, attempt, error_type=type(error).__name__, - error=redact(str(error))[:1000], + error=self.key_pool.redact(str(error))[:1000], ) - if attempt == 2: + failures += 1 + if failures >= 3: raise # 内容级失败（空正文/截断/非法 JSON）时升档思考强度后重试，不原样重发 current_effort = payload.get("reasoning_effort") if current_effort in _REASON_EFFORT_LADDER: payload["reasoning_effort"] = _REASON_EFFORT_LADDER[current_effort] + audit("retry_wait", started, attempt, retry_wait_seconds=30) time.sleep(30) - raise RuntimeError("Model request retries exhausted; no valid JSON response") diff --git a/src/tandem_eval/__main__.py b/src/tandem_eval/__main__.py index e3ffc95..b4b30a8 100644 --- a/src/tandem_eval/__main__.py +++ b/src/tandem_eval/__main__.py @@ -41,12 +41,14 @@ def main(argv: list[str] | None = None) -> int: help="阶段内动作数上限，达到后请求剩余工作的重规划") run_cmd.add_argument("--slow-model", default=None, metavar="NAME", help="慢模型（大脑）API 名：config 的 harness.slow_model.apis 里选；默认取 default") + run_cmd.add_argument("--slow-key-ids", default=None, metavar="1,2,3", + help="号池 key 编号；多个编号按 worker 均分，未指定则取配置") run_cmd.add_argument("--fast-model", default=None, metavar="NAME", help="快模型（小脑）API 名：config 的 harness.fast_model.apis 里选；默认取 default") run_cmd.add_argument("--slow-max-tokens", type=int, default=None, help="慢模型单次响应上限") - run_cmd.add_argument("--slow-reasoning", choices=["off", "low", "medium", "high"], - default=None, help="慢模型思考强度（默认 low）") + run_cmd.add_argument("--slow-reasoning", choices=["off", "low", "medium", "high", "max"], + default=None, help="慢模型思考强度（默认取所选 API 配置）") run_cmd.add_argument("--export-paths", default="", help="逗号分隔的 VM 内交付物路径") run_cmd.add_argument("--poll", type=float, default=60) run_cmd.add_argument("--timeout", type=float, default=None, help="watch 超时秒数") @@ -76,6 +78,14 @@ def main(argv: list[str] | None = None) -> int: config = load_config(args.config) if args.rjob_command == "run": overrides: dict = {"benchmark": args.benchmark} + if args.slow_key_ids is not None: + try: + overrides["slow_key_ids"] = ( + [int(item.strip()) for item in args.slow_key_ids.split(",")] + if args.slow_key_ids.strip() else [] + ) + except ValueError: + parser.error("--slow-key-ids 需填写正整数编号，多个编号用逗号分隔") manifest: dict = {} if args.tasks: manifest["tasks"] = [item.strip() for item in args.tasks.split(",") if item.strip()] diff --git a/src/tandem_eval/config_schema.py b/src/tandem_eval/config_schema.py index 9736f42..e6d9ceb 100644 --- a/src/tandem_eval/config_schema.py +++ b/src/tandem_eval/config_schema.py @@ -20,6 +20,8 @@ from pathlib import Path from typing import Any +from tandem.models.key_pool import POOL_BASE_URLS, validate_key_ids + #: 必填路径（点号路径，逐段下钻；None / 空串 / 空列表视为缺失） REQUIRED_PATHS = ( "harness.slow_model.default", @@ -72,10 +74,10 @@ FAST_BACKENDS = ("jev", "kev") #: 慢模型每个 API 条目必填的键 -SLOW_API_KEYS = ("base_url", "api_key", "model") +SLOW_API_KEYS = ("base_url", "model") #: 思考强度档位（与 harness 侧 SessionConfig.slow_reasoning_effort 保持一致） -REASONING_EFFORTS = ("off", "low", "medium", "high") +REASONING_EFFORTS = ("off", "low", "medium", "high", "max") class ConfigError(ValueError): @@ -121,6 +123,28 @@ def _validate_slow_models(config: dict) -> list[str]: base_url = api.get("base_url") if isinstance(base_url, str) and base_url.strip() and not base_url.startswith("http"): problems.append(f"harness.slow_model.apis.{name}.base_url（应为 http(s) 地址）") + pool = api.get("key_pool") + if isinstance(pool, dict): + try: + validate_key_ids(pool.get("key_ids", [])) + except ValueError: + problems.append(f"harness.slow_model.apis.{name}.key_pool.key_ids（应为不重复的正整数列表）") + if api.get("api_key") and pool.get("key_ids"): + problems.append(f"harness.slow_model.apis.{name}（api_key 与 key_ids 不能同时指定）") + if _is_missing(api.get("api_key")): + prefix = f"harness.slow_model.apis.{name}.key_pool" + if not isinstance(pool, dict): + problems.append(f"harness.slow_model.apis.{name}.api_key / .key_pool（至少给一个）") + else: + provider = pool.get("provider") + if provider not in POOL_BASE_URLS: + problems.append(f"{prefix}.provider（应为 minimax / kimi）") + elif not isinstance(base_url, str) or base_url.rstrip("/") != POOL_BASE_URLS[provider]: + problems.append(f"{prefix}.provider 与官方 base_url 不匹配") + for key in ("credentials_file", "state_file"): + path = pool.get(key) + if not isinstance(path, str) or not path.startswith("/"): + problems.append(f"{prefix}.{key}（必须是绝对路径）") # 思考强度必须落在 harness 认识的档位里，否则 SessionConfig 会在运行期才报错 effort = api.get("reasoning_effort") if effort is not None and effort not in REASONING_EFFORTS: diff --git a/src/tandem_eval/rjob_spec.py b/src/tandem_eval/rjob_spec.py index b33194e..99250ba 100644 --- a/src/tandem_eval/rjob_spec.py +++ b/src/tandem_eval/rjob_spec.py @@ -20,6 +20,8 @@ import subprocess from pathlib import Path +from tandem.models.key_pool import validate_key_ids + from .config_schema import select_api @@ -65,7 +67,10 @@ def _secrets_section(benchmark: str, config: dict, slow_api: dict, fast_api: dic if benchmark not in {"osworld1", "osworld2", "osworld21"}: return {} secrets = { - "slow_model": {key: slow_api[key] for key in ("base_url", "api_key", "model")}, + "slow_model": { + "base_url": slow_api["base_url"], "api_key": slow_api.get("api_key", ""), + "model": slow_api["model"], + }, # 快模型部署在闲时 GPU，ip/端口随重新部署变化：endpoint 为空时由 worker # 从 endpoint_file（部署后写入的 {"base_url": ...}）自动发现 "fast_model": { @@ -74,6 +79,8 @@ def _secrets_section(benchmark: str, config: dict, slow_api: dict, fast_api: dic }, "registry": config["eval"].get("registry"), } + if not slow_api.get("api_key") and slow_api.get("key_pool"): + secrets["slow_model"]["key_pool"] = dict(slow_api["key_pool"]) if benchmark == "osworld21": section = config["eval"]["benchmarks"][benchmark] environment = {} @@ -111,7 +118,7 @@ def _harness_config_section(config: dict, slow_api: dict, fast_api: dict) -> dic "cli_timeout_seconds": int(route.get("cli_timeout_seconds", 90)), "slow_max_tokens": int(slow_api.get("max_tokens", 16384)), "slow_timeout_seconds": float(slow_api.get("timeout_seconds", 120)), - "slow_reasoning_effort": slow_api.get("reasoning_effort", "low"), + "slow_reasoning_effort": slow_api.get("reasoning_effort"), "slow_direct_api": bool(slow_api.get("direct_api", False)), "fast_backend": fast_api.get("backend", "jev"), "fast_timeout_seconds": float(fast_api.get("timeout_seconds", 300)), @@ -153,6 +160,13 @@ def build_job_spec(*, config: dict, run_id: str, repo_root: Path, requested = dict(overrides.pop("models", {}) or {}) slow_name, slow_api = select_api(config, "slow_model", requested.get("slow")) fast_name, fast_api = select_api(config, "fast_model", requested.get("fast")) + key_ids = overrides.pop("slow_key_ids", None) + if key_ids is not None: + validate_key_ids(key_ids) + if key_ids and (slow_api.get("api_key") or not slow_api.get("key_pool")): + raise ValueError("slow_key_ids requires key_pool without a manual api_key") + if slow_api.get("key_pool"): + slow_api["key_pool"] = {**slow_api["key_pool"], "key_ids": key_ids} inner_image = config["eval"]["inner_image"] if benchmark == "osworld21": diff --git a/tests/eval/test_config.py b/tests/eval/test_config.py index e8b3d9d..5739758 100644 --- a/tests/eval/test_config.py +++ b/tests/eval/test_config.py @@ -40,7 +40,9 @@ def test_slow_model_entries_use_known_knobs(): config = load_config(CONFIG_PATH) for name, api in config["harness"]["slow_model"]["apis"].items(): assert api["base_url"].startswith("http"), f"{name}: base_url" - assert api["reasoning_effort"] in {"off", "low", "medium", "high"}, f"{name}: reasoning_effort" + assert api["reasoning_effort"] in { + None, "off", "low", "medium", "high", "max", + }, f"{name}: reasoning_effort" assert api["max_tokens"] > 0 and api["timeout_seconds"] > 0, f"{name}: 旋钮必须为正" diff --git a/tests/harness/test_session.py b/tests/harness/test_session.py index 2576ac3..1726632 100644 --- a/tests/harness/test_session.py +++ b/tests/harness/test_session.py @@ -325,7 +325,7 @@ def fake_urlopen(request, timeout=None): original = urllib.request.urlopen urllib.request.urlopen = fake_urlopen try: - config = SessionConfig() + config = SessionConfig(slow_reasoning_effort="low") client = SlowModelClient({"base_url": "http://x"}, config, tmp_path / "a.jsonl") client.json_request([{"role": "user", "content": "hi"}], lambda v: True, role="r") assert captured["payload"]["max_tokens"] == 16384 @@ -401,7 +401,8 @@ def fake_urlopen(request, timeout=None): original = urllib.request.urlopen urllib.request.urlopen = fake_urlopen try: - client = SlowModelClient({"base_url": "http://x"}, SessionConfig(), tmp_path / "a.jsonl") + client = SlowModelClient({"base_url": "http://x"}, SessionConfig(slow_reasoning_effort="low"), + tmp_path / "a.jsonl") value = client.json_request([{"role": "user", "content": "hi"}], lambda v: v == {"done": True}, role="plan_step") assert value == {"done": True}