0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@thayate

watsonx.ai Model Gatewayでレート制限を実装

0
Posted at

はじめに

watsonx.ai Model Gatewayは、複数のAIモデルを統一的に管理・利用できるサービスです。本記事では、公式SDKを使用してRate Limitを実装する方法を、実践的なコード例とともに解説します。

対象読者

  • watsonx.aiを使用している開発者
  • AIモデルのAPI利用にコスト管理が必要な方
  • Rate Limitの実装方法を学びたい方

Model Gatewayの設定方法については以下の記事を参考にしました。
https://qiita.com/YokeSin/items/3c4a8ce0c9b5b5e9e3ce

Model Gatewayが解決する課題

watsonx.ai Model Gatewayは、以下のような課題を持つ方々に最適です：

  • 複数AIプロバイダーの管理: OpenAI、Anthropic、Meta（Llama）など、複数のプロバイダーを統一APIで管理
  • コスト管理: Rate Limit機能でリクエスト数・トークン数を制御し、予算超過を防止
  • セキュリティ: 中央集権的なアクセス制御とAPIキー管理でセキュリティリスクを低減

特に、開発環境でのコスト制御や部門別のリソース管理に有効です。

watsonx.ai Model Gatewayは2026年3月現在プレビュー版のみでの提供となっており、正式リリース時に機能に変更がある場合がございます。0

環境

本記事で使用した環境は以下の通りです：

システム環境

  • OS: macOS Sequoia
  • Python: 3.11.11

パッケージバージョン

  • ibm-watsonx-ai: 1.5.3
  • python-dotenv: 1.0.0
  • pandas: 2.1.4

必要なパッケージのインストール

pip install ibm-watsonx-ai python-dotenv

注意: Python 3.8以上が必要です。本記事のコードはPython 3.11.11で動作確認済みです。

バージョン確認方法

# Pythonバージョン確認
python --version

# パッケージバージョン確認
pip show ibm-watsonx-ai
pip show python-dotenv

環境変数の設定

.envファイルを作成し、以下の情報を設定します：

WATSONX_API_KEY=your_api_key_here
WATSONX_URL=https://ca-tor.ml.cloud.ibm.com
WATSONX_PROJECT_ID=your_project_id_here

Rate Limitの基本概念

watsonx.ai Model GatewayのRate Limitはトークンバケットアルゴリズムを使用しています。

主要なパラメータ

  • capacity: バケットの最大容量（同時に保持できるトークン数）
  • amount: 補充されるトークン数
  • duration: トークン補充の間隔（例：1m = 1分）

動作の仕組み

  1. 初期状態でバケットにcapacity個のトークンが入っている
  2. リクエストごとに1トークンを消費
  3. durationごとにamount個のトークンが補充される
  4. トークンが0になるとRate Limitが発動（HTTP 429エラー）

RateLimitsクラスの詳細

RateLimitsクラスは、watsonx.ai Model GatewayのRate Limitを管理するための公式クラスです。

クラスの初期化

from ibm_watsonx_ai import APIClient
from ibm_watsonx_ai.gateway.rate_limits import RateLimits

rate_limits = RateLimits(api_client=client)

主要なメソッド

1. create_for_model()

特定のモデルに対してRate Limitを設定します。

result = rate_limits.create_for_model(
    model_id: str,
    request: RateLimitSettings | None = None,
    token: RateLimitSettings | None = None
) -> dict

パラメータ:

  • model_id (str): Model GatewayのモデルID
  • request (RateLimitSettings, optional): リクエスト数のRate Limit設定
  • token (RateLimitSettings, optional): トークン数のRate Limit設定

戻り値:

  • dict: Rate Limitの詳細情報（IDを含む）

使用例:

result = rate_limits.create_for_model(
    model_id="your-model-id",
    request=RateLimitSettings(amount=5, capacity=5, duration="1m"),
    token=RateLimitSettings(amount=50000, capacity=50000, duration="1m")
)
rate_limit_id = result.get('uuid') or result.get('id')

2. create_for_provider()

特定のプロバイダーに対してRate Limitを設定します。

result = rate_limits.create_for_provider(
    provider_id: str,
    request: RateLimitSettings | None = None,
    token: RateLimitSettings | None = None
) -> dict

3. create_for_tenant()

テナント全体に対してRate Limitを設定します。

result = rate_limits.create_for_tenant(
    request: RateLimitSettings | None = None,
    token: RateLimitSettings | None = None
) -> dict

4. list()

登録されているRate Limitの一覧を取得します。

df = rate_limits.list() -> pandas.DataFrame

戻り値:

  • pandas.DataFrame: Rate Limitの一覧

使用例:

