多 Key 的价值就三件事：把并发压力摊到几条独立额度上、按业务线把费用分开记账、某一条 Key 出问题时业务不停摆。轮询也不是"随便换一条"，而是先看健康状态再挑：cooling 或者 dead 的直接跳过，只在 healthy 里按顺序轮。
一条 Key 扛整条业务线的全部请求，高峰期就是单点。拆成几条之后，同一时刻的请求落在不同 Key 上，单条 Key 的瞬时并发就下来了。这里分摊的是并发和流量分布，不是去撬动什么额度上限，池子只负责把流量排得均匀一点。
每条业务线用自己的 Key，月底拉账单不用靠猜：客服机器人烧了多少、批量翻译烧了多少，各算各的。全挤在一条上，只能看见总数，看不见钱花在哪儿。
Key 失效的原因很多：账户欠费、权限被改、配置写错、上游临时收紧。只有一条 Key 时，一句 openai.AuthenticationError: Error code: 401 就能让整条业务线全挂，只能停机改配置。多条 Key 配上健康检查和自动切换，这条挂了就换下一条。
多 Key 不是必选项。一天几百次调用、一条 Key 半年没出过事，加池子只是给自己找活干。
同时跑着客服机器人、内容审核、数据清洗三条链路时，成本要能拆开报到各业务方头上。一条 Key 记不清谁用了多少，只能按人头平摊，谁都不服气。一条业务线一条 Key，用量和费用天然对齐。
三个人共用一个开发用的 Key，谁把脚本写成死循环，谁都说不清。分开之后，影响面限制在他自己那条 Key 上，看调用曲线就知道是谁的锅。
高并发下单条 Key 很容易撞到 openai.RateLimitError: Error code: 429，被限流之后整条链路跟着卡住。流量分到几条 Key 上，每条的请求数都低一些，撞限流的概率也就小了。这条业务线如果支撑线上服务，切换能力本身就是可用性的一部分。
池子能不能自动扛住故障，全看状态设计得对不对：设计歪了，要么坏 Key 一直被挑中，要么好 Key 被误杀。
healthy 表示现在能用；cooling 表示暂时不能用，旁边挂一个 cool_until 时间戳，到点自己回来；dead 表示认证类失败，比如 Key 被删、权限被收，这类错误重试多少次结果都一样，标记出来等人处理。别再往里加状态：状态越多，流转组合越乱。
状态只说明"此刻能不能用"，说明不了"最近是不是老出问题"。一条 Key 每分钟偶发一次超时，状态永远是 healthy，可它一直在拖慢链路。用 fail_count 记连续失败次数，到阈值就推进冷却，snapshot() 里也能一眼看出该查哪条；成功一次归零，免得旧账挂着。
这张表就是池子的全部逻辑，照着写代码不会跑偏：
| 上游返回 | 池子动作 | 持续多久 | 怎么恢复 |
|---|---|---|---|
429（openai.RateLimitError） | 进 cooling | 60 秒 | 到期自动回 healthy |
401 / 403 | 直接标 dead 并告警 | 一直保持 | 人工换 Key 或改配置 |
402（额度不足） | 标 dead 并告警 | 一直保持 | 充值或换 Key |
连续 3 次 5xx | 进 cooling | 300 秒 | 到期自动回 healthy |
| 超时、连接错误 | fail_count 加一，状态不动 | 无 | 下轮照常轮询 |
cooling 到期 | 回 healthy，计数清零 | 无 | 不用人工干预 |
429 给 60 秒，是因为限流一般按分钟窗口统计，等一个窗口过去再试，比立刻重试划算。5xx 给 300 秒，是上游抖动往往几分钟才稳。401 和 403 直接标死，因为这两类错误不会因为"再等等"而变好。
| 策略 | 怎么挑 | 好处 | 代价 |
|---|---|---|---|
| 轮询（round-robin） | 按顺序一条条轮着用 | 每条 Key 被用到的次数接近 | 要维护一个游标 |
| 加权 | 额度大的多分流量 | 额度不会被闲置 | 权重靠人工维护 |
| 最少使用（least-used） | 挑累计调用次数少的 | 长期看分布更均匀 | 历史次数要落盘 |
| 随机 | 从健康 Key 里随机挑 | 几行代码写完 | 可能连着撞同一条 |
轮询的好处是每条 Key 被用到的次数接近，月末看账单，哪条消耗偏高一眼就能看出来；再叠一层冷却过滤，坏掉的 Key 自动退出轮转。实现成本也低：一个游标加一个列表，几十行就写完。加权适合额度差得多的场景，最少使用适合长跑服务，代价是历史次数得落盘。多 Key 在这里是为了分摊并发和隔离故障，不是为了凑模型——一个 Key 就覆盖全系列模型。
下面这个类只用标准库，threading.Lock 保护共享状态，状态文件用原子替换写入。直接存成 keypool.py 就能跑。
初始化干三件事：检查 Key 列表非空、查出重复项、把上一次的冷却状态读回来。重复 Key 一定要拦下来，不然它会被当成两条 Key 参与轮询，看着像两条并发，实际全打在同一份额度上。
acquire() 先顺手把到期的 cooling 放回 healthy，再只在 healthy 里按游标挑一条，并支持传 exclude 集合，重试时避开刚失败的那条。一条都挑不出来时，别返回 None 让调用方去猜，直接抛 NoHealthyKeyError，把"最近的一条还要等多少秒"塞进异常，上层才能决定是等一下还是降级返回。
成功要把 fail_count 归零、累加调用次数；失败按状态码分流，决定进冷却还是标死。snapshot() 返回纯数据快照，日志和监控都从它取值，里面只打 Key 的后六位，完整 Key 不要写进日志文件。
# keypool.py —— 只用标准库，可直接运行
import json
import os
import time
from dataclasses import dataclass
from threading import Lock
@dataclass
class KeyState:
key: str
status: str = "healthy" # healthy / cooling / dead
cool_until: float = 0.0 # 冷却截止时间（秒级时间戳）
fail_count: int = 0 # 连续失败次数，成功一次归零
calls: int = 0 # 累计成功调用次数
last_error: str = "" # 最近一次失败原因，方便排查
class NoHealthyKeyError(RuntimeError):
"""池子里没有可用 Key。把等待时间带出去，让上层决定排队还是降级。"""
def __init__(self, wait_seconds, total):
self.wait_seconds = wait_seconds
self.total = total
super().__init__(
"池子里 %d 条 Key 都不可用，最近的冷却还要等 %.1f 秒" % (total, wait_seconds)
)
class KeyPool:
COOL_429 = 60 # 429：冷却 60 秒
COOL_5XX = 300 # 连续 3 次 5xx：冷却 300 秒
MAX_5XX = 3
def __init__(self, keys, state_path="keypool_state.json"):
if not keys:
raise ValueError("Key 池是空的，先检查 XYU_API_KEYS 配置")
dup = sorted({k for k in keys if keys.count(k) > 1})
if dup:
raise ValueError("Key 有重复：%s，重复项不会带来额外并发" % ",".join(dup))
self._lock = Lock()
self._state_path = state_path
self._states = {k: KeyState(key=k) for k in keys}
self._rr = 0
self._load_state() # 重启后先把上次的冷却状态读回来
# ---------- 状态持久化：先写临时文件，再原子改名 ----------
def _load_state(self):
if not os.path.exists(self._state_path):
return
try:
with open(self._state_path, "r", encoding="utf-8") as f:
saved = json.load(f)
except (OSError, ValueError) as e:
print("[KeyPool] 状态文件读不了，忽略并重新开始：%r" % e)
return
now = time.time()
for key, st in self._states.items():
old = saved.get(key)
if not isinstance(old, dict):
continue
try:
st.fail_count = int(old.get("fail_count", 0))
st.calls = int(old.get("calls", 0))
st.status = str(old.get("status", "healthy"))
st.cool_until = float(old.get("cool_until", 0.0))
except (TypeError, ValueError):
continue # 单条脏数据不影响其它 Key
if st.status == "cooling" and st.cool_until <= now:
st.status, st.fail_count, st.cool_until = "healthy", 0, 0.0
def _save_state(self):
data = {k: {"status": s.status, "cool_until": s.cool_until,
"fail_count": s.fail_count, "calls": s.calls}
for k, s in self._states.items()}
tmp = self._state_path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
f.flush()
os.fsync(f.fileno()) # 先落盘，避免断电留下半截文件
os.replace(tmp, self._state_path) # Windows / Linux 上都是原子替换
# ---------- 挑 Key ----------
def acquire(self, exclude=None):
exclude = exclude or set()
with self._lock: # 锁只包状态读写，不包网络请求
now = time.time()
for st in self._states.values(): # 冷却到期的顺手放回 healthy
if st.status == "cooling" and st.cool_until <= now:
st.status, st.fail_count, st.cool_until = "healthy", 0, 0.0
pool = [st for st in self._states.values()
if st.status == "healthy" and st.key not in exclude]
if not pool:
waits = [st.cool_until - now for st in self._states.values()
if st.status == "cooling"]
raise NoHealthyKeyError(min(waits) if waits else 0.0, len(self._states))
self._rr = (self._rr + 1) % len(pool)
return pool[self._rr].key
def report_success(self, key):
with self._lock:
st = self._states.get(key)
if st is None:
return # 传进来的 Key 不属于本池，直接忽略
st.status, st.cool_until, st.fail_count = "healthy", 0.0, 0
st.calls += 1
self._save_state()
def report_failure(self, key, status_code=None, message=""):
with self._lock:
st = self._states.get(key)
if st is None:
return
st.last_error = ("%s %s" % (status_code, message))[:160]
st.fail_count += 1
if status_code in (401, 403, 402): # 认证 / 额度类，重试没意义
st.status = "dead"
elif status_code == 429:
st.status, st.cool_until = "cooling", time.time() + self.COOL_429
elif status_code is not None and status_code >= 500:
if st.fail_count >= self.MAX_5XX:
st.status, st.cool_until = "cooling", time.time() + self.COOL_5XX
self._save_state()
def snapshot(self):
"""返回纯数据快照，日志和监控都用它；只打 Key 后六位，别把完整 Key 写进日志。"""
with self._lock:
now = time.time()
rows = []
for st in self._states.values():
left = round(max(0.0, st.cool_until - now), 1) if st.status == "cooling" else 0.0
rows.append({"key_tail": st.key[-6:], "status": st.status,
"cool_left": left, "fail_count": st.fail_count,
"calls": st.calls, "last_error": st.last_error})
return rows
if __name__ == "__main__":
pool = KeyPool(["sk-demo0001", "sk-demo0002"])
ok_key = pool.acquire()
pool.report_success(ok_key)
pool.report_failure("sk-demo0002", 429, "RateLimitError")
for row in pool.snapshot():
print(row)两个线程同时调 acquire()，都读到游标是 3，都算出同一个下标，于是拿到同一条 Key：你以为是分摊并发，实际上是两条线程抢一份额度。第二个坑更隐蔽，一个线程读了 fail_count 准备写回去，另一个线程也在做同样的事，一交错前一次的计数就没了，日志里的失败次数比告警次数少。
不加锁的版本在压测里很好认：某条 Key 的调用量明显偏高，另外几条几乎没被碰过，snapshot() 里的计数也和实际请求数差一截。加锁之后，acquire() 和 report_*() 对共享字段的读写都在同一把锁里完成，谁先谁后有了明确顺序。
锁只包住"挑 Key、改计数"这几行。别把 client.chat.completions.create() 写进锁里：一次请求要几秒到几十秒，锁一直被占着，其他线程全在门口排队，等于把并发压回单线程。挑完 Key 立刻出锁，发请求时不持锁。
协程版要改两处。锁换成 async with asyncio.Lock()，写成 with threading.Lock() 会阻塞整个事件循环；等待换成 await asyncio.sleep()，协程里写一句 time.sleep(1)，事件循环被按住一秒，这一秒内所有并发请求一起停。
# async_keypool.py —— 同步版的关键改动都在这几行上
import asyncio
import time
class AsyncKeyPool:
COOL_429 = 60
COOL_5XX = 300
MAX_5XX = 3
def __init__(self, keys):
if not keys:
raise ValueError("Key 池是空的")
self._lock = asyncio.Lock() # 换掉 threading.Lock
self._states = {k: {"status": "healthy", "cool_until": 0.0, "fail_count": 0}
for k in keys}
self._rr = 0
async def acquire(self, exclude=None):
exclude = exclude or set()
async with self._lock: # 必须 async with，with 会阻塞事件循环
now = time.time()
for st in self._states.values():
if st["status"] == "cooling" and st["cool_until"] <= now:
st["status"], st["fail_count"], st["cool_until"] = "healthy", 0, 0.0
pool = [k for k, st in self._states.items()
if st["status"] == "healthy" and k not in exclude]
if not pool:
waits = [st["cool_until"] - now for st in self._states.values()
if st["status"] == "cooling"]
raise RuntimeError("全池不可用，最近的还要等 %.1f 秒"
% (min(waits) if waits else 0.0))
self._rr = (self._rr + 1) % len(pool)
return pool[self._rr]
async def report_failure(self, key, status_code):
async with self._lock:
st = self._states[key]
st["fail_count"] += 1
if status_code in (401, 403, 402):
st["status"] = "dead"
elif status_code == 429:
st["status"], st["cool_until"] = "cooling", time.time() + self.COOL_429
elif status_code and status_code >= 500 and st["fail_count"] >= self.MAX_5XX:
st["status"], st["cool_until"] = "cooling", time.time() + self.COOL_5XX
async def acquire_with_wait(pool, timeout=30.0):
"""池子全在冷却时等一下再拿：协程里用 asyncio.sleep，不要用 time.sleep。"""
deadline = time.time() + timeout
while time.time() < deadline:
try:
return await pool.acquire()
except RuntimeError as e:
print("这一轮没拿到 Key：%s" % e)
await asyncio.sleep(1.5) # time.sleep 会把整个事件循环按住
return None # 等超时了交给上层降级Key 只放在环境变量或者 .env 文件里，代码里一行都不许写死。启动时做校验，三类情况直接让进程起不来：列表里有空项（多半是结尾多打了一个逗号）、出现重复 Key、格式不对（少了前缀或者混进空格）。校验要在服务监听端口之前跑完，失败就用非零退出码结束进程：进程管理看到退出码会重启并留下日志，值班的人一眼就知道是配置问题。
# .env 放在项目根目录并加进 .gitignore，一行一条 Key，逗号分隔
# XYU_API_KEYS=sk-aaaa1111,sk-bbbb2222,sk-cccc3333
# XYU_BASE_URL=https://xyuapi.top/v1
# XYU_STATE_FILE=./keypool_state.json
# 启动前校验：空项 / 重复 / 格式不对，任何一种都非零退出（Windows 下存成 check_env.py 再运行）
python - <<'PY'
import re, sys
raw = open(".env", encoding="utf-8").read()
m = re.search(r"^XYU_API_KEYS=(.*)$", raw, re.M)
if not m:
sys.exit("XYU_API_KEYS 没配置，先确认 .env 放在了项目根目录")
keys = [k.strip() for k in m.group(1).split(",")]
if not keys or any(not k for k in keys):
sys.exit("Key 列表里有空项，多半是结尾多了一个逗号：%r" % keys)
dup = sorted({k for k in keys if keys.count(k) > 1})
if dup:
sys.exit("Key 重复，重复项不会带来额外并发：%s" % ",".join(dup))
bad = [k for k in keys if not re.match(r"^sk-[A-Za-z0-9_\-]{8,}$", k)]
if bad:
sys.exit("格式不对（少了 sk- 前缀或者混进了空格）：%s" % bad)
print("校验通过，共 %d 条 Key" % len(keys))
PY
# 想确认某条 Key 现在能不能用，别拿业务代码去试，直接打 /v1/models 看状态码
curl -s -o /dev/null -w '%{http_code}\n' https://xyuapi.top/v1/models -H "Authorization: Bearer $KEY"进程重启、容器滚动更新，都会把内存里的状态清空。如果 cool_until 只存在内存里，重启之后所有 Key 又都变回 healthy，明摆着还在限流的 Key 立刻被挑走，接着再撞一次 429。把 status、cool_until、fail_count 写进 JSON 文件，重启后读回来，冷却状态不丢，标为 dead 的 Key 也继续保持标记提醒人处理。
直接往目标文件上覆盖写有风险：写到一半进程被杀或者断电，文件只剩半截，下次读的时候 JSON 解析直接失败。稳妥做法是先写同目录下的临时文件，flush() 加 os.fsync() 落到磁盘，再用 os.replace() 改名覆盖。os.replace 在 Windows 和 Linux 上都是原子的，读的一方不会读到写了一半的内容。
限流管速度、重试管单次失败、Key 池管账号级故障，三层各管一段，别混在一处写。调用函数只做四件事：从池子拿 Key、发请求、按结果回报池子、失败时换一条重试一次。重试一次就够了，无脑重试会把一次小故障放大成雪崩。池子空或者全在冷却时，把等待时间抛给上层，由上层决定排队等还是降级返回缓存。
# caller.py —— 拿 Key、发请求、回报池子、换 Key 重试一次
import os
import time
from openai import OpenAI
from keypool import KeyPool, NoHealthyKeyError
def call_with_pool(pool, messages, model="gemini-2.5-pro", retry=1, timeout=60):
base_url = os.environ.get("XYU_BASE_URL", "https://xyuapi.top/v1")
tried, last_error = set(), None
for _ in range(retry + 1):
try:
key = pool.acquire(exclude=tried)
except NoHealthyKeyError as e:
# 池子空或者全在冷却：把等待时间交给上层，别在这里死循环
raise RuntimeError("暂时没有可用 Key，约 %.1f 秒后再来" % e.wait_seconds) from e
tried.add(key)
client = OpenAI(api_key=key, base_url=base_url, timeout=timeout)
try:
resp = client.chat.completions.create(model=model, messages=messages)
pool.report_success(key)
return resp.choices[0].message.content
except Exception as e:
# 各版本 SDK 的异常类名不完全一致，按状态码分流更稳；取不到就按可恢复错误处理
code = getattr(e, "status_code", None)
if code is None:
text = str(e)
for guess in (401, 403, 429, 500):
if ("Error code: %d" % guess) in text:
code = guess
break
pool.report_failure(key, code if code is not None else 500, str(e))
last_error = e
time.sleep(1.0) # 简单退避，别贴着上一次的失败猛打
raise RuntimeError("换了 Key 还是失败：%s" % last_error)每天早上扫一眼 snapshot()：有没有 dead 的 Key 挂着、哪条的 fail_count 一直往上涨。dead 代表认证或者额度问题，重试没有用，得去后台查清楚是欠费、被删还是权限变动。调用次数也要看，如果某条的调用量是别人的三倍，要么是轮询写错了，要么是另外几条长期在冷却。
把调用量和失败率打到监控里，按小时看趋势。失败率突然抬头，通常是上游在收紧或者额度快见底；调用量长时间偏斜，多半是有 Key 长期不可用，实际参与轮转的只剩一两条。发现偏斜就回去看 snapshot()，别等撞上大规模 429 才动手。
按次计费的好处是账好算：一次调用一个固定价格，输入写多长都不影响单价，分摊并发时不用纠结谁的消息更长。3 个人的小团队，每人每天 2000 次调用，用 gemini-2.5-pro：每人每天 2000 × 0.031 = 62 元，团队一天 186 元，一个月按 30 天算大约 5580 元。短问答挑 0.031 到 0.05 元那一档，代码改造再用 0.09 元往上的。
| 模型 | 单价（元/次） |
|---|---|
gemini-2.5-pro | 0.031 |
deepseek-v3.2-thinking | 0.049 |
deepseek-r1-thinking | 0.049 |
deepseek-v4-flash-thinking | 0.05 |
grok-4.1 | 0.05 |
gemini-3-pro-preview | 0.05 |
grok-4.5 / grok-4.6 | 0.09 |
claude-sonnet-4-5-thinking | 0.09 |
deepseek-v4-pro-thinking | 0.09 |
kimi-k2.5 / kimi-k2.6 | 0.09 |
gemini-3.1-pro-preview | 0.09 |
claude-opus-4-5-thinking | 0.12 |
claude-sonnet-4-6-thinking | 0.2 |
gpt-5.5 | 0.2 |
claude-opus-4-6-thinking | 0.25 |
gpt-5.3-pro | 0.3 |
多 Key 在这里的作用是把账拆开：三个人各自一条 Key，谁跑了 2000 次、谁跑了 200 次，账单上写得清清楚楚；混在一条上，你只知道花了 186 元，不知道这钱算谁头上。可用性的账更值得算：只有一条 Key 时，它一旦失效，从发现异常到改配置、重启服务，停摆通常按小时计。按一个月 720 小时粗算，98% 可用意味着约 14 个小时用不了，99.9% 大约是 43 分钟，差别主要来自切换是自动的还是靠人。数字不是承诺，取决于切换逻辑写得稳不稳，但这个量级的差距值不值得写一百多行代码，账是清楚的。
接入上，主入口 https://xyuapi.top/v1，备用 https://xyuai.cc/v1，Gemini 原生协议走 /v1beta，其余按 OpenAI 兼容协议走 /v1/chat/completions。按次计费为主，另有按量计费与无限卡套餐；最低充值 7 元，支付宝、微信都能付，不需要海外信用卡。一个 Key 就覆盖全系列模型，池子里的多条 Key 是为了分摊并发、隔离业务和故障切换，不是按模型各备一条。