mcpField
P
portkey-python-sdk

portkey-python-sdk

github.com/portkey-ai開発272 回インストール32 週間前に更新
リポジトリ

portkey-python-sdk について

portkey-python-sdkは、Portkey AI Gateway経由で200以上のLLMにリクエストを送るPython SDKです。OpenAI SDK互換のインターフェースを基盤に、フォールバック、再試行、負荷分散、キャッシュ、可観測性を備えます。

主な機能

  1. 01200以上のLLM向け統一インターフェース
  2. 02プロバイダーの自動フォールバックと再試行
  3. 03トラフィックの負荷分散
  4. 04セマンティック・完全一致キャッシュ
  5. 05トレースIDとリクエストメタデータ

活用シーン

  1. 01複数のLLMプロバイダー間でのリクエストルーティング
  2. 02プロバイダー障害時の自動切り替え
  3. 03設定済みプロバイダーへのトラフィック分散
  4. 04本番リクエストの追跡と確認

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/portkey-ai/skills --skill portkey-python-sdk

インストール後、Claude Code で/portkey-python-sdkを入力して読み込まれたか確認します。

同梱ファイル

SKILL.md と一緒にインストールされる参考資料やスクリプト。

assets/
.gitkeep
references/
.gitkeep
ADVANCED_FEATURES.md
API_REFERENCE.md
INTEGRATIONS.md
PROVIDERS.md
scripts/
.gitkeep

詳細

Portkey Python SDK

Portkey Python SDK は、Portkey AI Gateway を通じて 200 以上の LLM に統一されたインターフェースを提供します。OpenAI SDK をベースに構築されているためシームレスに互換性があり、自動フォールバック、再試行、負荷分散、セマンティックキャッシュ、ガードレール、包括的な可観測性など、本番環境向けの機能を追加します。

参考資料:


インストール

pip install portkey-ai

# Or with poetry/uv
poetry add portkey-ai
uv add portkey-ai

クイックスタート

import os
from portkey_ai import Portkey

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    virtual_key="your-openai-virtual-key"
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

print(response.choices[0].message.content)

認証

API キー + 仮想キー(推奨)

仮想キーを使うと、プロバイダーの API キーを Portkey のボルトに安全に保存できます。

import os
from portkey_ai import Portkey

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],  # From app.portkey.ai
    virtual_key="openai-virtual-key-xxx"     # From app.portkey.ai/virtual-keys
)
Config ID の使用

ダッシュボードでルーティング、フォールバック、キャッシュを事前に設定できます。

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    config="pc-config-xxx"  # Config ID from dashboard
)

チャット補完

基本的なリクエスト
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain quantum computing briefly."}
    ]
)

print(response.choices[0].message.content)
print(f"Tokens used: {response.usage.total_tokens}")
ストリーミング
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a short story"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
非同期処理のサポート
import asyncio
from portkey_ai import AsyncPortkey

async def main():
    client = AsyncPortkey(
        api_key=os.environ["PORTKEY_API_KEY"],
        virtual_key="openai-key"
    )
    
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}]
    )
    print(response.choices[0].message.content)

asyncio.run(main())
非同期ストリーミング
async def stream_response():
    client = AsyncPortkey(
        api_key=os.environ["PORTKEY_API_KEY"],
        virtual_key="openai-key"
    )
    
    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Write a poem"}],
        stream=True
    )
    
    async for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

ゲートウェイ機能

フォールバック

プロバイダーで障害が発生した際に、自動的に切り替えます。

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    config={
        "strategy": {"mode": "fallback"},
        "targets": [
            {
                "virtual_key": "openai-key",
                "override_params": {"model": "gpt-4o"}
            },
            {
                "virtual_key": "anthropic-key",
                "override_params": {"model": "claude-3-5-sonnet-20241022"}
            }
        ]
    }
)

# If OpenAI fails, automatically tries Anthropic
response = client.chat.completions.create(
    messages=[{"role": "user", "content": "Hello!"}]
)
負荷分散

プロバイダー間でトラフィックを分散します。

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    config={
        "strategy": {"mode": "loadbalance"},
        "targets": [
            {"virtual_key": "openai-key-1", "weight": 0.7},
            {"virtual_key": "openai-key-2", "weight": 0.3}
        ]
    }
)
自動再試行
client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    config={
        "retry": {
            "attempts": 3,
            "on_status_codes": [429, 500, 502, 503, 504]
        },
        "virtual_key": "openai-key"
    }
)
セマンティックキャッシュ

インテリジェントなキャッシュにより、コストとレイテンシーを削減します。

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    config={
        "cache": {
            "mode": "semantic",  # or "simple" for exact match
            "max_age": 3600      # TTL in seconds
        },
        "virtual_key": "openai-key"
    }
)

# Similar queries return cached responses
response1 = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What is the capital of France?"}]
)

response2 = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me France's capital"}]
)  # Returns cached response
リクエストのタイムアウト
client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    virtual_key="openai-key",
    request_timeout=30  # 30 seconds
)

可観測性

トレース ID

関連するリクエストを紐付けてデバッグに役立てます。

import uuid

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    virtual_key="openai-key",
    trace_id=str(uuid.uuid4())
)
カスタムメタデータ

検索可能なメタデータをリクエストに追加します。

client = Portkey(
    api_key=os.environ["PORTKEY_API_KEY"],
    virtual_key="openai-key",
    metadata={
        "user_id": "user-123",
        "session_id": "session-456",
        "environment": "production"
    }
)
リクエストごとのオプション
response = client.with_options(
    trace_id="unique-trace-id",
    metadata={"request_type": "summarization"}
).chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarize this..."}]
)

よく使われるパターン

マルチターンの会話
messages = [
    {"role": "system", "content": "You are a helpful coding assistant."},
    {"role": "user", "content": "What is Python?"},
    {"role": "assistant", "content": "Python is a high-level programming language..."},
    {"role": "user", "content": "Show me a hello world example."}
]

response = client.chat.completions.create(model="gpt-4o", messages=messages)
JSON 出力
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Extract as JSON with name and age fields."},
        {"role": "user", "content": "John is 30 years old."}
    ],
    response_format={"type": "json_object"}
)
# Returns: {"name": "John", "age": 30}
フォールバックとキャッシュを使った本番環境のセットアップ
def create_production_client():
    return Portkey(
        api_key=os.environ["PORTKEY_API_KEY"],
        config={
            "strategy": {"mode": "fallback"},
            "targets": [
                {
                    "virtual_key": os.environ["OPENAI_VIRTUAL_KEY"],
                    "override_params": {"model": "gpt-4o"},
                    "retry": {"attempts": 2, "on_status_codes": [429, 500]}
                },
                {
                    "virtual_key": os.environ["ANTHROPIC_VIRTUAL_KEY"],
                    "override_params": {"model": "claude-3-5-sonnet-20241022"}
                }
            ],
            "cache": {"mode": "semantic", "max_age": 3600}
        },
        trace_id="production-session",
        metadata={"environment": "production"}
    )

ベストプラクティス

  1. 環境変数を使用する - API キーをコードに直接記述しないでください
  2. フォールバックを実装する - 本番環境では必ずバックアッププロバイダーを用意してください
  3. ストリーミングを使用する - 長いレスポンスでもユーザー体験が向上します
  4. トレースを追加する - トレース ID とメタデータで可観測性を有効にしてください
  5. キャッシュを有効にする - セマンティックキャッシュでコストを削減できます
  6. エラーを処理する - 指数バックオフによる再試行ロジックを実装してください

リソース

コメント