本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# 使用 OpenAI-compatible API 调用终端节点
<a name="realtime-endpoints-openai-compatible"></a>

Amazon SageMaker AI 实时推理终端节点支持 OpenAI-compatible API 路径。使用 OpenAI SDK 或 Strands Agents 的客户只需更改端点 URL 即可在 SageMaker AI 上调用模型，而无需自定义客户端、SigV4 包装器或代码重写。 LangChain

借助此功能， SageMaker AI 端点会公开一条`/openai/v1/chat/completions`路径，该路径接受聊天完成请求并直接从容器返回响应，包括流式传输。 OpenAI-compatible 使用标准 SageMaker AI API 和 SDK，端点可在所有端点和推理组件上使用。

SageMaker AI 根据网址中的终端节点名称路由请求。任何 OpenAI-compatible 客户端都无需额外配置即可运行。您可以为终端节点创建短期不记名令牌，并将其用于 OpenAI 客户端。

## 先决条件
<a name="realtime-endpoints-openai-compatible-prerequisites"></a>

在开始之前，请确保您具有以下各项：
+ 有权创建 SageMaker AI 终端节点的 AWS 账户。
+  SageMaker AI Python 软件开发工具包已安装 (`pip install sagemaker`)。
+ OpenAI Python SDK 已安装 () `pip install openai`。
+ 存储在亚马逊 S3 中的模型（例如，从 Hugging Face Qwen3-4B 下载）。
+ 一个 IAM 执行角色，其`AmazonSageMakerFullAccess`策略用于创建终端节点。
+ 具有`sagemaker:CallWithBearerToken`和调用终端节点`sagemaker:InvokeEndpoint`权限的 IAM 角色或用户。

## 使用不记名令牌进行身份验证
<a name="realtime-endpoints-openai-compatible-auth"></a>

SageMaker AI OpenAI-compatible 端点使用不记名令牌身份验证。 SageMaker AI Python SDK 包含一个令牌生成器，该生成器可根据您的现有 AWS 凭证创建短期令牌（有效期最长 12 小时）。不需要额外的密钥或 API 密钥。

该令牌包含您的角色或用户凭证，并且需要`sagemaker:CallWithBearerToken`和`sagemaker:InvokeEndpoint`操作权限。

### 生成代币
<a name="realtime-endpoints-openai-compatible-auth-generate"></a>

使用 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`函数生成一个短暂的不记名令牌，用于使用 AI API 进行 SageMaker 身份验证。默认情况下，令牌的有效期为 12 小时。您可以使用 1 秒到 12 小时之间的任意`timedelta`值使用`expiry`参数来覆盖此值。

该函数接受 a `region`、可选`aws_credentials_provider`函数和`expiry`持续时间。如果未提供区域，则会回退到`AWS_REGION`环境变量。如果未提供凭据提供程序，则它使用默认 AWS 凭证链解析证书，该证书链会搜索多个来源，包括环境变量、、`~/.aws/credentials``~/.aws/config`、容器凭证和实例配置文件。有关完整解析顺序，请参阅 [boto3 凭证文档](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html)。

### Auto-refresh 用于长时间运行的应用程序的代币
<a name="realtime-endpoints-openai-compatible-auth-refresh"></a>

对于持续运行的应用程序，您可以使用实现自动刷新模式，`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 权限
<a name="realtime-endpoints-openai-compatible-auth-iam"></a>

