调用实时端点

使用 SageMaker 托管服务部署模型后，可以通过发送测试数据在该端点上测试模型。您可以使用 Amazon SageMaker Studio、Amazon SDK 或 Amazon CLI 测试端点。

使用 Amazon SageMaker Studio 调用端点

将模型部署到端点后（参见创建端点并部署模型），您可以使用 Amazon SageMaker Studio 检查该端点。

卡片顶部显示了发送到端点的请求类型（仅接受 JSON）。卡片中显示了以下字段：

使用 Amazon SDK for Python (Boto3) 调用端点

将模型部署到端点后（参见创建端点并部署模型），您可以使用其中一个 Amazon SDK 检查端点，包括 Amazon SDK for Python (Boto3)。要使用此 SDK 测试端点，请使用以下方法之一：

在客户端代码中使用上述方法之前，必须创建一个 SageMaker Runtime 客户端，并且必须指定端点的名称。以下示例为接下来的示例设置了客户端和端点：

import boto3

# Create a low-level client representing Amazon SageMaker Runtime
sagemaker_runtime = boto3.client(
    "sagemaker-runtime", region_name='aws_region')

# The endpoint name must be unique within 
# an AWS Region in your AWS account. 
endpoint_name='endpoint-name'

调用以获取推理响应

以下示例使用 invoke_endpoint 方法，通过 Amazon SDK for Python (Boto3) 调用端点：

# Gets inference from the model hosted at the specified endpoint:
response = sagemaker_runtime.invoke_endpoint(
    EndpointName=endpoint_name, 
    Body=bytes('{"features": ["This is great!"]}', 'utf-8')
    )

# Decodes and prints the response body:
print(response['Body'].read().decode('utf-8'))

此示例在 Body 字段中提供输入数据，以供 SageMaker 传递给模型。此数据的格式必须与用于训练的数据格式相同。该示例将响应存储在 response 变量中。

response 变量提供了对 HTTP 状态、已部署模型的名称以及其他字段的访问。以下代码段打印了 HTTPStatusCode：

print(response["HTTPStatusCode"])

调用以流式处理推理响应

如果您部署了支持推理流的模型，则可调用该模型以流的形式接收其推理负载部分。模型在生成推理响应时，会以增量方式交付这些部分。应用程序在接收推理流时，无需等待模型生成整个响应负载。而是在部分响应变得可用时，应用程序立即开始接收。

通过在应用程序中使用推理流，您可以创建交互，在交互中用户会认为推理速度很快，因为他们能立即获得第一部分。例如，您可以创建一个聊天机器人，以增量方式显示大型语言模型 (LLM) 生成的文本。

要获取推理流，可使用 SDK for Python (Boto3) 中的 invoke_endpoint_with_response_stream 方法。在响应正文中，SDK 提供了 EventStream 对象，该对象以一系列 PayloadPart 对象的形式给出推理。

{'PayloadPart': {'Bytes': b'{"outputs": [" a"]}\n'}}
{'PayloadPart': {'Bytes': b'{"outputs": [" challenging"]}\n'}}
{'PayloadPart': {'Bytes': b'{"outputs": [" problem"]}\n'}}
. . .

{'PayloadPart': {'Bytes': b'{"outputs": '}}
{'PayloadPart': {'Bytes': b'[" problem"]}\n'}}

{"outputs": [" a"]}
{"outputs": [" challenging"]}
{"outputs": [" problem"]}
. . .

以下示例 Python 类 SmrInferenceStream 演示了如何处理以 JSON 格式发送文本数据的推理流：

import io
import json

# Example class that processes an inference stream:
class SmrInferenceStream:
    
    def __init__(self, sagemaker_runtime, endpoint_name):
        self.sagemaker_runtime = sagemaker_runtime
        self.endpoint_name = endpoint_name
        # A buffered I/O stream to combine the payload parts:
        self.buff = io.BytesIO() 
        self.read_pos = 0
        
    def stream_inference(self, request_body):
        # Gets a streaming inference response 
        # from the specified model endpoint:
        response = self.sagemaker_runtime\
            .invoke_endpoint_with_response_stream(
                EndpointName=self.endpoint_name, 
                Body=json.dumps(request_body), 
                ContentType="application/json"
        )
        # Gets the EventStream object returned by the SDK:
        event_stream = response['Body']
        for event in event_stream:
            # Passes the contents of each payload part
            # to be concatenated:
            self._write(event['PayloadPart']['Bytes'])
            # Iterates over lines to parse whole JSON objects:
            for line in self._readlines():
                resp = json.loads(line)
                part = resp.get("outputs")[0]
                # Returns parts incrementally:
                yield part
    
    # Writes to the buffer to concatenate the contents of the parts:
    def _write(self, content):
        self.buff.seek(0, io.SEEK_END)
        self.buff.write(content)

    # The JSON objects in buffer end with '\n'.
    # This method reads lines to yield a series of JSON objects:
    def _readlines(self):
        self.buff.seek(self.read_pos)
        for line in self.buff.readlines():
            self.read_pos += len(line)
            yield line[:-1]

此示例通过执行以下操作处理推理流：

以下示例创建并使用了 SmrInferenceStream 对象：

request_body = {"inputs": ["Large model inference is"],
                "parameters": {"max_new_tokens": 100,
                               "enable_sampling": "true"}}
smr_inference_stream = SmrInferenceStream(
    sagemaker_runtime, endpoint_name)
stream = smr_inference_stream.stream_inference(request_body)
for part in stream:
    print(part, end='')

此示例将请求正文传递给 stream_inference 方法。该方法将遍历响应，以打印推理流返回的每个部分。

此示例假设指定端点处的模型是生成文本的 LLM。此示例的输出是生成的文本正文，文本以增量方式打印：

a challenging problem in machine learning. The goal is to . . .

使用 Amazon CLI 调用端点

您可以通过使用 Amazon Command Line Interface (Amazon CLI) 运行命令来测试端点。Amazon CLI 支持使用 invoke-endpoint 命令发送标准推理请求，并支持使用 invoke-endpoint-async 命令发送异步推理请求。

以下示例使用 invoke-endpoint 命令，向模型端点发送推理请求：

aws sagemaker-runtime invoke-endpoint \
    --endpoint-name endpoint_name \
    --body fileb://$file_name \
    output_file.txt

对于 --endpoint-name 参数，可提供您在使用 CreateEndpoint 创建端点时为 EndpointName 指定的名称。对于 --body 参数，可提供输入数据以供 SageMaker 传递给模型。数据的格式必须与用于训练的数据格式相同。此示例显示了如何向端点发送二进制数据。

有关在将文件内容传递给 Amazon CLI 的参数时，何时使用 file:// 而非 fileb:// 的更多信息，请参阅本地文件参数的最佳实践。

有关更多信息以及可以传递的其他参数，请参阅《Amazon CLI命令参考》中的 invoke-endpoint。

如果 invoke-endpoint 命令成功，则将返回如下所示的响应：

{
    "ContentType": "<content_type>; charset=utf-8",
    "InvokedProductionVariant": "<Variant>"
}

如果命令不成功，请检查输入负载的格式是否正确。

可通过检查文件输出文件（在此例中为 output_file.txt），查看调用的输出。

more output_file.txt