existing_limits = rate_limits.list()
print(existing_limits)

5. get_details()

Rate Limitの詳細情報を取得します。

details = rate_limits.get_details(
    rate_limit_id: str | None = None
) -> dict

パラメータ:

  • rate_limit_id (str, optional): Rate LimitのID。指定しない場合は全てのRate Limitの詳細を取得

6. delete()

Rate Limitを削除します。

status = rate_limits.delete(rate_limit_id: str) -> Literal["SUCCESS"]

パラメータ:

  • rate_limit_id (str): 削除するRate LimitのID

戻り値:

  • "SUCCESS": 削除が成功した場合

使用例:

rate_limits.delete("your-rate-limit-id")

7. update_for_model()

モデルのRate Limitを更新します。

result = rate_limits.update_for_model(
    rate_limit_id: str,
    model_id: str,
    request: RateLimitSettings | None = None,
    token: RateLimitSettings | None = None
) -> dict

8. update_for_provider()

プロバイダーのRate Limitを更新します。

result = rate_limits.update_for_provider(
    rate_limit_id: str,
    provider_id: str,
    request: RateLimitSettings | None = None,
    token: RateLimitSettings | None = None
) -> dict

9. update_for_tenant()

テナントのRate Limitを更新します。

result = rate_limits.update_for_tenant(
    rate_limit_id: str,
    request: RateLimitSettings | None = None,
    token: RateLimitSettings | None = None
) -> dict

静的メソッド

get_id()

Rate Limitの詳細情報からIDを取得します。

rate_limit_id = RateLimits.get_id(rate_limit_details: dict) -> str

RateLimitSettingsクラスの詳細

RateLimitSettingsは、Rate Limitの設定を定義するTypedDictクラスです。

パラメータ

from ibm_watsonx_ai.gateway.rate_limits import RateLimitSettings

settings = RateLimitSettings(
    amount: int,      # 必須
    capacity: int,    # 必須
    duration: str     # 必須
)

amount (int)

補充間隔ごとにバケットに追加されるトークンの数。

例:

  • amount=5: 1分ごとに5トークンが補充される（duration="1m"の場合）

capacity (int)

バケットが保持できる最大トークン数。

例:

  • capacity=10: 最大10リクエストまで同時に保持可能

duration (str)

トークン補充の間隔。Go言語のduration形式で指定します。

例:

  • "1m": 1分
  • "30s": 30秒
  • "1h": 1時間
  • "1h30m": 1時間30分

参考: Go time.ParseDuration

使用例

# 1分あたり5リクエスト
request_limit = RateLimitSettings(
    amount=5,
    capacity=5,
    duration="1m"
)

# 1分あたり50000トークン
token_limit = RateLimitSettings(
    amount=50000,
    capacity=50000,
    duration="1m"
)

# 30秒あたり3リクエスト
custom_limit = RateLimitSettings(
    amount=3,
    capacity=3,
    duration="30s"
)

requestとtokenの違い

  • request: リクエスト数の制限（APIコール回数）
  • token: トークン数の制限（処理されるトークン数）

通常、両方を設定することで、リクエスト数とトークン数の両方を制御できます。

rate_limits.create_for_model(
    model_id=model_id,
    request=RateLimitSettings(amount=5, capacity=5, duration="1m"),      # 5リクエスト/分
    token=RateLimitSettings(amount=50000, capacity=50000, duration="1m")  # 50000トークン/分
)

実装例1: シンプルなチャットアプリケーション

Rate Limit付きでLlamaモデルとチャットするアプリケーションです。
スクリプトはIBM Bobを使用して実装しました。
Model GatewayではOpenAIやGoogleなど様々なプロバイダーのAPIを管理することができるのですが、今回はRate Limitの実装にフォーカスするため、あえてwatsonx.aiのモデルを使用しました。

"""
watsonx.ai Model Gateway チャットアプリケーション
Rate Limit付きでLlamaモデルとチャット
"""

import os
import time
from dotenv import load_dotenv

from ibm_watsonx_ai import Credentials, APIClient
from ibm_watsonx_ai.gateway import Gateway
from ibm_watsonx_ai.gateway.rate_limits import RateLimits, RateLimitSettings

load_dotenv()

# 設定
MODEL_NAME = "meta-llama/llama-3-3-70b-instruct"
 = 5