调用终端节点的 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": "*"
        }
    ]
}
```

**重要**  
始终将 for 限制`Resource``sagemaker:InvokeEndpoint`为特定的端点 ARN，而不是使用通配符。由此角色生成的持有者令牌具有相同的访问级别，因此，如果无意中暴露了令牌，则范围狭窄的策略会限制爆炸半径。

**注意**  
`sagemaker:CallWithBearerToken`该`Resource`字段需要通配符 (`"*"`)。它不支持资源级限制。

### 代币的工作原理
<a name="realtime-endpoints-openai-compatible-auth-how-it-works"></a>

持有者令牌是 base64 编码的 Sigv4 预签名网址。当您调用时`generate_token`， SageMaker AI SDK 会构造向 SageMaker AI 服务发送的`CallWithBearerToken`操作请求，使用您的 AWS 凭据在本地对其进行签名，并将生成的签名 URL 编码为便携式令牌字符串。代币生成期间不会进行网络调用——签名完全在客户端进行。

当您向 SageMaker AI 终端节点出示此令牌时，该服务会对其进行解码，验证 Sigv4 签名，验证令牌是否未过期，并确认原始 IAM 身份是否具有所需的权限。令牌的有效寿命是其`expiry`值和用于签名的 AWS 证书的剩余有效期中较小的一个。

### 安全最佳实践
<a name="realtime-endpoints-openai-compatible-auth-security"></a>

持有者令牌具有与生成该令牌的底层 AWS 凭证相同的授权。像对待凭证一样谨慎对待令牌。遵循如下最佳实践：
+ 将用于生成令牌的 IAM 角色限定为所需的最低权限，具体而言`sagemaker:InvokeEndpoint`，仅限`sagemaker:CallWithBearerToken`于调用者需要访问的终端节点 ARN。
+ 不要从具有广泛权限的角色（例如由`AdministratorAccess`或`AmazonSageMakerFullAccess`托管策略授予的角色）生成令牌。
+ 不要将令牌存储在磁盘、环境变量、配置文件、数据库或分布式缓存中。不要记录令牌，只能通过 HTTPS 等加密通信协议进行传输。
+ 代币生成是一项本地操作，没有网络开销。在使用时生成新的令牌或使用上面显示的自动刷新`httpx.Auth`模式。
+ 将令牌到期时间设置为工作负载所需的最短持续时间。

## 调用单模型端点
<a name="realtime-endpoints-openai-compatible-single-model"></a>

单一模型端点托管一个模型并直接为请求提供服务。以下示例 Qwen3-4B 使用 SageMaker AI vLLM 深度学习容器在实例上进行部署。`ml.g6.2xlarge`

**注意**  
SageMaker 无论流量如何，AI 端点在服务期间都会产生费用。有关详细信息，请参阅 [SageMaker AI 定价页面](https://aws.amazon.com/sagemaker/pricing/)。

### 部署终端节点
<a name="realtime-endpoints-openai-compatible-single-model-deploy"></a>

```
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-compatible 路径`/openai/v1/chat/completions`。

### 调用端点
<a name="realtime-endpoints-openai-compatible-single-model-invoke"></a>

终端节点投入使用后，使用 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 会根据网址中的终端节点名称路由请求，因此您可以将此字段留空或将其设置为与容器期望的模型名称相匹配。

## 调用推理组件
<a name="realtime-endpoints-openai-compatible-inference-components"></a>

推理组件允许您在单个端点上托管多个模型，每个模型都有专用的计算资源分配。对于推理组件，模型与组件相关联，而不是与端点配置相关联。

### 部署推理组件端点
<a name="realtime-endpoints-openai-compatible-ic-deploy"></a>

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

您可以在同一个端点上创建其他推理组件，以托管具有独立扩展和资源分配功能的多个模型。

### 调用推理组件
<a name="realtime-endpoints-openai-compatible-ic-invoke"></a>

要调用特定的推理组件，请在网址路径中包含其名称：

```
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 会话和连接池。

## 支持的容器
<a name="realtime-endpoints-openai-compatible-containers"></a>

以下容器支持 A OpenAI-compatible I 上的 AP SageMaker I。容器必须实现`/v1/chat/completions`路径并以 SSE 格式返回流式响应。


|  Container  |  Support 状态  | 
| --- | --- | 
| SageMaker AI vlLM 深度学习容器 | 支持 | 
| SageMaker AI sGlang 深度学习容器 | 支持 | 
| 实现 OpenAI API 路径的自定义容器以及 `/ping` | 支持 |