⚠️ 待更新·2026-08-29核验 · 更新时间待核验 · 本文信息可能已过期,请以官方文档为准
更新时间:2026-08-29 · 核验状态:待更新 · 官方溯源待补
简介
免费 API 几乎都有速率限制,常见的有三种:RPM(每分钟请求数)、TPM(每分钟 token 数)、日配额(每天总请求数)。本篇讲清楚如何识别和处理它们,避免脚本一跑就被封,并给出退避重试与多 provider 容灾的完整代码。
架构图
flowchart TD
A[429 Too Many Requests] --> B[Exponential backoff]
B --> C[Retry after delay]
C -->|Still 429| D[Switch to next provider]
D --> A
C -->|Success| E[Response returned]
A --> F[Identify limit type]
F -->|RPM| G[Spread calls in time]
F -->|TPM| H[Shorten prompts]
F -->|Daily quota| I[Switch provider]
识别限流响应
各家平台返回的 429 错误体格式不一,但都包含关键信息:
- OpenAI / OpenRouter:
headers["x-ratelimit-remaining-requests"]
- Groq:
error.code == "rate_limit_exceeded"
- Gemini:
error.status == 429 + RESOURCE_EXHAUSTED
正确的做法是统一捕获 RateLimitError,而不是只看 HTTP 状态码。同时把响应头里的 x-ratelimit-* 写入日志,提前预警。
指数退避重试
import os, time, random
from openai import OpenAI, RateLimitError
client = OpenAI(
api_key=os.environ["GROQ_API_KEY"],
base_url="https://api.groq.com/openai/v1",
)
def chat_with_retry(messages, max_retries=5):
for attempt in range(max_retries):
try:
resp = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=messages,
max_tokens=256,
)
return resp.choices[0].message.content
except RateLimitError as e:
if attempt == max_retries - 1:
raise
# 指数退避 + 抖动,避免 thundering herd
wait = (2 ** attempt) + random.uniform(0, 1)
print(f"rate limited, retry in {wait:.1f}s")
time.sleep(wait)
print(chat_with_retry([{"role":"user","content":"hi"}]))
多 provider 自动切换
单家触顶时,可切换到备用 provider:
PROVIDERS = [
("groq", OpenAI(api_key=os.environ["GROQ_API_KEY"], base_url="https://api.groq.com/openai/v1"), "llama-3.3-70b-versatile"),
("together", OpenAI(api_key=os.environ["TOGETHER_API_KEY"], base_url="https://api.together.xyz/v1"), "meta-llama/Llama-3.3-70B-Instruct-Turbo"),
("openrouter",OpenAI(api_key=os.environ["OPENROUTER_API_KEY"],base_url="https://openrouter.ai/api/v1"), "meta-llama/llama-3.3-70b-instruct:free"),
]
def chat_failover(prompt):
for name, client, model in PROVIDERS:
try:
r = client.chat.completions.create(
model=model,
messages=[{"role":"user","content":prompt}],
max_tokens=128,
)
return name, r.choices[0].message.content
except Exception as e:
print(f"[{name}] failed: {e}, trying next")
raise RuntimeError("all providers failed")
who, ans = chat_failover("hi")
print(f"answered by {who}: {ans}")
本地令牌桶限速
主动控制请求频率,避免被服务端拒绝:
import time, threading
from collections import deque
class TokenBucket:
def __init__(self, rate, capacity):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last = time.monotonic()
self.lock = threading.Lock()
def acquire(self):
with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
self.last = now
if self.tokens < 1:
wait = (1 - self.tokens) / self.rate
time.sleep(wait)
self.tokens = 0
else:
self.tokens -= 1
bucket = TokenBucket(rate=15/60, capacity=5) # 15 RPM, 突发 5
def safe_chat(prompt):
bucket.acquire()
return chat_with_retry([{"role":"user","content":prompt}])
实战技巧
- 本地缓存:对相同 prompt 的结果做 hash key 缓存,直接命中本地。
- 批量合并:多个独立请求合并成一次 batch(OpenRouter 支持)。
- 错峰:免费层高峰时段(北京时间 21:00-23:00)容易排队,脚本可放到凌晨跑。
- 监控剩余配额:把
x-ratelimit-remaining 写日志,提前预警。
监控与告警
在生产环境,建议把每次请求的 x-ratelimit-remaining-* 头写入时序数据库(如 Prometheus + Grafana),当剩余配额低于 20% 时自动告警。同时记录每次 429 的 provider 与时间,定位是哪家触顶。简单的做法是在 chat_with_retry 里加一层装饰器,把异常与重试次数打到日志服务。多线程并发场景下,TokenBucket 已加锁,可直接多线程共享一个实例;缓存击穿则通过请求合并(同一 prompt 在 100ms 内只发一次)解决。
常见问题
- 退避后仍 429:可能日配额已耗尽,等次日或换 provider。
- TPM 超限:减小
max_tokens,或截断历史 message。
- 被封 IP:多家平台共享限流规则,频繁刷会被临时拉黑,务必带退避。
合理限流应对能让免费额度发挥 10 倍价值。
退避算法示例
import time, random
def retry_with_backoff(call, max_retries=5, base_delay=1.0):
for i in range(max_retries):
try:
return call()
except (RateLimitError, ServerError) as e:
if i == max_retries - 1: raise
delay = base_delay * (2 ** i) + random.uniform(0, 1)
time.sleep(delay)
最佳实践
- 指数退避 + jitter:纯指数退避会让多个客户端同时重试,加 jitter 避免惊群。
- 退避上限设 60s:单次退避超过 60s 用户已经放弃,再退避没意义。
- 多 provider 切换:第一家连续 2 次 429 就切到备用 provider,不要在一家死磕。
- 限速类型识别:429 响应头里
X-RateLimit-Remaining 字段告诉你剩多少,靠这个判断是 RPM 还是 daily quota。
⚠️ Pending Update · 2026-08-29 Verification · Content may be outdated, please refer to official docs
Updated: 2026-08-29 · Status: Pending Verification
Introduction
Almost every free API has rate limits, in three common forms: RPM (requests per minute), TPM (tokens per minute), and daily quota (requests per day). This article explains how to identify and handle them so your scripts do not get banned, with full code for backoff retry and multi-provider failover.
架构图
flowchart TD
A[429 Too Many Requests] --> B[Exponential backoff]
B --> C[Retry after delay]
C -->|Still 429| D[Switch to next provider]
D --> A
C -->|Success| E[Response returned]
A --> F[Identify limit type]
F -->|RPM| G[Spread calls in time]
F -->|TPM| H[Shorten prompts]
F -->|Daily quota| I[Switch provider]
Identify Rate-limit Responses
Each platform returns 429 differently, but key info is always present:
- OpenAI / OpenRouter:
headers["x-ratelimit-remaining-requests"]
- Groq:
error.code == "rate_limit_exceeded"
- Gemini:
error.status == 429 with RESOURCE_EXHAUSTED
The correct approach is to catch RateLimitError rather than only checking HTTP status. Also log x-ratelimit-* headers for early warning.
Exponential Backoff Retry
import os, time, random
from openai import OpenAI, RateLimitError
client = OpenAI(
api_key=os.environ["GROQ_API_KEY"],
base_url="https://api.groq.com/openai/v1",
)
def chat_with_retry(messages, max_retries=5):
for attempt in range(max_retries):
try:
resp = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=messages,
max_tokens=256,
)
return resp.choices[0].message.content
except RateLimitError as e:
if attempt == max_retries - 1:
raise
# Exponential backoff + jitter to avoid thundering herd
wait = (2 ** attempt) + random.uniform(0, 1)
print(f"rate limited, retry in {wait:.1f}s")
time.sleep(wait)
print(chat_with_retry([{"role":"user","content":"hi"}]))
Multi-provider Failover
When one provider hits the limit, switch to another:
PROVIDERS = [
("groq", OpenAI(api_key=os.environ["GROQ_API_KEY"], base_url="https://api.groq.com/openai/v1"), "llama-3.3-70b-versatile"),
("together", OpenAI(api_key=os.environ["TOGETHER_API_KEY"], base_url="https://api.together.xyz/v1"), "meta-llama/Llama-3.3-70B-Instruct-Turbo"),
("openrouter",OpenAI(api_key=os.environ["OPENROUTER_API_KEY"],base_url="https://openrouter.ai/api/v1"), "meta-llama/llama-3.3-70b-instruct:free"),
]
def chat_failover(prompt):
for name, client, model in PROVIDERS:
try:
r = client.chat.completions.create(
model=model,
messages=[{"role":"user","content":prompt}],
max_tokens=128,
)
return name, r.choices[0].message.content
except Exception as e:
print(f"[{name}] failed: {e}, trying next")
raise RuntimeError("all providers failed")
who, ans = chat_failover("hi")
print(f"answered by {who}: {ans}")
Local Token Bucket
Throttle outbound requests proactively to avoid server-side rejection:
import time, threading
class TokenBucket:
def __init__(self, rate, capacity):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last = time.monotonic()
self.lock = threading.Lock()
def acquire(self):
with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
self.last = now
if self.tokens < 1:
wait = (1 - self.tokens) / self.rate
time.sleep(wait)
self.tokens = 0
else:
self.tokens -= 1
bucket = TokenBucket(rate=15/60, capacity=5) # 15 RPM, burst 5
def safe_chat(prompt):
bucket.acquire()
return chat_with_retry([{"role":"user","content":prompt}])
Practical Tips
- Local cache: Cache results keyed by prompt hash for direct hits.
- Batching: Merge independent requests into a single batch (supported by OpenRouter).
- Off-peak: Free tiers are busiest at peak hours; schedule heavy jobs at midnight.
- Monitor quota: Log
x-ratelimit-remaining for early warning.
Monitoring and Alerting
In production, log every request's x-ratelimit-remaining-* header to a time-series DB (e.g. Prometheus + Grafana) and alert when remaining quota drops below 20%. Also record each 429 with the provider and timestamp to pinpoint which one hit the cap. A simple approach is to add a decorator around chat_with_retry that ships exceptions and retry counts to your log service. TokenBucket is already locked, so multiple threads can share one instance; cache stampedes are solved by request coalescing (one in-flight call per prompt within 100ms).
Troubleshooting
- Still 429 after backoff: Likely daily quota exhausted. Wait until tomorrow or switch providers.
- TPM exceeded: Reduce
max_tokens or trim the message history.
- IP banned: Providers share anti-abuse signals. Always include backoff.
Good rate-limit handling multiplies the value of your free credits tenfold.
Backoff Algorithm Example
import time, random
def retry_with_backoff(call, max_retries=5, base_delay=1.0):
for i in range(max_retries):
try:
return call()
except (RateLimitError, ServerError) as e:
if i == max_retries - 1: raise
delay = base_delay * (2 ** i) + random.uniform(0, 1)
time.sleep(delay)
Best Practices
- Exponential backoff + jitter: pure exponential makes multiple clients retry in sync; add jitter to avoid the thundering herd.
- Cap backoff at 60s: a single backoff over 60s means the user has already left; further backoff is pointless.
- Switch providers after 2 consecutive 429s: do not stick with one provider once it starts throttling.
- Identify limit type from headers: the
X-RateLimit-Remaining response header tells you how much is left, distinguishing RPM vs daily quota.