翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

OpenAI 互換 APIs呼び出す

Amazon SageMaker AI リアルタイム推論エンドポイントは、OpenAI 互換 API パスをサポートします。OpenAI SDK、LangChain、または Strands エージェントを使用しているお客様は、カスタムクライアント、SigV4 ラッパー、またはコードの書き換えを必要とせずに、エンドポイント URL のみを変更することで、SageMaker AI でモデルを呼び出すことができます。

この機能を使用すると、SageMaker AI エンドポイントは、チャット完了リクエストを受け入れ、ストリーミングを含むコンテナから直接レスポンスを返す/openai/v1/chat/completionsパスを公開します。OpenAI 互換エンドポイントは、標準の SageMaker AI APIs と SDKs を使用して、すべてのエンドポイントと推論コンポーネントで使用できます。

SageMaker AI は、URL のエンドポイント名に基づいてリクエストをルーティングします。OpenAI 互換クライアントは、追加の設定なしで動作します。エンドポイントの存続期間の短いベアラートークンを作成し、OpenAI クライアントで使用できます。

前提条件

作業を開始する前に、次の項目があることを確認します。

ベアラートークンによる認証

SageMaker AI OpenAI 互換エンドポイントは、ベアラートークン認証を使用します。SageMaker AI Python SDK には、既存の AWS 認証情報から有効期間の短いトークン (最大 12 時間有効) を作成するトークンジェネレーターが含まれています。追加のシークレットや API キーは必要ありません。

トークンにはロールまたはユーザーの認証情報が含まれており、 sagemaker:CallWithBearerTokenおよび sagemaker:InvokeEndpointアクションのアクセス許可が必要です。

トークンを生成する

SageMaker AI Python SDK の generate_token関数を使用して、ベアラートークンを作成します。

from sagemaker.core.token_generator import generate_token
from datetime import timedelta

token = generate_token(region="us-west-2", expiry=timedelta(minutes=5))

このgenerate_token関数は、SageMaker AI APIs で認証するための存続期間の短いベアラートークンを生成します。デフォルトでは、トークンは 12 時間有効です。これは、1 秒から 12 時間の間のtimedelta値を使用して expiryパラメータで上書きできます。

関数は、region、オプションの aws_credentials_provider、およびexpiry期間を受け入れます。リージョンが指定されていない場合は、 AWS_REGION環境変数にフォールバックします。認証情報プロバイダーが指定されていない場合、デフォルトの AWS 認証情報チェーンを使用して認証情報を解決し、環境変数、、、コンテナ認証情報~/.aws/credentials~/.aws/config、インスタンスプロファイルなどの複数のソースを検索します。完全な解決順序については、boto3 認証情報のドキュメントを参照してください。

長時間実行されるアプリケーションのトークンの自動更新

継続的に実行されるアプリケーションでは、 を使用して自動更新パターンを実装httpxし、リクエストごとに新しいトークンを生成できます。

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:REGION:ACCOUNT_ID:endpoint/ENDPOINT_NAME"
        },
        {
            "Effect": "Allow",
            "Action": "sagemaker:CallWithBearerToken",
            "Resource": "*"
        }
    ]
}

トークンの仕組み

ベアラートークンは、base64 でエンコードされた SigV4 署名付き URL です。を呼び出すとgenerate_token、SageMaker AI SDK はCallWithBearerTokenアクションの SageMaker AI サービスへのリクエストを作成し、 AWS 認証情報を使用してローカルで署名し、結果の署名付き URL をポータブルトークン文字列としてエンコードします。トークンの生成中にネットワーク呼び出しは行われません。署名はクライアント側で完全に行われます。

このトークンを SageMaker AI エンドポイントに提示すると、サービスはそれをデコードし、SigV4 署名を検証し、トークンの有効期限が切れていないことを確認し、発信元の IAM ID に必要なアクセス許可があることを確認します。トークンの有効期間は、expiry値と署名に使用される AWS 認証情報の残りの有効期間のうち小さい方です。

セキュリティのベストプラクティス

ベアラートークンは、それを生成するために使用される基盤となる AWS 認証情報と同じ認可を保持します。トークンは認証情報と同じ注意を払って処理します。次のベストプラクティスに従ってください。

単一モデルエンドポイントを呼び出す

単一モデルエンドポイントは 1 つのモデルをホストし、リクエストを直接処理します。次の例では、SageMaker AI vLLM Deep Learning Container を使用して Qwen3-4B をml.g6.2xlargeインスタンスにデプロイします。

エンドポイントをデプロイする

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 configuration
REGION = "us-west-2"

# Automatically resolve account ID and default SageMaker execution role
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)

# HF Model ID
MODEL_HF_ID = "Qwen/Qwen3-4B"

