本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。
使用 OpenAI 相容 APIs叫用端點
Amazon SageMaker AI 即時推論端點支援與 OpenAI 相容的 API 路徑。使用 OpenAI SDK、LangChain 或 Strands Agents 的客戶可以只變更其端點 URL 來叫用 SageMaker AI 上的模型,而不需要自訂用戶端、SigV4 包裝函式或程式碼重寫。
透過此功能,SageMaker AI 端點會公開接受聊天完成請求的/openai/v1/chat/completions路徑,並直接從容器傳回回應,包括串流。所有使用標準 SageMaker AI APIs 和 SDK 的端點和推論元件都可以使用 OpenAI 相容端點。 SDKs
SageMaker AI 會根據 URL 中的端點名稱路由請求。任何與 OpenAI 相容的用戶端都可以在沒有其他組態的情況下運作。您可以為端點建立短期承載字符,並將其與 OpenAI 用戶端搭配使用。
先決條件
開始前,請確定您具有下列項目:
-
具有建立 SageMaker AI 端點許可 AWS 的帳戶。
-
已安裝 SageMaker AI Python SDK (
pip install sagemaker)。 -
已安裝 OpenAI Python SDK (
pip install openai)。 -
存放在 Amazon S3 中的模型 (例如,從 Hugging Face 下載的 Qwen3-4B)。
-
具有建立端點之
AmazonSageMakerFullAccess政策的 IAM 執行角色。 -
具有叫用端點之
sagemaker:CallWithBearerToken和sagemaker:InvokeEndpoint許可的 IAM 角色或使用者。
使用承載字符進行身分驗證
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": "*" } ] }
重要
一律將 Resource的 限制sagemaker:InvokeEndpoint為特定端點 ARNs,而不是使用萬用字元。從此角色產生的承載符記具有相同的存取層級,因此如果符記不小心公開,縮小範圍的政策會限制爆量半徑。
注意
sagemaker:CallWithBearerToken 需要 Resource 欄位的萬用字元 ("*")。它不支援資源層級限制。
字符的運作方式
承載字符是 base64 編碼的 SigV4 預先簽章 URL。當您呼叫 時generate_token,SageMaker AI SDK 會建構對 SageMaker AI 服務的 CallWithBearerToken動作請求,使用您的 AWS 登入資料在本機簽署,並將產生的已簽署 URL 編碼為可攜式字符字串。在權杖產生期間不會進行網路呼叫 - 簽署完全發生在用戶端。
當您將此權杖呈現給 SageMaker AI 端點時,服務會將其解碼、驗證 SigV4 簽章、驗證權杖尚未過期,並確認原始 IAM 身分具有必要的許可。字符的有效存留期是expiry值的較小者,以及用於簽署它的 AWS 憑證的剩餘有效性。
安全最佳實務
承載字符具有與用來產生它的基礎 AWS 登入資料相同的授權。以與登入資料相同的小心處理字符。請遵循以下最佳實務:
-
將產生字符的 IAM 角色範圍限定為所需的最低許可,特別是在呼叫者需要存取ARNs
sagemaker:InvokeEndpointsagemaker:CallWithBearerToken上。 -
請勿從具有廣泛許可的角色產生字符,例如
AdministratorAccess或AmazonSageMakerFullAccess受管政策授予的角色。 -
請勿在磁碟、環境變數、組態檔案、資料庫或分散式快取中存放字符。請勿記錄字符,並僅透過加密通訊協定傳輸字符,例如 HTTPS。
-
權杖產生是本機操作,沒有網路額外負荷。在使用點產生新的權杖,或使用上面顯示的自動重新整理
httpx.Auth模式。 -
將權杖過期設定為工作負載所需的最短持續時間。
叫用單一模型端點
單一模型端點託管一個模型,並直接提供請求。下列範例使用ml.g6.2xlarge執行個體上的 SageMaker AI vLLM 深度學習容器部署 Qwen3-4B。
注意
SageMaker AI 端點在服務期間會產生費用,無論流量為何。如需詳細資訊,請參閱 SageMaker AI 定價頁面
部署端點
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 中的端點名稱路由請求,因此您可以將此欄位保留空白,或將其設定為符合容器預期的模型名稱。
叫用推論元件
推論元件可讓您在單一端點上託管多個模型,每個模型都有專用的運算資源配置。使用推論元件時,模型會與元件相關聯,而不是與端點組態相關聯。
部署推論元件端點
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 開發套件叫用推論元件:
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 格式傳回串流回應。
|
容器 |
支援狀態 |
|---|---|
|
SageMaker AI vLLM 深度學習容器 |
支援 |
|
SageMaker AI SGLang 深度學習容器 |
支援 |
|
實作 OpenAI API 路徑和 的自訂容器 |
支援 |