Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Invoquer des points de terminaison avec des API OpenAI-compatible

Les points de terminaison d'inférence en temps réel Amazon SageMaker AI prennent en charge un chemin d' OpenAI-compatible API. Les clients utilisant le SDK OpenAI ou les agents Strands peuvent invoquer des modèles sur l' SageMaker IA en modifiant uniquement l'URL de leur point de terminaison, sans avoir besoin de clients personnalisés, de wrappers SigV4 ou de réécritures de code. LangChain

Grâce à cette fonctionnalité, les points de terminaison SageMaker AI exposent un /openai/v1/chat/completions chemin qui accepte les demandes de fin de chat et renvoie les réponses directement depuis le conteneur, y compris le streaming. OpenAI-compatible les points de terminaison sont disponibles sur tous les points de terminaison et composants d'inférence à l'aide d'API et de SDK d' SageMaker IA standard.

SageMaker L'IA achemine les demandes en fonction du nom du point de terminaison indiqué dans l'URL. Tout OpenAI-compatible client fonctionne sans configuration supplémentaire. Vous pouvez créer des jetons porteurs de courte durée pour vos terminaux et les utiliser avec vos clients OpenAI.

Conditions préalables

Avant de commencer, assurez-vous de disposer des éléments suivants :

Authentification avec des jetons au porteur

SageMaker Les OpenAI-compatible points de terminaison AI utilisent l'authentification par jeton porteur. Le SDK SageMaker AI Python inclut un générateur de jetons qui crée des jetons de courte durée (valables jusqu'à 12 heures) à partir de vos informations d'identification existantes AWS . Aucun secret ou clé d'API supplémentaire n'est requis.

Le jeton contient vos informations d'identification de rôle ou d'utilisateur et nécessite les autorisations sagemaker:InvokeEndpoint d'action sagemaker:CallWithBearerToken et.

Générer un jeton

Utilisez la generate_token fonction du SDK SageMaker AI Python pour créer un jeton porteur :

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

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

La generate_token fonction génère un jeton porteur de courte durée pour l'authentification avec SageMaker les API d'IA. Par défaut, les jetons sont valides pendant 12 heures. Vous pouvez le remplacer par le expiry paramètre en utilisant une timedelta valeur comprise entre 1 seconde et 12 heures.

La fonction accepte unregion, un optionnel aws_credentials_provider et la expiry durée. Si aucune région n'est spécifiée, elle revient à la variable d'AWS_REGIONenvironnement. Si aucun fournisseur d'informations d'identification n'est fourni, il résout les informations d'identification à l'aide de la chaîne AWS d'informations d'identification par défaut, qui recherche plusieurs sources, notamment les variables d'environnement ~/.aws/credentials~/.aws/config, les informations d'identification des conteneurs et les profils d'instance. Pour l'ordre de résolution complet, consultez la documentation des informations d'identification boto3.

Auto-refresh jetons pour les applications de longue durée

Pour les applications qui s'exécutent en continu, vous pouvez implémenter un modèle d'actualisation automatique en utilisant de httpx telle sorte qu'un nouveau jeton soit généré à chaque demande :

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

Autorisations IAM

Le rôle IAM ou l'utilisateur qui appelle le point de terminaison a besoin des autorisations suivantes :

{
    "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": "*"
        }
    ]
}

Comment fonctionne le jeton

Le jeton porteur est une URL Sigv4 pré-signée codée en base64. Lorsque vous appelezgenerate_token, le SDK SageMaker AI crée une demande d'CallWithBearerTokenaction adressée au service SageMaker AI, la signe localement à l'aide de vos AWS informations d'identification et code l'URL signée qui en résulte sous forme de chaîne de jeton portable. Aucun appel réseau n'est effectué lors de la génération du jeton : la signature se fait entièrement du côté client.

Lorsque vous présentez ce jeton à un point de terminaison SageMaker AI, le service le décode, valide la signature SigV4, vérifie que le jeton n'a pas expiré et confirme que l'identité IAM d'origine dispose des autorisations requises. La durée de vie effective du jeton correspond au montant le moins élevé entre la expiry valeur et la durée de validité restante des AWS informations d'identification utilisées pour le signer.

Bonnes pratiques de sécurité

Le jeton porteur possède la même autorisation que les informations d' AWS identification sous-jacentes utilisées pour le générer. Traitez les jetons avec le même soin que les informations d'identification. Suivez ces bonnes pratiques :

Invoquer un point de terminaison à modèle unique

Un point de terminaison à modèle unique héberge un modèle et répond directement aux demandes. L'exemple suivant montre comment Qwen3-4B utiliser le conteneur d'apprentissage profond SageMaker AI vLLM sur une instance. ml.g6.2xlarge

Déployer le terminal

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)

Créez le modèle, la configuration du point de terminaison et le point de terminaison :

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

Le terminal passe à InService l'état en quelques minutes. Une fois prêt, il dessert à la fois le /invocations chemin standard de l' SageMaker IA et le OpenAI-compatible chemin à/openai/v1/chat/completions.

Appeler le point de terminaison

Lorsque le point de terminaison est en service, invoquez-le à l'aide du SDK OpenAI Python. L'URL de base suit le format suivant :

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()

Le model champ est transmis au conteneur. Étant donné que l' SageMaker IA achemine les demandes en fonction du nom du point de terminaison indiqué dans l'URL, vous pouvez laisser ce champ vide ou le définir pour qu'il corresponde au nom de modèle attendu par votre conteneur.

Invoquer des composants d'inférence

Les composants d'inférence vous permettent d'héberger plusieurs modèles sur un seul point de terminaison, chacun étant doté d'allocations de ressources de calcul dédiées. Dans le cas des composants d'inférence, le modèle est associé au composant plutôt qu'à la configuration du point de terminaison.

Déployer un terminal de composant d'inférence

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)

Vous pouvez créer des composants d'inférence supplémentaires sur le même point de terminaison pour héberger plusieurs modèles avec une mise à l'échelle et une allocation de ressources indépendantes.

Invoquer un composant d'inférence

Pour appeler un composant d'inférence spécifique, incluez son nom dans le chemin de l'URL :

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

L'exemple suivant montre comment invoquer un composant d'inférence à l'aide du SDK OpenAI avec un pool de connexions partagé :

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)

Le partage httpx.Client permet à plusieurs instances clientes OpenAI de réutiliser les mêmes sessions TLS et le même pool de connexions lors du ciblage de différents composants d'inférence sur le même point de terminaison.

Conteneurs compatibles

Les conteneurs suivants prennent en charge OpenAI-compatible les API sur l' SageMaker IA. Le conteneur doit implémenter le /v1/chat/completions chemin et renvoyer les réponses de streaming au format SSE.