レート制限

Rustellar プラットフォームでの API レート制限の理解と管理。

概要

レート制限は、特定の時間枠内で行うことができる API リクエストの数を制御します。これにより、すべてのユーザーに対して公平な使用とプラットフォームの安定性が保証されます。

レート制限ティア

無料ティア

制限タイプ	値
毎分リクエスト数 (RPM)	20
毎分トークン数 (TPM)	40,000
毎日リクエスト数 (RPD)	200

スタンダードティア

制限タイプ	値
毎分リクエスト数 (RPM)	60
毎分トークン数 (TPM)	150,000
毎日リクエスト数 (RPD)	10,000

プロティア

制限タイプ	値
毎分リクエスト数 (RPM)	300
毎分トークン数 (TPM)	500,000
毎日リクエスト数 (RPD)	100,000

エンタープライズティア

お客様の要件に基づいたカスタム制限。詳細については営業チームにお問い合わせください。

レート制限ヘッダー

すべての API レスポンスには、現在のレート制限ステータスを示すヘッダーが含まれます:

X-RateLimit-Limit-Requests: 60
X-RateLimit-Limit-Tokens: 150000
X-RateLimit-Remaining-Requests: 45
X-RateLimit-Remaining-Tokens: 125000
X-RateLimit-Reset-Requests: 2024-12-14T12:35:00Z
X-RateLimit-Reset-Tokens: 2024-12-14T12:35:00Z

ヘッダーの説明

ヘッダー	説明
X-RateLimit-Limit-Requests	毎分許可される最大リクエスト数
X-RateLimit-Limit-Tokens	毎分許可される最大トークン数
X-RateLimit-Remaining-Requests	現在のウィンドウで残っているリクエスト数
X-RateLimit-Remaining-Tokens	現在のウィンドウで残っているトークン数
X-RateLimit-Reset-Requests	リクエスト制限がリセットされる時刻
X-RateLimit-Reset-Tokens	トークン制限がリセットされる時刻

レート制限超過レスポンス

レート制限を超えると、429 ステータスコードが返されます:

{
  "error": {
    "message": "Rate limit exceeded. Please try again in 30 seconds.",
    "type": "rate_limit_error",
    "code": "rate_limit_exceeded"
  }
}

レート制限の処理

基本的なリトライロジック

import requests
import time

def make_request_with_retry(url, headers, data, max_retries=3):
    """リトライロジック付きAPIリクエスト"""
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data)

            # 成功した場合はレスポンスを返す
            if response.status_code == 200:
                return response.json()

            # レート制限エラーの場合
            if response.status_code == 429:
                # Retry-Afterヘッダーから待機時間を取得
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                time.sleep(retry_after)
                continue

            # その他のエラー
            response.raise_for_status()

        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # エクスポネンシャルバックオフ
            else:
                raise

    return None

# 使用例
result = make_request_with_retry(
    "https://api.rustellar.com/v1/chat/completions",
    headers={
        "Authorization": "Bearer YOUR_API_KEY",
        "Content-Type": "application/json"
    },
    data={
        "model": "helix-v1",
        "messages": [{"role": "user", "content": "Hello!"}]
    }
)

トークンバケット方式を使用した高度なレート制限

import time
from collections import deque

class RateLimiter:
    """トークンバケット方式のレート制限クラス"""

    def __init__(self, max_requests_per_minute):
        """
        Args:
            max_requests_per_minute: 1分あたりの最大リクエスト数
        """
        self.max_requests = max_requests_per_minute
        self.window = 60  # 1分 = 60秒
        self.requests = deque()

    def can_make_request(self):
        """リクエストが可能かチェック"""
        now = time.time()

        # 古いリクエスト記録を削除(1分以上前)
        while self.requests and now - self.requests[0] > self.window:
            self.requests.popleft()

        # リクエスト数が上限未満かチェック
        return len(self.requests) < self.max_requests

    def wait_if_needed(self):
        """必要に応じて待機"""
        while not self.can_make_request():
            # 最も古いリクエストが期限切れになるまで待機
            oldest_request = self.requests[0]
            wait_time = self.window - (time.time() - oldest_request)
            if wait_time > 0:
                print(f"Rate limit reached. Waiting {wait_time:.1f} seconds...")
                time.sleep(wait_time + 0.1)  # 少し余裕を持たせる

    def record_request(self):
        """リクエストを記録"""
        self.requests.append(time.time())