# SageMaker vLLM Deep Learning Container
VLLM_IMAGE = (
    f"763104351884.dkr.ecr.{REGION}.amazonaws.com/"
    f"vllm:0.20.2-gpu-py312-cu130-ubuntu22.04-sagemaker"
)

# Instance type (1x NVIDIA L4 GPU)
INSTANCE_TYPE = "ml.g6.2xlarge"

sagemaker_client = boto3.client("sagemaker", region_name=REGION)

モデル、エンドポイント設定、エンドポイントを作成します。

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}"

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",
        },
    },
)

sagemaker_client.create_endpoint_config(
    EndpointConfigName=SME_ENDPOINT_CONFIG_NAME,
    ProductionVariants=[
        {
            "VariantName": "variant1",
            "ModelName": SME_MODEL_NAME,
            "InstanceType": INSTANCE_TYPE,
            "InitialInstanceCount": 1,
        }
    ],
)

sagemaker_client.create_endpoint(
    EndpointName=SME_ENDPOINT_NAME,
    EndpointConfigName=SME_ENDPOINT_CONFIG_NAME,
)

# Wait for endpoint to reach InService status (5-10 minutes)
waiter = sagemaker_client.get_waiter("endpoint_in_service")
waiter.wait(
    EndpointName=SME_ENDPOINT_NAME,
    WaiterConfig={"Delay": 30, "MaxAttempts": 40},
)

エンドポイントは数分以内にInServiceステータスに移行します。準備ができたら、標準の SageMaker AI /invocationsパスと OpenAI 互換パスの両方を で提供します/openai/v1/chat/completions。

エンドポイントの呼び出し

エンドポイントを稼働状態で、OpenAI Python SDK を使用してエンドポイントを呼び出します。ベース URL は次の形式に従います。

https://runtime.sagemaker.REGION.amazonaws.com/endpoints/ENDPOINT_NAME/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"
    f"/endpoints/{SME_ENDPOINT_NAME}/openai/v1"
)

client = OpenAI(
    base_url=sme_base_url,
    api_key=generate_token(region=REGION),
)

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 のエンドポイント名に基づいてリクエストをルーティングするため、このフィールドを空のままにするか、コンテナが想定するモデル名と一致するように設定できます。

推論コンポーネントを呼び出す

推論コンポーネントを使用すると、1 つのエンドポイントで複数のモデルをホストし、それぞれに専用のコンピューティングリソースを割り当てることができます。推論コンポーネントでは、モデルはエンドポイント設定ではなくコンポーネントに関連付けられます。

推論コンポーネントエンドポイントをデプロイする

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}"

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",
        },
    },
)

sagemaker_client.create_endpoint_config(
    EndpointConfigName=IC_ENDPOINT_CONFIG_NAME,
    ExecutionRoleArn=EXECUTION_ROLE,
    ProductionVariants=[
        {
            "VariantName": "variant1",
            "InstanceType": INSTANCE_TYPE,
            "InitialInstanceCount": 1,
        }
    ],
)

sagemaker_client.create_endpoint(
    EndpointName=IC_ENDPOINT_NAME,
    EndpointConfigName=IC_ENDPOINT_CONFIG_NAME,
)

# Wait for endpoint
waiter = sagemaker_client.get_waiter("endpoint_in_service")
waiter.wait(
    EndpointName=IC_ENDPOINT_NAME,
    WaiterConfig={"Delay": 30, "MaxAttempts": 40},
)

# Create the inference component
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},
)

# Wait for inference component
while True:
    desc = sagemaker_client.describe_inference_component(InferenceComponentName=IC_NAME)
    status = desc["InferenceComponentStatus"]
    if status == "InService":
        break
    elif status == "Failed":
        raise RuntimeError(f"Inference component failed: {desc.get('FailureReason', 'unknown')}")
    time.sleep(30)

同じエンドポイントに追加の推論コンポーネントを作成して、独立したスケーリングとリソース割り当てで複数のモデルをホストできます。

推論コンポーネントを呼び出す

特定の推論コンポーネントを呼び出すには、URL パスにその名前を含めます。

https://runtime.sagemaker.REGION.amazonaws.com/endpoints/ENDPOINT_NAME/inference-components/IC_NAME/openai/v1

次の例は、共有接続プールで OpenAI SDK を使用して推論コンポーネントを呼び出す方法を示しています。

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(response.choices[0].message.content)

共有 httpx.Clientでは、複数の OpenAI クライアントインスタンスが同じエンドポイントで異なる推論コンポーネントをターゲットにするときに、同じ TLS セッションと接続プールを再利用できます。

サポートされているコンテナ

次のコンテナは、SageMaker AI での OpenAI 互換 APIs をサポートしています。コンテナは/v1/chat/completionsパスを実装し、ストリーミングレスポンスを SSE 形式で返す必要があります。