feat: add priority API key pools and worker key selection

README.md

@@ -216,21 +216,38 @@ tandem-eval rjob run --run-id <name> --benchmark osworld2 --tasks <task-id> \
 | `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`。
 
 ## 排序后端

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
         }
       }
     },

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")

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