使用 Studio 经典版调使用 SDK for Python (Boto3) 调用使用调用 AWS CLI

调用模型进行实时推理

使用 SageMaker 托管服务部署模型后，您可以通过向该端点发送测试数据，在该端点上测试您的模型。您可以使用 Amazon SageMaker Studio、 AWS 软件开发工具包或测试您的终端节点。 AWS CLI

使用亚马逊 SageMaker Studio 调用您的终端节点

将模型部署到终端节点后，您可以通过 Amazon SageMaker Studio 查看终端节点，并通过发送单个推理请求来测试您的终端节点。

注意

SageMaker 仅支持在 Studio 中对实时端点进行端点测试。

向端点发送测试推理请求

启动 Amazon SageMaker Studio。
在左侧导航窗格中，选择 “部署”。
从下拉菜单中选择端点。
按名称查找端点，然后在表中选择名称。端点面板中列出的端点名称是在部署模型时定义的。Studio 工作区在新选项卡中打开端点页面。
选择测试推理选项卡。
在 “测试选项” 中，选择以下选项之一：
1. 选择测试示例请求以立即向您的终端节点发送请求。使用 JSON 编辑器提供 JSON 格式的示例数据，然后选择发送请求将请求提交到您的终端节点。提交请求后，Studio 会在 JSON 编辑器右侧的卡片中显示推理输出。
2. 选择 “使用 Python SDK 示例代码”，查看向终端节点发送请求的代码。然后，复制推理请求示例部分中的代码示例，然后从您的测试环境中运行代码。

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

状态 – 显示以下状态类型之一：
- Success – 请求成功。
- Failed – 请求失败。响应显示在失败原因下方。
- Pending – 当推理请求处于待处理状态时，状态会显示一个旋转的圆形图标。
执行时长 – 调用耗费的时间（结束时间减去开始时间），以毫秒为单位。
请求时间 – 自发送请求以来过去的分钟数。
结果时间 – 自返回结果以来过去的分钟数。

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

将模型部署到终端节点后，您可以使用其中一个 AWS 软件开发工具包来检查您的终端节点，包括作为。 AWS SDK for Python (Boto3)要使用此 SDK 测试端点，请使用以下方法之一：

invoke_endpoint – 向模型端点发送推理请求，并返回模型生成的响应。在模型完成生成推理负载后，此方法将其作为一个响应返回。有关更多信息，请参阅《AWS SDK for Python (Boto3) API 参考》中的 invoke_endpoint。
invoke_endpoint_with_response_stream – 向模型端点发送推理请求，并在模型生成推理的同时，以增量部分流式传输响应。如使用此方法，当部分响应变得可用时，客户端应用程序会立即开始接收。客户端无需等待模型生成整个响应负载。您可以实现流式处理，以支持快速的交互式体验，如聊天机器人、虚拟助手和音乐生成器。

此方法仅用于调用支持推理流的模型。

当容器处理流式推理请求时，它会在模型生成推理时，以递增形式返回一系列的内容，每个内容都是一部分模型推理。客户端应用程序会在相关响应可用时立即开始接收响应。它们无需等待模型生成完整的响应。您可以实施流式处理以支持快速的交互式体验，例如聊天机器人、虚拟助手和音乐生成器。

在客户端代码中使用这些方法之前，必须创建一个 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 方法，通过 AWS 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 对象流：


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

在每个负载部分中，Bytes 字段提供模型推理响应的一部分。此部分可以是模型生成的任何内容类型，如文本、图像或音频数据。在此示例中，这些部分是 JSON 对象，其中包含 LLM 生成的文本。

通常，负载部分包含来自模型的离散数据块。在本示例中，离散块是整个 JSON 对象。有时，流媒式响应会将数据块分成多个负载部分，或者将多个数据块组合成一个负载部分。以下示例显示了一个 JSON 格式的数据块，该数据块分为两个负载部分：


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

在编写处理推理流的应用程序代码时，应包括处理这些偶尔的数据拆分和组合的逻辑。作为一种策略，您可以编写代码，在应用程序接收负载部分的同时，连接 Bytes 的内容。通过连接此处的示例 JSON 数据，可以将这些数据组合成一个以换行符分隔的 JSON 正文。然后，您的代码可以通过解析每行上的整个 JSON 对象来处理流。

以下示例显示了您在连接 Bytes 的以下示例内容时，创建的以换行符分隔的 JSON：


{"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]

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

使用 SageMaker Runtime 客户端和模型端点的名称进行初始化。在获得推理流之前，端点托管的模型必须支持推理流。
在示例 stream_inference 方法中，接收请求正文并将其传递给 SDK 的 invoke_endpoint_with_response_stream 方法。
遍历 SDK 返回的 EventStream 对象中的每个事件。
从每个事件中获取 PayloadPart 对象中 Bytes 对象的内容。
在示例 _write 方法中，写入缓冲区以连接 Bytes 对象的内容。组合后的内容构成以换行符分隔的 JSON 正文。
使用示例 _readlines 方法获取一系列可迭代的 JSON 对象。
在每个 JSON 对象中，获取推理的一部分。
使用 yield 表达式，以增量方式返回这些部分。

以下示例创建并使用了 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 . . .

使用调用您的终端节点 AWS CLI

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

注意

AWS CLI 不支持流式推理请求。

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


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

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

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

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

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


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

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

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


more output_file.txt

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

文档惯例

部署模型

管理终端节点