レート制限とリトライロジック
指数バックオフ、429エラーへの対応、APIを適切に利用する方法を学びます。
「レート制限とリトライロジック」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
レート制限とは
レート制限とは、APIが過負荷から自身を保護する仕組みです。エージェントが短時間に多くのリクエストを送ると、APIは429 Too Many Requestsを返します。一般的な制限には、1秒、1分、または1日あたりのリクエスト数があります。
レート制限を無視すると、エージェントのブロック、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を使用する
特定のステータスコード（429や5xxなど）の場合だけ再試行し、再試行しても意味がないクライアントエラー（4xx）では直ちに停止するよう、tenacityに設定できます。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)サーキットブレーカー・パターン
サーキットブレーカーパターンは、エージェントが障害中のサービスにリクエストを送り続けるのを防ぎます。一定回数の障害が発生すると、サーキットが「開いた」状態になり、ネットワークに接続せずにすべてのリクエストを直ちに失敗させます。クールダウン期間が過ぎると1件のリクエストを試し、成功すればサーキットが閉じて通常の動作に戻ります。
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秒に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により、再試行するたびに待ち時間を2倍にする - ジッター — 複数のエージェントインスタンス間で再試行を分散するためにランダム性を加える
- tenacity — デコレーターと簡潔な設定で、すべての再試行ロジックを処理する
- サーキットブレーカー — 一定回数の障害後、障害中のサービスへのリクエスト送信を停止する
- プロアクティブなスロットリング —
X-RateLimit-Remainingを確認し、制限に達する前に速度を落とす
よくある質問
「レート制限とリトライロジック」レッスンは無料ですか？
はい。「レート制限とリトライロジック」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「レート制限とリトライロジック」で何を学びますか？
指数バックオフ、429エラーへの対応、APIを適切に利用する方法を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「レート制限とリトライロジック」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- エージェント開発者のためのREST API基礎
- 認証：API KeysとOAuth
- APIレスポンスとエラーの処理
- レート制限とリトライロジック