Together AI：调用量上来以后会遇到 429（太快）和 503（服务暂时不可用）

调用量上来以后会遇到 429（太快）和 503（服务暂时不可用）。 这篇做 retry_client.py：只对 429/503 重试，指数退避加随机抖动，服务器给了 Retry-After 就按它等；401/400 这种重试也没用的错误立即抛出。再写一个本地模拟服务器按脚本返回状态码，把各种情况都跑一遍。

先看清成品

• 本地模拟服务器上真实运行 3 个场景（429→503→200、401、连续 503），重试次数和等待时间都打印出来。
• 用无效 Key 对真实 Together 接口调用，401 只请求了 1 次就停止。
• 4 项 pytest 通过。

合规提醒：只用 Together AI 官方 API 和你自己的 API Key，Key 只放环境变量或服务端，不共享账号。按 token 计费，价格以 together.ai/pricing 为准，本文不写价格。每个开源模型有自己的许可证，商用前看模型页面；处理用户消息时注意个人信息保护，不把别人的隐私数据发给第三方服务而不告知。

运行说明：我没有 Together API Key，没有对 Together 发出过成功的请求（也没产生费用）。真实跑过的是：用无效 Key 调用官方接口（2026-10-21 21:44–21:46，UTC+8，返回原样贴在文中），以及全部本地代码。429/503 来自我写的本地模拟服务器，不是 Together 真实返回的。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

86-together-ai-C-src/
├── retry_client.py
├── mock_server.py
└── test_retry.py

第 1 步：重试客户端

"""调用 Together chat/completions 的重试封装。
官方速率限制文档：429 → 降低速率并指数退避；503 → 短暂等待后带退避重试。文档没给具体数值，下面的参数是我自己的默认值。
401/400 等客户端错误重试也没用，立即抛出。"""
import json, random, time, urllib.request, urllib.error

RETRY = {429, 503}


class APIError(Exception):
    def __init__(self, status, body): super().__init__(f"HTTP {status}: {body[:200]}"); self.status = status


def backoff(attempt, base=1.0, cap=30.0, retry_after=None, rnd=random.random):
    if retry_after is not None:
        try:
            return min(cap, float(retry_after))
        except ValueError:
            pass
    return min(cap, base * 2 ** attempt) * (0.5 + rnd() / 2)   # 指数 + 抖动，避免多个进程同时醒来


def chat(payload, key, url="https://api.together.ai/v1/chat/completions", max_attempts=5,
         sleep=time.sleep, log=print, base=1.0):
    data = json.dumps(payload).encode()
    for attempt in range(max_attempts):
        req = urllib.request.Request(url, data=data, method="POST", headers={
            "Authorization": f"Bearer {key}", "Content-Type": "application/json",
            "User-Agent": "binbho-tutorial/1.0"})
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", "replace")
            if e.code not in RETRY or attempt == max_attempts - 1:
                raise APIError(e.code, body) from None
            wait = backoff(attempt, base=base, retry_after=e.headers.get("Retry-After"))
            log(f"HTTP {e.code}，第 {attempt + 1} 次失败，{wait:.2f}s 后重试")
            sleep(wait)

第 2 步：模拟服务器

"""本地模拟服务器（教程作者编写，不是 Together 的行为录像）：按脚本依次返回状态码，用来练习重试逻辑。"""
import json, sys, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


def serve(script, port=0):
    calls = []

    class H(BaseHTTPRequestHandler):
        def do_POST(self):
            self.rfile.read(int(self.headers.get("Content-Length", 0)))
            code = script[min(len(calls), len(script) - 1)]; calls.append(code)
            body = {"choices": [{"message": {"role": "assistant", "content": "ok"}}]} if code == 200 \
                else {"error": {"message": f"mock {code}"}}
            self.send_response(code); self.send_header("Content-Type", "application/json")
            if code == 429: self.send_header("Retry-After", "0.2")
            self.end_headers(); self.wfile.write(json.dumps(body).encode())
        def log_message(self, *a): pass

    s = ThreadingHTTPServer(("127.0.0.1", port), H)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    return s, calls


if __name__ == "__main__":
    import retry_client, time
    s, calls = serve([int(x) for x in sys.argv[1:]] or [429, 503, 200])
    url = f"http://127.0.0.1:{s.server_address[1]}/v1/chat/completions"
    t = time.time()
    try:
        r = retry_client.chat({"model": "m", "messages": []}, "k", url=url, base=0.3)
        print("结果：", r["choices"][0]["message"]["content"])
    except retry_client.APIError as e:
        print("放弃：", e)
    print(f"服务器收到 {len(calls)} 次请求 {calls}，耗时 {time.time() - t:.2f}s")

