速率限制与重试逻辑
指数退避、429 状态码处理以及合理使用 API。
速率限制与重试逻辑 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
什么是速率限制？
速率限制是 API 保护自身、避免过载的机制。当您的代理在短时间内发送过多请求时，API 会返回 429 Too Many Requests。常见限制包括每秒、每分钟或每天的请求数。
忽视速率限制会导致代理被阻止、API 密钥被撤销以及产生额外费用。
import requests
response = requests.get(
'https://api.example.com/data',
headers={'Authorization': 'Bearer YOUR_KEY'}
)
if response.status_code == 429:
print('Rate limit exceeded!')
# Check headers for limit details
limit = response.headers.get('X-RateLimit-Limit')
remaining = response.headers.get('X-RateLimit-Remaining')
reset = response.headers.get('X-RateLimit-Reset')
print(f'Limit: {limit}, Remaining: {remaining}, Reset: {reset}')Retry-After 标头
当 API 返回 429 时，通常会包含 Retry-After 标头，准确告知您重试前需要等待多少秒。请始终遵守此标头——忽视它并立即重试只会再次得到 429。
import requests
import time
def request_with_retry_after(url, headers):
response = requests.get(url, headers=headers)
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 60))
print(f'Rate limited. Waiting {retry_after} seconds...')
time.sleep(retry_after)
# Retry once after waiting
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()指数退避
指数退避是标准的重试策略：每次失败后等待更长时间。如果第 1 次尝试等待 2 秒，第 2 次等待 4 秒，第 3 次等待 8 秒，依此类推。这会逐步降低服务器负载，并给服务器恢复的时间。
公式：wait = 2 ** attempt
import requests
import time
def get_with_exponential_backoff(url, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 200:
return response.json()
if response.status_code in (429, 500, 502, 503):
wait = 2 ** attempt # 1, 2, 4, 8, 16 seconds
print(f'Attempt {attempt+1} failed ({response.status_code}). '
f'Waiting {wait}s before retry...')
time.sleep(wait)
else:
response.raise_for_status() # non-retryable error
raise Exception(f'Failed after {max_retries} retries')为退避添加抖动
如果许多代理同时重试（短暂中断后很常见），它们会同时唤醒——形成一个惊群，并立即再次触发速率限制。添加抖动（随机延迟）可以分散重试时间，降低服务器负载。
import requests
import time
import random
def get_with_jittered_backoff(url, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 200:
return response.json()
if response.status_code in (429, 500, 502, 503):
base_wait = 2 ** attempt
# Add random jitter: actual wait is 50%-100% of base
jitter = random.uniform(0.5, 1.0)
wait = base_wait * jitter
print(f'Waiting {wait:.1f}s (attempt {attempt+1})')
time.sleep(wait)
else:
response.raise_for_status()
raise Exception(f'Failed after {max_retries} retries')tenacity 库
tenacity 是最受欢迎的 Python 重试逻辑库。它通过简洁的装饰器语法处理指数退避、抖动、最大重试次数和自定义停止条件。它比手写的重试循环可靠得多。
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type, before_sleep_log
)
import requests
import logging
logger = logging.getLogger(__name__)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=60),
retry=retry_if_exception_type(requests.exceptions.HTTPError),
before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_data(url, headers):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 429:
response.raise_for_status() # triggers retry
response.raise_for_status()
return response.json()使用自定义重试条件的 tenacity
您可以让 tenacity 只针对特定状态码（例如 429 和 5xx）进行重试，并在不会从重试中受益的客户端错误（4xx）出现时立即停止。使用 retry_if_result 或自定义可调用对象来检查响应。
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_result
)
import requests
def is_retryable_response(response):
return response.status_code in (429, 500, 502, 503, 504)
@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=2, min=2, max=30),
retry=retry_if_result(is_retryable_response)
)
def resilient_get(url, headers):
response = requests.get(url, headers=headers, timeout=(5, 30))
return response # retry logic inspects the response object
# Usage
response = resilient_get(
'https://api.example.com/data',
{'Authorization': 'Bearer YOUR_KEY'}
)
data = response.json()主动管理速率限制
最佳策略是从一开始就避免触及速率限制。检查每个响应中的速率限制标头，并在接近限制时降低速度。许多 API 会提供 X-RateLimit-Remaining 和 X-RateLimit-Reset 标头。
import requests
import time
class RateLimitAwareClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.headers = {'Authorization': f'Bearer {api_key}'}
self.remaining = 1000 # assume generous limit
def get(self, path):
# Proactively slow down if nearly exhausted
if self.remaining < 10:
print('Rate limit nearly exhausted, sleeping 5s...')
time.sleep(5)
response = requests.get(
f'{self.base_url}{path}', headers=self.headers
)
# Update remaining from response headers
remaining_str = response.headers.get('X-RateLimit-Remaining')
if remaining_str:
self.remaining = int(remaining_str)
response.raise_for_status()
return response.json()最大重试次数与放弃重试
重试逻辑必须始终设有限制。无限重试可能造成级联故障，使所有代理都陷入重试循环。在达到 max_retries 后，请引发包含失败上下文的最终异常，以便代理记录该异常并继续处理其他工作。
import requests
import time
class MaxRetriesExceeded(Exception):
def __init__(self, url, attempts, last_status):
self.url = url
self.attempts = attempts
self.last_status = last_status
super().__init__(
f'Failed {url} after {attempts} attempts '
f'(last status: {last_status})'
)
def fetch_with_limit(url, headers, max_retries=3):
last_response = None
for attempt in range(max_retries):
last_response = requests.get(url, headers=headers)
if last_response.status_code == 200:
return last_response.json()
time.sleep(2 ** attempt)
raise MaxRetriesExceeded(url, max_retries, last_response.status_code)熔断器模式
熔断器模式可以防止代理持续冲击出现故障的服务。失败次数达到阈值后，熔断器会“打开”，所有请求都会立即失败，而不会访问网络。冷却时间结束后，它会尝试发送一个请求——如果成功，熔断器就会“关闭”，恢复正常运行。
import time
class CircuitBreaker:
CLOSED, OPEN, HALF_OPEN = 'closed', 'open', 'half_open'
def __init__(self, failure_threshold=5, recovery_timeout=60):
self.state = self.CLOSED
self.failures = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.opened_at = None
def call(self, func, *args, **kwargs):
if self.state == self.OPEN:
if time.time() - self.opened_at > self.recovery_timeout:
self.state = self.HALF_OPEN
else:
raise Exception('Circuit OPEN — service unavailable')
try:
result = func(*args, **kwargs)
self.failures = 0
self.state = self.CLOSED
return result
except Exception as e:
self.failures += 1
if self.failures >= self.failure_threshold:
self.state = self.OPEN
self.opened_at = time.time()
print(f'Circuit OPENED after {self.failures} failures')
raise
# --- demo ---
def flaky():
raise ValueError('upstream 500')
def works():
return 'ok'
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
for i in range(3):
try:
cb.call(flaky)
except Exception as e:
print(f'call {i+1} failed: {e}')
print(f'Breaker state after 3 failures: {cb.state}')
try:
cb.call(flaky)
except Exception as e:
print(f'Rejected without calling flaky(): {e}')
对请求排队以遵守限制
对于需要批量发起大量调用的代理，请使用令牌桶或基于简单休眠的限流方式，以遵守限制。根据 API 的速率限制计算调用之间的安全间隔（例如，每分钟 60 次调用 = 每秒 1 次调用）。
import requests
import time
def batch_requests(urls, headers, calls_per_minute=60):
interval = 60.0 / calls_per_minute # seconds between calls
results = []
for i, url in enumerate(urls):
start = time.time()
response = requests.get(url, headers=headers, timeout=(5, 30))
response.raise_for_status()
results.append(response.json())
print(f'Processed {i+1}/{len(urls)}')
# Sleep for remaining time in the interval
elapsed = time.time() - start
sleep_time = interval - elapsed
if sleep_time > 0:
time.sleep(sleep_time)
return results将重试逻辑与退避标头结合
最稳健的模式是将服务器指定的等待时间（Retry-After）与指数退避结合起来，后者作为备用方案。只要服务器提供了指导，就应始终优先遵循服务器的建议——服务器最清楚何时可以再次重试。
import requests
import time
import random
def smart_retry(url, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 200:
return response.json()
if response.status_code == 429:
# Use Retry-After if provided, else exponential backoff
retry_after = response.headers.get('Retry-After')
if retry_after:
wait = int(retry_after)
else:
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'429 rate limit. Waiting {wait:.1f}s...')
time.sleep(wait)
elif response.status_code >= 500:
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'Server error {response.status_code}. Waiting {wait:.1f}s...')
time.sleep(wait)
else:
response.raise_for_status() # non-retryable
raise Exception(f'Gave up after {max_retries} attempts')快速检查：指数退避
测试您对重试策略的理解。
速率限制与重试回顾
现在，您的代理已经能够从容处理速率限制：
- 429 Too Many Requests——遵守
Retry-After标头；等待后再重试 - 指数退避——
wait = 2^attempt会使每次重试的等待时间加倍 - 抖动——增加随机性，将多个代理实例的重试分散开
- tenacity——通过装饰器和清晰的配置处理所有重试逻辑
- 熔断器——达到阈值后停止持续冲击故障服务
- 主动限流——检查
X-RateLimit-Remaining，在触及限制前降低速度
常见问题解答
「速率限制与重试逻辑」课时是免费的吗？
是的 — 「速率限制与重试逻辑」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「速率限制与重试逻辑」这节课中我会学到什么？
指数退避、429 状态码处理以及合理使用 API。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「速率限制与重试逻辑」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。