portkey-python-sdk について
portkey-python-sdkは、Portkey AI Gateway経由で200以上のLLMにリクエストを送るPython SDKです。OpenAI SDK互換のインターフェースを基盤に、フォールバック、再試行、負荷分散、キャッシュ、可観測性を備えます。
主な機能
- 01200以上のLLM向け統一インターフェース
- 02プロバイダーの自動フォールバックと再試行
- 03トラフィックの負荷分散
- 04セマンティック・完全一致キャッシュ
- 05トレースIDとリクエストメタデータ
活用シーン
- 01複数のLLMプロバイダー間でのリクエストルーティング
- 02プロバイダー障害時の自動切り替え
- 03設定済みプロバイダーへのトラフィック分散
- 04本番リクエストの追跡と確認
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/ / github.com/ portkey-ai/ skills --skill portkey-python-sdkインストール後、Claude Code で/portkey-python-sdkを入力して読み込まれたか確認します。
同梱ファイル
SKILL.md と一緒にインストールされる参考資料やスクリプト。
assets/.gitkeepreferences/.gitkeepADVANCED_FEATURES.mdAPI_REFERENCE.mdINTEGRATIONS.mdPROVIDERS.mdscripts/.gitkeep詳細
Portkey Python SDK
Portkey Python SDK は、Portkey AI Gateway を通じて 200 以上の LLM に統一されたインターフェースを提供します。OpenAI SDK をベースに構築されているためシームレスに互換性があり、自動フォールバック、再試行、負荷分散、セマンティックキャッシュ、ガードレール、包括的な可観測性など、本番環境向けの機能を追加します。
参考資料:
- API リファレンス - レスポンス構造、エラー処理
- 高度な機能 - ツール呼び出し、エンベディング、音声、画像
- フレームワーク連携 - LangChain、LlamaIndex、Strands、Google ADK
- プロバイダー設定 - Azure、AWS Bedrock、Vertex AI のセットアップ
インストール
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"}
)
ベストプラクティス
- 環境変数を使用する - API キーをコードに直接記述しないでください
- フォールバックを実装する - 本番環境では必ずバックアッププロバイダーを用意してください
- ストリーミングを使用する - 長いレスポンスでもユーザー体験が向上します
- トレースを追加する - トレース ID とメタデータで可観測性を有効にしてください
- キャッシュを有効にする - セマンティックキャッシュでコストを削減できます
- エラーを処理する - 指数バックオフによる再試行ロジックを実装してください
リソース
- ダッシュボード: app.portkey.ai
- ドキュメント: docs.portkey.ai
- GitHub: github.com/portkey-ai/portkey-python-sdk
- Discord: portkey.ai/discord
コメント
Markdown 対応