第 3 步：测试

import pytest, retry_client as rc
from mock_server import serve

def run(script, **kw):
    s, calls = serve(script); waits = []
    url = f"http://127.0.0.1:{s.server_address[1]}/v1/chat/completions"
    try:
        return rc.chat({"model": "m", "messages": []}, "k", url=url, sleep=waits.append, log=lambda *_: None, **kw), calls, waits
    finally:
        s.shutdown()

def test_recovers_after_429_503():
    r, calls, waits = run([429, 503, 200])
    assert calls == [429, 503, 200] and r["choices"][0]["message"]["content"] == "ok"
    assert waits[0] == 0.2          # 429 带 Retry-After 时优先用它

def test_401_not_retried():
    with pytest.raises(rc.APIError) as e: run([401, 200])
    assert e.value.status == 401

def test_gives_up():
    with pytest.raises(rc.APIError): run([503] * 9, max_attempts=3)

def test_backoff_grows_and_caps():
    v = [rc.backoff(a, rnd=lambda: 1.0) for a in range(8)]
    assert v[:3] == [1.0, 2.0, 4.0] and max(v) == 30.0
    assert rc.backoff(0, rnd=lambda: 0.0) == 0.5
    assert rc.backoff(3, retry_after="bad", rnd=lambda: 1.0) == 8.0

第 4 步：运行

$ python mock_server.py 429 503 200
HTTP 429，第 1 次失败，0.20s 后重试
HTTP 503，第 2 次失败，0.34s 后重试
结果： ok
服务器收到 3 次请求 [429, 503, 200]，耗时 0.55s
$ python mock_server.py 401 200
放弃： HTTP 401: {"error": {"message": "mock 401"}}
服务器收到 1 次请求 [401]，耗时 0.01s
$ python mock_server.py 503 503 503 503 503
HTTP 503，第 1 次失败，0.23s 后重试
HTTP 503，第 2 次失败，0.55s 后重试
HTTP 503，第 3 次失败，0.65s 后重试
HTTP 503，第 4 次失败，2.15s 后重试
放弃： HTTP 503: {"error": {"message": "mock 503"}}
服务器收到 5 次请求 [503, 503, 503, 503, 503]，耗时 3.59s

# 真实 Together，无效 Key（2026-10-21 21:46 UTC+8）：只请求一次
$ python -c "import retry_client as r; r.chat({'model':'Qwen/Qwen3.5-9B','messages':[{'role':'user','content':'hi'}]}, 'invalid-demo')"
retry_client.APIError: HTTP 401: { "id": "p685PwA-3pDw3Z-a4e0b5d7ae4a289e", "error": { "message": "Invalid API key provided. …", "type": "invalid_request_error", …
$ python -m pytest -q
4 passed

怎么检查结果？

• 第一个场景里 429 的等待正好是 0.20s：模拟服务器发了 Retry-After: 0.2，客户端按它等。
• 503 的等待大约每次翻倍，但因为有抖动不是精确的 2 倍。
• 401 只请求一次。

调试示例

1. 真实运行里的等待时间不是 1、2、4。 看上面连续 503 的输出：0.23、0.55、0.65、2.15 秒（base=0.3），第 3 次比第 2 次只多了一点。这是抖动造成的：实际等待是 基数 × 2^n × (0.5~1.0)。所以测试不能直接断言具体秒数，我把随机函数做成参数（rnd=lambda: 1.0 测上限，rnd=lambda: 0.0 测下限 0.5s）。

2. Retry-After 可能不是数字。 HTTP 规范里 Retry-After 也可以是日期字符串。直接 float() 会抛 ValueError，把整次调用弄崩。现在解析失败就退回指数退避，有测试 retry_after="bad" 覆盖。

3. 别对 401 重试。 用 openai SDK 时（A 篇设了 max_retries=2），SDK 自己也会重试某些错误；如果外面再套一层重试，429 时请求数会成倍增加。二选一：要么用 SDK 自带的，要么像本篇这样自己控制，别两层叠加。

常见问题

模板规则

• 只重试 429/503。
• 指数退避加抖动，并设上限。
• 优先遵守 Retry-After。
• 重试只保留一层。

项目资源

重试是为了扛住偶发问题，不是为了把限流硬顶过去。