def main():
    print("\n" + "="*60)
    print("watsonx.ai Model Gateway チャットアプリ")
    print("="*60)
    
    # 初期化
    credentials = Credentials(
        api_key=os.getenv("WATSONX_API_KEY"),
        url=os.getenv("WATSONX_URL")
    )
    client = APIClient(credentials=credentials)
    client.set.default_project(os.getenv("WATSONX_PROJECT_ID"))
    gateway = Gateway(api_client=client)
    
    # モデルIDを取得
    models_df = gateway.models.list()
    print(f"\n利用可能なモデル:")
    print(models_df[["MODEL", "TYPE"]].to_string())
    
    model_row = models_df[models_df['MODEL'] == MODEL_NAME]
    if model_row.empty:
        print(f"\nエラー: モデル '{MODEL_NAME}' が見つかりません")
        return
    
    gateway_model_id = model_row.iloc[0]['ID']
    print(f"\n使用モデル: {MODEL_NAME}")
    print(f"Rate Limit: {REQUESTS_PER_MINUTE}リクエスト/分")
    
    # 既存のRate Limitを削除
    rate_limits = RateLimits(api_client=client)
    existing = rate_limits.list()
    if not existing.empty:
        for _, row in existing.iterrows():
            limit_id = row.get('ID') or row.get('id') or row.get('uuid')
            if limit_id:
                rate_limits.delete(limit_id)
    
    # Rate Limitを設定
    result = rate_limits.create_for_model(
        model_id=gateway_model_id,
        request=RateLimitSettings(
            amount=REQUESTS_PER_MINUTE,
            capacity=REQUESTS_PER_MINUTE,
            duration="1m"
        ),
        token=RateLimitSettings(
            amount=50000,
            capacity=50000,
            duration="1m"
        )
    )
    
    rate_limit_id = result.get('uuid') or result.get('id')
    print(f"✓ Rate Limit設定完了 (ID: {rate_limit_id})")
    
    # Rate Limitが反映されるまで待機
    print("\n待機中...")
    time.sleep(5)
    
    # チャット実行
    print("\n" + "="*60)
    print("チャット実行")
    print("="*60)
    
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant that answers in Japanese."
        },
        {
            "role": "user",
            "content": "テストです。短く自己紹介してください。"
        }
    ]
    
    try:
        response = gateway.chat.completions.create(
            model=MODEL_NAME,
            messages=messages
        )
        
        answer = response["choices"][0]["message"]["content"]
        print(f"\nアシスタント:\n{answer}")
        print("\n✓ チャット成功")
        
    except Exception as e:
        print(f"\n✗ エラー: {e}")
    
    # クリーンアップ
    print("\n" + "="*60)
    cleanup = input("Rate Limit設定を削除しますか? (y/N): ")
    if cleanup.lower() == 'y':
        rate_limits.delete(rate_limit_id)
        print("✓ 削除完了")
    
    print("="*60 + "\n")


if __name__ == "__main__":
    main()

実行結果

============================================================
watsonx.ai Model Gateway チャットアプリ
============================================================

使用モデル: meta-llama/llama-3-3-70b-instruct
Rate Limit: 5リクエスト/分
✓ Rate Limit設定完了

チャット実行
アシスタント:
こんにちは！私はAIのアシスタントです。質問に答えたり、
情報を提供したりすることができます。

✓ チャット成功

実装例2: Rate Limitテストスクリプト

Rate Limitが正しく動作するかを検証するテストスクリプトです。
こちらもIBM Bobを使って実装しました。

"""
watsonx.ai Model Gateway Rate Limitテストスクリプト
Rate Limit: 2リクエスト/分の設定で、5リクエストを連続送信してテスト
"""

import os
import time
from dotenv import load_dotenv

from ibm_watsonx_ai import Credentials, APIClient
from ibm_watsonx_ai.gateway import Gateway
from ibm_watsonx_ai.gateway.rate_limits import RateLimits, RateLimitSettings

load_dotenv()

# 設定
MODEL_NAME = "meta-llama/llama-3-3-70b-instruct"
REQUESTS_PER_MINUTE = 2
NUM_TEST_REQUESTS = 5


