APIを使ってみたいけれど「429 Too Many Requests」というエラーが出て困っている方は多いのではないでしょうか。今日はこの問題を解決するリクエストキューと優先順位スケジューリングについて、ゼロから丁寧に説明します。
速率限制（Rate Limit）とは何か
API提供服务者はサーバーに負担をかけないよう、一定時間内のリクエスト回数に上限を設けていません。この上限のことを「速率制限」と呼びます。Gemini APIの場合、通常は1分間に60回や100回という制限があります。
この制限を超えると、APIは429エラーを返してきます。ただし、心配必要はありません。適切な仕組みを作れば、このエラーを適切に処理できます。
リクエストキューの基本的な仕組み
リクエストキューとは、要するに到着したリクエストを一時的に保管しておく箱のようなものです。APIが込み合っている時は、ここで順番を待ち、 空いてきた時に順番に処理していきます。
Pythonでの基本的な実装
import time
import queue
import threading
from dataclasses import dataclass, field
from typing import Optional
import requests
@dataclass
class APIRequest:
"""APIリクエストを保持するデータクラス"""
prompt: str
priority: int = 5 # 優先度（1が最高、10が最低）
timestamp: float = field(default_factory=time.time)
retry_count: int = 0
class SimpleRequestQueue:
"""基本的なリクエストキュー"""
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.api_key = api_key
self.request_queue = queue.Queue()
self.max_retries = 3
self.retry_delay = 2.0 # 秒
def add_request(self, prompt: str, priority: int = 5):
"""新しいリクエストをキューに追加"""
request = APIRequest(prompt=prompt, priority=priority)
self.request_queue.put(request)
print(f"リクエスト追加完了: 優先度{priority}, キュー内{self.request_queue.qsize()}件")
def process_request(self, api_request: APIRequest) -> Optional[dict]:
"""单个リクエストを処理"""
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
data = {
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": api_request.prompt}],
"max_tokens": 1000
}
try:
response = requests.post(url, headers=headers, json=data, timeout=30)
if response.status_code == 429:
# 速率制限エラーの場合
if api_request.retry_count < self.max_retries:
api_request.retry_count += 1
wait_time = self.retry_delay * api_request.retry_count
print(f"速率制限発生、{wait_time}秒後に再試行（{api_request.retry_count}回目）")
time.sleep(wait_time)
return self.process_request(api_request)
else:
print(f"最大リトライ回数を超過: {api_request.prompt[:30]}...")
return None
elif response.status_code == 200:
return response.json()
else:
print(f"エラー発生: {response.status_code}")
return None
except requests.exceptions.RequestException as e:
print(f"リクエスト失敗: {e}")
return None
def process_queue(self):
"""キュー内の全リクエストを処理"""
processed = 0
while not self.request_queue.empty():
request = self.request_queue.get()
result = self.process_request(request)
if result:
processed += 1
print(f"処理成功 [{processed}]件")
time.sleep(0.5) # API負荷軽減のための待機
print(f"キュー処理完了: {processed}件成功")
使用例
if __name__ == "__main__":
queue_handler = SimpleRequestQueue(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# リクエストを追加
queue_handler.add_request("今日の天気を教えて", priority=1)
queue_handler.add_request("簡単な計算の仕方を教えて", priority=3)
queue_handler.add_request("おすすめの映画は何ですか", priority=5)
# キューを処理
queue_handler.process_queue()優先順位スケジューリングの実装
キューだけでは十分ではありません。例えば緊急で大きな仕事をしたい時、優先度の高いリクエストを先に処理したいですよね。この問題を解決するのが「優先順位スケジューリング」です。
優先順位付きキューの実装
import heapq
import time
import threading
from enum import IntEnum
from typing import List, Dict, Any, Optional
import requests
class Priority(IntEnum):
"""リクエスト優先度の定義"""
CRITICAL = 1 # 最優先（システム重要処理）
HIGH = 3 # 高優先度（ユーザー直接操作）
NORMAL = 5 # 通常（標準処理）
LOW = 7 # 低優先度（バックグラウンド処理）
BULK = 10 # 一括処理（分析など）
class PriorityRequestQueue:
"""優先順位付きリクエストキュー（スレッドセーフ）"""
def __init__(self, base_url: str, api_key: str, max_concurrent: int = 5):
self.base_url = base_url
self.api_key = api_key
self.max_concurrent = max_concurrent
self._queue: List[tuple] = [] # (priority, timestamp, request_id, data)
self._lock = threading.Lock()
self._request_id_counter = 0
self._processed_results: Dict[int, Any] = {}
# 速率制限追跡
self._request_timestamps: List[float] = []
self._rate_limit_window = 60 # 1分window
self._max_requests_per_window = 50 # 1分あたり最大50リクエスト
def _get_next_request_id(self) -> int:
"""一意のリクエストIDを生成"""
with self._lock:
self._request_id_counter += 1
return self._request_id_counter
def _check_rate_limit(self) -> float:
"""速率制限を確認し、待機時間を返す"""
current_time = time.time()
# 古いタイムスタンプを削除
self._request_timestamps = [
ts for ts in self._request_timestamps
if current_time - ts < self._rate_limit_window
]
if len(self._request_timestamps) >= self._max_requests_per_window:
# 最も古いリクエストからの経過時間を計算
oldest = min(self._request_timestamps)
wait_time = self._rate_limit_window - (current_time - oldest)
return max(0, wait_time)
return 0
def enqueue(self, prompt: str, priority: int = Priority.NORMAL,
metadata: Optional[Dict] = None) -> int:
"""リクエストをキューに追加"""
request_id = self._get_next_request_id()
timestamp = time.time()
request_data = {
"prompt": prompt,
"metadata": metadata or {}
}
with self._lock:
heapq.heappush(
self._queue,
(priority, timestamp, request_id, request_data)
)
print(f"追加: ID={request_id}, 優先度={priority}, キュー内={len(self._queue)}件")
return request_id
def dequeue(self) -> Optional[tuple]:
"""最高優先度のリクエストを取り出す"""
with self._lock:
if self._queue:
return heapq.heappop(self._queue)
return None
def size(self) -> int:
"""キューサイズを返す"""
with self._lock:
return len(self._queue)
def process_next(self) -> Optional[Dict[str, Any]]:
"""次のリクエストを処理"""
# 速率制限チェック
wait_time = self._check_rate_limit()
if wait_time > 0:
print(f"速率制限により{wait_time:.1f}秒待機...")
time.sleep(wait_time)
request = self.dequeue()
if not request:
return None
priority, timestamp, request_id, data = request
# API呼び出し
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": data["prompt"]}],
"temperature": 0.7,
"max_tokens": 2000
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=60)
# 成功した場合、タイムスタンプを記録
if response.status_code == 200:
self._request_timestamps.append(time.time())
result = {
"request_id": request_id,
"priority": priority,
"status_code": response.status_code,
"response": response.json() if response.status_code == 200 else None,
"error": response.text if response.status_code != 200 else None
}
self._processed_results[request_id] = result
if response.status_code == 200:
print(f"✓ 完了: ID={request_id}, 優先度={priority}")
else:
print(f"✗ 失敗: ID={request_id}, ステータス={response.status_code}")
return result
except Exception as e:
error_result = {
"request_id": request_id,
"priority": priority,
"status_code": None,
"response": None,
"error": str(e)
}
self._processed_results[request_id] = error_result
print(f"✗ 例外: ID={request_id}, エラー={e}")
return error_result
def process_all(self, progress_interval: int = 5):
"""全リクエストを処理（進捗表示付き）"""
total = self.size()
processed = 0
succeeded = 0
print(f"\n{'='*50}")
print(f"処理開始: {total}件のリクエスト")
print(f"{'='*50}\n")
while self.size() > 0:
result = self.process_next()
if result:
processed += 1
if result.get("status_code") == 200:
succeeded += 1
if processed % progress_interval == 0:
print(f"進捗: {processed}/{total}件 ({succeeded}件成功)")
# APIへの負荷を考慮して小さな待機
time.sleep(0.3)
print(f"\n{'='*50}")
print(f"処理完了: {processed}件中{succeeded}件成功")
print(f"{'='*50}")
return self._processed_results
使用例
if __name__ == "__main__":
# HolySheep APIを使用（¥1=$1、超低コスト）
queue = PriorityRequestQueue(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=3
)
# 様々な優先度でリクエストを追加
queue.enqueue("至急！サーバーの状態を確認して", Priority.CRITICAL)
queue.enqueue("ユーザーのログイン処理", Priority.HIGH)
queue.enqueue("日次レポートの生成", Priority.NORMAL)
queue.enqueue("過去のログ分析", Priority.LOW)
queue.enqueue("類似データの批量検索（100件）", Priority.BULK)
# 全リクエストを処理
results = queue.process_all()
# 結果の確認
print("\n処理結果サマリー:")
for req_id, result in results.items():
status = "✓" if result["status_code"] == 200 else "✗"
print(f" {status} ID={req_id}, 優先度={result['priority']}")指数バックオフによるリトライ戦略
APIが込み合っている時、ただ待って再試行するだけでは非効率です。「指数バックオフ」という手法を使えば、待つ時間を徐々に長くして、無駄なく再試行できます。
import random
import time
from typing import Callable, Any, Optional
import functools
def exponential_backoff(
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True
):
"""
指数バックオフデコレーター
Args:
max_retries: 最大リトライ回数
base_delay: 初期待機時間（秒）
max_delay: 最大待機時間（秒）
exponential_base: 指数の底
jitter: ランダム要素を追加するか
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt == max_retries:
print(f"最大リトライ回数（{max_retries}回）に到達。断念します。")
raise last_exception
# 指数的に待機時間を計算
delay = min(
base_delay * (exponential_base ** attempt),
max_delay
)
# ランダム要素（コンプリティブ衝突を避ける）
if jitter:
delay = delay * (0.5 + random.random())
print(f"試行 {attempt + 1} 回目失敗: {e}")
print(f"{delay:.2f}秒後に再試行...")
time.sleep(delay)
raise last_exception
return wrapper
return decorator
使用例
class RobustAPIClient:
"""堅牢なAPIクライアント（指数バックオフ実装）"""
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
@exponential_backoff(max_retries=5, base_delay=1.0, max_delay=30.0)
def call_api_with_backoff(self, prompt: str, model: str = "gemini-2.5-flash") -> dict:
"""指数バックオフ付きでAPIを呼び出す"""
url = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1500
}
response = self.session.post(url, json=payload, timeout=60)
if response.status_code == 429:
# 速率制限エラー
retry_after = response.headers.get('Retry-After', None)
if retry_after:
print(f"サーバーからの指示: {retry_after}秒後に再試行")
time.sleep(int(retry_after))
raise Exception("Rate limit exceeded")
elif response.status_code >= 500:
# サーバーエラーはリトライ対象
raise Exception(f"Server error: {response.status_code}")
elif response.status_code != 200:
raise Exception(f"API error: {response.status_code}, {response.text}")
return response.json()
実用的な例
if __name__ == "__main__":
client = RobustAPIClient(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
prompts = [
"夏の 인기�� recipesについて教えてください",
"AIの未来について、あなたの考えを聞かせてください",
"効果的な学习方法を知りたいです"
]
print("一括リクエスト処理（指数バックオフ付き）\n")
for i, prompt in enumerate(prompts, 1):
print(f"\n[{i}/{len(prompts)}] 処理中: {prompt[:30]}...")
try:
result = client.call_api_with_backoff(prompt)
print(f"✓ 成功: 応答トークン数={result.get('usage', {}).get('total_tokens', 'N/A')}")
except Exception as e:
print(f"✗ 最終エラー: {e}")
finally:
time.sleep(1) # リクエスト間の最小間隔HolySheep AI 活用のヒント
ここで、私が実際にHolySheep AIを今すぐ登録して気づいたメリットをお伝えします。
- 驚異的なコスト効率：Gemini 2.5 Flashが$2.50/MTokという破格の安さ。私が以前使っていたサービスとは比べ物になりません。
- 超低レイテンシ：平均50ms未満の応答速度で、リアルタイムアプリケーションにも十分使えます。
- シンプルな決済：WeChat PayやAlipayに対応しているので、日本の私も困ることはありません。
よくあるエラーと対処法
エラー1：429 Too Many Requests が出る
原因：一定時間内のリクエスト回数が上限を超えている。
解決方法：リクエスト間に適切な待機時間を入れましょう。
# 悪い例（すぐに上限に達する）
for prompt in prompts:
response = call_api(prompt) # 連続呼び出し
良い例（待機時間を入れる）
for prompt in prompts:
response = call_api(prompt)
time.sleep(1.2) # 1.2秒待機（1分あたり50リクエストの場合）
# または
await asyncio.sleep(1.2) # 非同期版エラー2：401 Unauthorized / 403 Forbidden
原因：APIキーが無効、有効期限切れ、または権限不足。
解決方法：APIキーの確認と正しいフォーマットで送信。
# APIキー確認と正しいヘッダー設定
headers = {
"Authorization": f"Bearer {api_key}", # "Bearer "を忘れない
"Content-Type": "application/json"
}
キーの先頭と末尾に空白が入っていないか確認
api_key = api_key.strip()
環境変数から読み込む方法（推奨）
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY", "")
if not api_key:
raise ValueError("APIキーが設定されていません")エラー3：Timeout が頻発する
原因：リクエスト処理に時間がかかりすぎている、またはネットワーク問題。
解決方法：タイムアウト設定の調整と非同期処理の導入。
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
タイムアウトとリトライを設定したセッション
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
適切なタイムアウト設定
response = session.post(
url,
json=payload,
headers=headers,
timeout=(10, 60) # (接続タイムアウト, 読み取りタイムアウト)
)エラー4：応答が切れる・不完全な応答が返る
原因：max_tokens値が小さすぎる、またはネットワーク切断。
解決方法：max_tokens値を適切に増加させる。
# 十分なトークン数を確保
payload = {
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 4000, # 必要に応じて増加（料金に影響あり）
"temperature": 0.7
}
応答が途中で切れた場合の処理
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
# 応答が不完全な可能性がある場合のチェック
if result.get("usage", {}).get("completion_tokens", 0) >= 3900:
print("警告: 応答が最大トークン数に近いかも知れません")
# 続きを要求するプロンプトを送信
continuation = call_api(
f"前の応答の続きを詳しく説明してください: {content[-200:]}"
)まとめ
今日の内容をまとめましょう。
- リクエストキューを使って、リクエストを整理された形で管理できる
- 優先順位スケジューリングで、重要なリクエストを先に処理できる
- 指数バックオフで、API制限時に効率的に再試行できる
- 適切なエラー処理で、安定性の高いシステムを構築できる
これらのテクニックを組み合わせれば、速率制限に困ることはありません。むしろ、HolySheep AIの今すぐ登録して、コストを最小限に抑えながら効率的なAIアプリケーションを構築しましょう！
HolySheep AIでは、Gemini 2.5 Flashが$2.50/MTokという圧倒的なコストパフォーマンスで、私のプロジェクトコストが劇的に下がりました。登録するだけで無料クレジットも貰えるので、まずは試してみることをおすすめします。
👉 HolySheep AI に登録して無料クレジットを獲得