# 使用例
limiter = RateLimiter(max_requests_per_minute=60)

def make_controlled_request(messages):
    """レート制限を考慮したリクエスト"""
    # 必要に応じて待機
    limiter.wait_if_needed()

    # リクエストを実行
    response = requests.post(
        "https://api.rustellar.com/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "helix-v1",
            "messages": messages
        }
    )

    # リクエストを記録
    limiter.record_request()

    return response.json()

# 複数のリクエストを実行
for i in range(100):
    result = make_controlled_request([
        {"role": "user", "content": f"Question {i}"}
    ])
    print(f"Request {i}: {result['choices'][0]['message']['content'][:50]}...")

レート制限ステータスの監視

def check_rate_limit_status(response):
    """レート制限の状態を確認"""
    headers = response.headers

    remaining_requests = int(headers.get('X-RateLimit-Remaining-Requests', 0))
    remaining_tokens = int(headers.get('X-RateLimit-Remaining-Tokens', 0))

    # 警告を表示(残り20%以下の場合)
    if remaining_requests < 12:  # 60 * 0.2 = 12
        print(f"⚠️ Warning: Only {remaining_requests} requests remaining!")

    if remaining_tokens < 30000:  # 150000 * 0.2 = 30000
        print(f"⚠️ Warning: Only {remaining_tokens} tokens remaining!")

    return {
        "remaining_requests": remaining_requests,
        "remaining_tokens": remaining_tokens,
        "reset_time": headers.get('X-RateLimit-Reset-Requests')
    }

# 使用例
response = requests.post(
    "https://api.rustellar.com/v1/chat/completions",
    headers={
        "Authorization": "Bearer YOUR_API_KEY",
        "Content-Type": "application/json"
    },
    json={
        "model": "helix-v1",
        "messages": [{"role": "user", "content": "Hello!"}]
    }
)

status = check_rate_limit_status(response)
print(f"Rate limit status: {status}")

ベストプラクティス

1. 失敗したリクエストを再試行する際にエクスポネンシャルバックオフを実装
2. レート制限ヘッダーを監視して制限に達することを回避
3. 可能な限りリクエストをバッチ処理して API 呼び出しを減らす
4. 頻繁にリクエストされるコンテンツをキャッシュ
5. 一貫してレート制限に達する場合はティアをアップグレード
6. 異なる時間枠に負荷を分散

制限の増加

より高いレート制限が必要な場合:

1. プランをアップグレードしてスタンダードまたはプロティアに
2. 営業チームに連絡してエンタープライズカスタム制限を取得
3. ユースケースの詳細を提供して制限増加リクエストを行う
4. 使用パターンを実証して高い制限を正当化

よくある問題

問題: 大きな会話でトークン制限に達する

解決策: 長いチャット履歴の会話要約または整理を実装します。

def trim_conversation(messages, max_tokens=3000):
    """会話履歴をトークン数に基づいてトリミング"""
    # 簡易的な実装(実際にはトークナイザーを使用)
    estimated_tokens = sum(len(m['content']) // 4 for m in messages)

    if estimated_tokens > max_tokens:
        # システムメッセージと最新のメッセージのみ保持
        system_messages = [m for m in messages if m['role'] == 'system']
        recent_messages = messages[-10:]  # 最新10メッセージ

        return system_messages + recent_messages

    return messages

問題: 同時リクエストが制限を超える

解決策: レート制限付きリクエストキューを使用します。

import asyncio
from asyncio import Semaphore

class AsyncRateLimiter:
    """非同期リクエスト用のレート制限"""

    def __init__(self, max_concurrent=5, requests_per_minute=60):
        self.semaphore = Semaphore(max_concurrent)
        self.limiter = RateLimiter(requests_per_minute)

    async def make_request(self, session, messages):
        """レート制限付き非同期リクエスト"""
        async with self.semaphore:
            self.limiter.wait_if_needed()

            # 実際のリクエスト(aiohttp使用を想定)
            async with session.post(
                "https://api.rustellar.com/v1/chat/completions",
                headers={
                    "Authorization": "Bearer YOUR_API_KEY",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "helix-v1",
                    "messages": messages
                }
            ) as response:
                self.limiter.record_request()
                return await response.json()

次のステップ

• 入力フォーマットについて学ぶ
• 出力フォーマットを確認
• 料金ティアを確認