def main():
    print("\n" + "="*60)
    print("watsonx.ai Model Gateway Rate Limitテスト")
    print("="*60)
    
    # 初期化
    credentials = Credentials(
        api_key=os.getenv("WATSONX_API_KEY"),
        url=os.getenv("WATSONX_URL")
    )
    client = APIClient(credentials=credentials)
    client.set.default_project(os.getenv("WATSONX_PROJECT_ID"))
    gateway = Gateway(api_client=client)
    
    # モデルIDを取得
    models_df = gateway.models.list()
    model_row = models_df[models_df['MODEL'] == MODEL_NAME]
    if model_row.empty:
        print(f"エラー: モデル '{MODEL_NAME}' が見つかりません")
        return
    
    gateway_model_id = model_row.iloc[0]['ID']
    print(f"\n使用モデル: {MODEL_NAME}")
    print(f"Rate Limit: {REQUESTS_PER_MINUTE}リクエスト/分")
    
    # 既存のRate Limitを削除
    rate_limits = RateLimits(api_client=client)
    existing = rate_limits.list()
    if not existing.empty:
        for _, row in existing.iterrows():
            limit_id = row.get('ID') or row.get('id') or row.get('uuid')
            if limit_id:
                rate_limits.delete(limit_id)
    
    # Rate Limitを設定
    result = rate_limits.create_for_model(
        model_id=gateway_model_id,
        request=RateLimitSettings(
            amount=REQUESTS_PER_MINUTE,
            capacity=REQUESTS_PER_MINUTE,
            duration="1m"
        ),
        token=RateLimitSettings(
            amount=50000,
            capacity=50000,
            duration="1m"
        )
    )
    
    rate_limit_id = result.get('uuid') or result.get('id')
    print(f"✓ Rate Limit設定完了 (ID: {rate_limit_id})")
    
    # Rate Limitが反映されるまで待機
    print("\n待機中...")
    time.sleep(5)
    
    # テスト実行
    print(f"\n{NUM_TEST_REQUESTS}リクエストを送信:")
    print("-" * 60)
    
    success = 0
    rate_limited = 0
    
    for i in range(NUM_TEST_REQUESTS):
        try:
            response = gateway.chat.completions.create(
                model=MODEL_NAME,
                messages=[{"role": "user", "content": "1+1は?"}],
                max_tokens=50
            )
            answer = response["choices"][0]["message"]["content"]
            print(f"{i+1}. ✓ 成功 - {answer[:30]}...")
            success += 1
        except Exception as e:
            if "429" in str(e):
                print(f"{i+1}. ⚠ Rate Limit到達")
                rate_limited += 1
            else:
                print(f"{i+1}. ✗ エラー: {str(e)[:50]}")
        
        time.sleep(0.1)
    
    # 結果
    print("-" * 60)
    print(f"\n結果: 成功 {success}/{NUM_TEST_REQUESTS}, Rate Limit {rate_limited}/{NUM_TEST_REQUESTS}")
    
    if rate_limited > 0:
        print("✓ Rate Limitが正常に動作しています")
    else:
        print("⚠ Rate Limitが検出されませんでした")
    
    # クリーンアップ
    cleanup = input("\nRate Limit設定を削除しますか? (y/N): ")
    if cleanup.lower() == 'y':
        rate_limits.delete(rate_limit_id)
        print("✓ 削除完了")
    
    print("="*60 + "\n")


if __name__ == "__main__":
    main()

テスト結果

============================================================
watsonx.ai Model Gateway Rate Limitテスト
============================================================

使用モデル: meta-llama/llama-3-3-70b-instruct
Rate Limit: 2リクエスト/分

5リクエストを送信:
------------------------------------------------------------
1. ✓ 成功 - 答えは2です。
2. ✓ 成功 - 答えは2です。
3. ⚠ Rate Limit到達
4. ⚠ Rate Limit到達
5. ⚠ Rate Limit到達
------------------------------------------------------------

結果: 成功 2/5, Rate Limit 3/5
✓ Rate Limitが正常に動作しています

Rate Limitの動作解析

トークン補充のタイミング

テストを長時間実行すると、以下のような結果になることがあります：

1. ✓ 成功
2. ✓ 成功
3. ⚠ Rate Limit到達
...（約30秒後）
11. ✓ 成功  ← トークンが1つ補充された
12. ⚠ Rate Limit到達

これは、duration="1m"の設定により、1分間でamount=2個のトークンが補充されるためです。

ベストプラクティス

1. 適切なRate Limit値の設定

# 開発環境: 緩めの設定
REQUESTS_PER_MINUTE = 10

# 本番環境: 厳しめの設定
REQUESTS_PER_MINUTE = 5

2. エラーハンドリング

try:
    response = gateway.chat.completions.create(...)
except Exception as e:
    if "429" in str(e):
        print("Rate Limit到達。しばらく待ってから再試行してください。")
        time.sleep(60)  # 1分待機
    else:
        print(f"エラー: {e}")

3. Rate Limitの監視

# 既存のRate Limitを確認
rate_limits = RateLimits(api_client=client)
existing = rate_limits.list()
print(existing)

4. クリーンアップ

テスト後は必ずRate Limit設定を削除しましょう：

rate_limits.delete(rate_limit_id)

まとめ

本記事では、watsonx.ai Model GatewayでRate Limitを実装する方法を解説しました。

重要なポイント

  1. トークンバケットアルゴリズムを理解する
  2. capacityとamountを適切に設定する
  3. テストで動作を確認する

参考リンク

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?