Amazon SageMaker AI エンドポイントが OpenAI 互換 API サポートを発表

本日、Amazon SageMaker AI は、リアルタイム推論エンドポイントに対する OpenAI 互換 API サポートを導入しました。OpenAI SDK、LangChain、または Strands Agents を使用している場合、エンドポイント URL の変更のみで SageMaker AI 上のモデルを呼び出すことができます。カスタムクライアントや SigV4 ラッパー、コードの書き換えは不要です。 ## オーバービュー 今回の発表により、SageMaker AI エンドポイントは Chat Completions リクエストを受け付け、コンテナからのレスポンス（ストリーミングを含む）をそのまま返す /openai/v1 パスを公開します。標準的な SageMaker AI API および SDK を使用するすべてのエンドポイントおよび推論コンポーネントで OpenAI エンドポイントが有効化されます。 SageMaker AI は URL 内のエンドポイント名に基づいてルーティングを行うため、OpenAI 互換のクライアントはそのまま動作します。現在、エンドポイントに対して期限付きのベアータークンを作成し、OpenAI クライアントで使用することが可能です。 デプロイと呼び出しを含む動作例については、以下の [GitHub のノートブック](https://github.com/aws-samples/sagemaker-genai-hosting-examples/blob/main/03-features/openai/sagemaker-inference-openai-api.ipynb) を参照してください。 ** 「私たちは、LLM ゲートウェイ（Bifrost）を介して複数の LLM プロバイダーを利用する AI コーディングエージェントを実行しています。このゲートウェイは OpenAI チャット完了プロトコルに対応しており、ベアラートークン機能により、SageMaker をカスタム SigV4 署名なしで OpenAI 互換の推論エンドポイントとして即座に追加できます。これにより、当社のゲートウェイ、Vercel AI SDK、および標準的な OpenAI クライアントとネイティブに連携します」と、Giorgio Piatti（AI/ML エンジニア – [Caffeine.AI](https://caffeine.ai/)）は述べています。 ## ユースケース ## 所有インフラ上でのエージェントワークフロー Strands Agents や LangChain などのフレームワークを使用して多段階の AI エージェントを構築している場合、これらのワークフローを SageMaker AI エンドポイント上で完全に実行できるようになりました。エージェントは、構築時に使用した同じ OpenAI 互換インターフェースを通じてモデルを呼び出しますが、推論処理はご自身のアカウント内の専用 GPU インスタンス上で行われます。 ## 単一インターフェースによる複数モデルのホスティング 一般タスク用として Llama、ドメイン固有作業用に微調整した Mistral、分類用として小型モデルなど、複数のモデルを実行している場合、推論コンポーネントを使用してこれらすべてを単一の SageMaker AI エンドポイント上でホストできます。各モデルには独自のリソース割り当てが与えられ、すべてのモデルを同じ OpenAI SDK から呼び出すことが可能です。アプリケーションコード内で別々の API クライアントやルーティングロジックを用意する必要はありません。 ## コード変更なしでファインチューニング済みモデルを提供する 特定のユースケース向けにオープンソースモデルをファインチューニングした場合、SageMaker AI 上でデプロイし、アプリケーションがすでに使用している同じ OpenAI 互換インターフェースを通じて呼び出すことができます。唯一の変更はエンドポイント URL です。残りのアプリケーション部分、つまり SDK の呼び出し、ストリーミングロジック、プロンプトのフォーマットはそのまま維持されます。 ## ソリューション概要 本記事では、以下の手順を解説します： - Bearer トークン認証が SageMaker AI エンドポイントでどのように機能するか。 - シングルモデルエンドポイントのデプロイと呼び出し。 - マルチモデルデプロイにおける推論コンポーネントのデプロイと呼び出し。 - Strands Agents フレームワークとの統合。 ## 前提条件 本ウォークスルーを完了するには、以下の環境が必要です： - SageMaker AI エンドポイントを作成する権限を持つ AWS アカウント。 - SageMaker Python SDK (pip install sagemaker)。 - OpenAI Python SDK (pip install openai)。 - Amazon Simple Storage Service (Amazon S3) に保存されたモデル。例として、Hugging Face からダウンロードした Qwen3-4B など。 - エンドポイント作成に使用する AWS Identity and Access Management (IAM) 実行ロールで、AmazonSageMakerFullAccess ポリシーが付与されていること。 - エンドポイントの呼び出しに使用する IAM 実行ロールで、sagemaker:CallWithBearerToken および sagemaker:InvokeEndpoint の権限が付与されていること。 ## Bearer トークンによる認証 SageMaker AI OpenAI 互換エンドポイントは、Bearer トークン認証を使用します。SageMaker Python SDK には、既存の AWS 資格情報から最大 12 時間有効な期限付きトークンを生成するトークンジェネレーターが含まれています。追加の秘密鍵や API キーは不要です。 このトークンには、ロールまたはユーザーの資格情報が含まれており、sagemaker:CallWithBearerToken および sagemaker:InvokeEndpoint というアクションの権限が必要です。 ## トークンの生成 以下の Python スクリプトを使用してトークンを生成してください。 ```python from sagemaker.core.token_generator import generate_token from datetime import timedelta token = generate_token(region="us-west-2", expiry=timedelta(minutes=5)) ``` このトークンジェネレーターは、環境内で利用可能な AWS 資格情報（IAM ユーザーの資格情報、Amazon Elastic Compute Cloud (Amazon EC2) のインスタンスプロファイル、または AWS IAM Identity Center (SSO) セッション）を自動的に使用します。 generate_token 関数は、SageMaker API との認証のために期限付きのベアラー トークンを生成します。デフォルトではトークンは 12 時間有効ですが、expiry パラメータに 1 秒から 12 時間の間の timedelta 値を指定することでこれを上書きできます。この関数はリージョン、オプションの aws_credentials_provider、および有効期限期間を受け取ります。AWS リージョンが指定されていない場合、AWS_REGION 環境変数がフォールバックとして使用されます。認証プロバイダが供給されない場合は、デフォルトの AWS 認証チェーンを使用して認証情報を解決します。このチェーンは、環境変数、~/.aws/credentials、~/.aws/config、コンテナ認証情報、およびインスタンスプロファイルなど複数のソースを検索します。完全な解決順序については、[Boto3 認証情報のドキュメント](https://docs.aws.amazon.com/boto3/latest/guide/configuration.html) を参照してください。 ## 長時間実行されるアプリケーション用のトークン自動リフレッシュ 継続的に実行されるアプリケーションの場合、httpx を使用して自動リフレッシュパターンを実装し、各リクエストごとに新しいトークンを生成することができます: ```python import httpx from sagemaker.core.token_generator import generate_token class SageMakerAuth(httpx.Auth): def __init__(self, region: str): self.region = region def auth_flow(self, request): request.headers["Authorization"] = f"Bearer {generate_token(region=self.region)}" yield request http_client = httpx.Client(auth=SageMakerAuth(region="us-west-2")) ``` ## IAM パーミッション エンドポイントを呼び出す IAM ロールまたはユーザーには、以下のパーミッションが必要です: { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "sagemaker:InvokeEndpoint", "Resource": "arn:aws:sagemaker:::endpoint/" }, { "Effect": "Allow", "Action": "sagemaker:CallWithBearerToken", "Resource": "*" } ] } ベストプラクティスとして、`InvokeEndpoint` のリソースにはワイルドカードを使用するのではなく、特定のエンドポイント ARN に制限を設けるようにしてください。このロールから生成されるベアートークンも同レベルのアクセス権限を持ちます。そのため、ポリシーのスコープを狭くすることで、トークンが誤って公開された場合の影響範囲（ブラスト・レイディアス）を限定できます。なお、`CallWithBearerToken` にはリソースフィールドにワイルドカード（"*"）が必要であり、リソースレベルでの制限はサポートされていません。 ## トークンの仕組み ベアータークンは、base64 でエンコードされた SigV4 事前署名 URL です。generate_token を呼び出すと、SageMaker AI SDK は SageMaker AI サービスに対して CallWithBearerToken アクションのリクエストを構築し、AWS 認証情報を使用してローカルで署名し、結果として得られた署名付き URL をポータブルなトークン文字列としてエンコードします。トークンの生成中にネットワーク呼び出しは行われません。署名処理はすべてクライアント側で行われます。このトークンを SageMaker AI エンドポイントに提示すると、サービス側でこれをデコードし、SigV4 署名を検証し、トークンの有効期限が切れていないかを確認し、元の IAM アイデンティティに必要な権限を持っているかを検証します。トークンの実効寿命は、有効期限値と、署名に使用された AWS 認証情報の残存有効期間のいずれか短い方になります。 セキュリティのベストプラクティス：**ベアータークンは、生成に使用された基盤となる AWS 認証情報と同じ権限を保持します。トークンには認証情報と同様の注意を払ってください。トークン生成に使用する IAM ロールは、必要な最小限の権限にスコープを制限してください。具体的には、呼び出し元がアクセスする必要があるエンドポイント ARN のみに対して sagemaker:InvokeEndpoint および sagemaker:CallWithBearerToken 権限のみ付与します。AdministratorAccess や SageMakerFullAccess マネージドポリシーによって付与されるような広範な権限を持つロールからトークンを生成してはいけません。 ディスク上、環境変数、設定ファイル、データベース、または分散キャッシュにトークンを保存しないでください。また、トークンをログ出力せず、HTTPS などの暗号化通信プロトコルを介してのみ送信してください。トークン生成はネットワークオーバーヘッドのないローカル操作であるため、推奨されるプラクティスは、使用箇所で新しいトークンを生成するか、前述の例で示された自動更新機能を持つ httpx.Auth パターンを使用することです。これにより、トークンの漏洩リスクを回避し、残存有効期間が最大となるトークンを利用できます。ベストプラクティスとして、ワークロードが必要とする最短の期間にトークンの有効期限を設定してください。 ## シングルモデルエンドポイントのデプロイ シングルモデルエンドポイントは 1 つのモデルをホストし、リクエストを直接処理します。以下の例では、ml.g6.2xlarge インスタンス上で SageMaker AI vLLM Deep Learning Container を使用して Qwen3-4B をデプロイする方法を示しています。 注釈：SageMaker AI エンドポイントは、トラフィックの有無に関わらず、稼働中に課金が発生します。詳細については、[Amazon SageMaker AI の価格ページ](https://aws.amazon.com/sagemaker-ai/pricing/)をご覧ください。 ```python import boto3 import sagemaker import time from sagemaker.core.helper.session_helper import Session from sagemaker.core.helper.session_helper import get_execution_role # AWS 設定 REGION = "us-west-2" ``` # アカウント ID とデフォルトの SageMaker 実行ロールを自動的に解決 session = Session(boto_session=boto3.Session(region_name=REGION)) ACCOUNT_ID = boto3.client("sts", region_name=REGION).get_caller_identity()["Account"] EXECUTION_ROLE = get_execution_role(sagemaker_session=session) # Hugging Face モデル ID MODEL_HF_ID = "Qwen/Qwen3-4B" # SageMaker vLLM ディープラーニングコンテナ VLLM_IMAGE = f"763104351884.dkr.ecr.{REGION}.amazonaws.com/vllm:0.20.2-gpu-py312-cu130-ubuntu22.04-sagemaker" # インスタンスタイプ (NVIDIA L4 GPU 1 基) INSTANCE_TYPE = "ml.g6.2xlarge" sagemaker_client = boto3.client("sagemaker", region_name=REGION) print(f"Region: {REGION}") print(f"Account ID: {ACCOUNT_ID}") print(f"Execution role: {EXECUTION_ROLE}") print(f"Model HF ID: {MODEL_HF_ID}") import time TIMESTAMP = str(int(time.time())) SME_MODEL_NAME = f"openai-compat-sme-model-{TIMESTAMP}" SME_ENDPOINT_CONFIG_NAME = f"openai-compat-sme-epc-{TIMESTAMP}" SME_ENDPOINT_NAME = f"openai-compat-sme-ep-{TIMESTAMP}" print(f"Timestamp suffix: {TIMESTAMP}") print(f"Model: {SME_MODEL_NAME}") print(f"Endpoint config: {SME_ENDPOINT_CONFIG_NAME}") print(f"Endpoint: {SME_ENDPOINT_NAME}") sagemaker_client.create_model( ModelName=SME_MODEL_NAME, ExecutionRoleArn=EXECUTION_ROLE, PrimaryContainer={ "Image": VLLM_IMAGE, "Environment": { "HF_MODEL_ID": MODEL_HF_ID, "SM_VLLM_TENSOR_PARALLEL_SIZE": "1", "SM_VLLM_MAX_NUM_SEQS": "4", "SM_VLLM_ENABLE_AUTO_TOOL_CHOICE": "true", "SM_VLLM_TOOL_CALL_PARSER": "hermes", "SAGEMAKER_ENABLE_LOAD_AWARE": "1", }, }, ) print(f"Model created: {SME_MODEL_NAME}") sagemaker_client.create_endpoint_config( EndpointConfigName=SME_ENDPOINT_CONFIG_NAME, ProductionVariants=[ { "VariantName": "variant1", "ModelName": SME_MODEL_NAME, "InstanceType": INSTANCE_TYPE, "InitialInstanceCount": 1, } ], ) print(f"Endpoint configuration created: {SME_ENDPOINT_CONFIG_NAME}") sagemaker_client.create_endpoint( EndpointName=SME_ENDPOINT_NAME, EndpointConfigName=SME_ENDPOINT_CONFIG_NAME, ) print(f"Endpoint creation initiated: {SME_ENDPOINT_NAME}") print("Waiting for endpoint to reach InService status (this takes 5-10 minutes)...") waiter = sagemaker_client.get_waiter("endpoint_in_service") waiter.wait( EndpointName=SME_ENDPOINT_NAME, WaiterConfig={"Delay": 30, "MaxAttempts": 40}, ) print(f"Endpoint is InService: {SME_ENDPOINT_NAME}") エンドポイントは数分以内に InService ステータスに遷移します。準備ができると、標準的な SageMaker AI の /invocations パスと、OpenAI 互換の /openai/v1/chat/completions パスの両方でリクエストを受け付けます。 ## シングルモデルエンドポイントの呼び出し エンドポイントが稼働状態になったら、OpenAI Python SDK を使用して呼び出します。ベース URL は以下の形式に従います： https://runtime.sagemaker..amazonaws.com/endpoints//openai/v1 from openai import OpenAI from sagemaker.core.token_generator import generate_token REGION = "us-west-2" sme_base_url = f"https://runtime.sagemaker.{REGION}.amazonaws.com/endpoints/{SME_ENDPOINT_NAME}/openai/v1" client = OpenAI( base_url=sme_base_url, api_key=generate_token(region=REGION) ) print(f"Base URL: {sme_base_url}") stream = client.chat.completions.create( model="", messages=[ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Explain how transformers work in machine learning, in three sentences."}, ], stream=True, ) for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="") print() model フィールドはコンテナに渡されます。SageMaker AI は URL 内のエンドポイント名に基づいてリクエストをルーティングするため、このフィールドは空のままにするか、コンテナが期待するモデル名と一致するように設定できます。 推論コンポーネントエンドポイントのデプロイ 推論コンポーネントを使用すると、単一のエンドポイント上で複数のモデルをホストでき、それぞれに専用の計算リソース割り当てを行うことができます。推論コンポーネントでは、モデルはエンドポイント設定ではなくコンポーネントに関連付けられます: IC_MODEL_NAME = f"openai-compat-ic-model-{TIMESTAMP}" IC_ENDPOINT_CONFIG_NAME = f"openai-compat-ic-epc-{TIMESTAMP}" IC_ENDPOINT_NAME = f"openai-compat-ic-ep-{TIMESTAMP}" IC_NAME = f"openai-compat-ic-qwen3-4b-{TIMESTAMP}" print(f"Model: {IC_MODEL_NAME}") print(f"Endpoint config: {IC_ENDPOINT_CONFIG_NAME}") print(f"Endpoint: {IC_ENDPOINT_NAME}") print(f"Inference comp: {IC_NAME}") sagemaker_client.create_model( ModelName=IC_MODEL_NAME, ExecutionRoleArn=EXECUTION_ROLE, PrimaryContainer={ "Image": VLLM_IMAGE, "Environment": { "HF_MODEL_ID": MODEL_HF_ID, "SM_VLLM_TENSOR_PARALLEL_SIZE": "1", "SM_VLLM_MAX_NUM_SEQS": "4", "SM_VLLM_ENABLE_AUTO_TOOL_CHOICE": "true", "SM_VLLM_TOOL_CALL_PARSER": "hermes", "SAGEMAKER_ENABLE_LOAD_AWARE": "1", }, }, ) print(f"Model created: {IC_MODEL_NAME}") sagemaker_client.create_endpoint_config( EndpointConfigName=IC_ENDPOINT_CONFIG_NAME, ExecutionRoleArn=EXECUTION_ROLE, ProductionVariants=[ { "VariantName": "variant1", "InstanceType": INSTANCE_TYPE, "InitialInstanceCount": 1, } ], ) print(f"Endpoint configuration created: {IC_ENDPOINT_CONFIG_NAME}") sagemaker_client.create_endpoint( EndpointName=IC_ENDPOINT_NAME, EndpointConfigName=IC_ENDPOINT_CONFIG_NAME, ) print(f"Endpoint creation initiated: {IC_ENDPOINT_NAME}") print("Waiting for endpoint to reach InService status (this takes 5-10 minutes)...") waiter = sagemaker_client.get_waiter("endpoint_in_service") waiter.wait( EndpointName=IC_ENDPOINT_NAME, WaiterConfig={"Delay": 30, "MaxAttempts": 40}, ) print(f"Endpoint is InService: {IC_ENDPOINT_NAME}") sagemaker_client.create_inference_component( InferenceComponentName=IC_NAME, EndpointName=IC_ENDPOINT_NAME, VariantName="variant1", Specification={ "ModelName": IC_MODEL_NAME, "ComputeResourceRequirements": { "MinMemoryRequiredInMb": 1024, "NumberOfCpuCoresRequired": 2, "NumberOfAcceleratorDevicesRequired": 1, }, }, RuntimeConfig={"CopyCount": 1}, ) print(f"Inference component creation initiated: {IC_NAME}") print("Waiting for inference component to reach InService status...") while True: desc = sagemaker_client.describe_inference_component(InferenceComponentName=IC_NAME) status = desc["InferenceComponentStatus"] if status == "InService": print(f"Inference component is InService: {IC_NAME}") break elif status == "Failed": raise RuntimeError(f"Inference component failed: {desc.get('FailureReason', 'unknown')}") time.sleep(30) 同じエンドポイントに追加の推論コンポーネントを作成して、独立したスケーリングとリソース割り当てで複数のモデルをホストすることができます。 ## 推論コンポーネントの呼び出し 特定の推論コンポーネントを呼び出すには、URL パスにその名前を含めます: https://runtime.sagemaker..amazonaws.com/endpoints//inference-components//openai/v1 以下の例は、共有されたエンドポイント上の 2 つの推論コンポーネントを示しており、それぞれが接続プールを共有する個別の OpenAI クライアントによってターゲットされています: import httpx from openai import OpenAI from sagemaker.core.token_generator import generate_token shared_http = httpx.Client() client_a = OpenAI( base_url=( f"https://runtime.sagemaker.{REGION}.amazonaws.com" f"/endpoints/{IC_ENDPOINT_NAME}/inference-components/{IC_NAME}/openai/v1" ), api_key=generate_token(region=REGION), http_client=shared_http, ) response = client_a.chat.completions.create( model="", messages=[{"role": "user", "content": "What is 42 * 3? Reply with the number."}], ) print(f"Response: {response.choices[0].message.content}") print(f"Connection pool active: shared_http is reusable across multiple IC clients") The shared httpx.Client allows both OpenAI client instances to reuse the same TLS sessions and connection pool. ## Integrate with Strands Agents Strands Agents は、AI エージェントを構築するためのオープンソース SDK です。Strands Agents は OpenAI 互換のモデルプロバイダーをサポートしているため、現在ではマルチエージェントワークフローを SageMaker AI インフラストラクチャ上で完全に実行できるようになりました。これにより、専用エンドポイントによる制御性を保ちながら、エージェントアプリケーションの柔軟性を得ることができます。あなたのデータはあなたのアカウントから決して離れません

Amazon SageMaker AI エンドポイントが OpenAI 互換 API サポートを発表

背景や根拠まで確認しますか？

関連記事

調べる

選ぶ

サイト