使用工作室經典調用使用適用於 Python 的 SDK (Boto3) 調用使用呼叫 AWS CLI

叫用模型以進行即時推論

使用 SageMaker 託管服務部署模型後，您可以透過傳送測試資料在該端點上測試模型。您可以使用 Amazon SageMaker 工作室、開 AWS 發套件或. AWS CLI

使用 Amazon SageMaker 工作室調用端點

將模型部署到端點後，您可以透過 Amazon SageMaker Studio 檢視端點，並透過傳送單一推論請求來測試您的端點。

注意

SageMaker 僅支援 Studio 中針對即時端點進行端點測試。

將測試推論請求傳送至您的端點

啟動 Amazon SageMaker 工作室。
在左側的導覽窗格中，選擇「部署」。
從下拉式清單中，選擇端點。
依名稱尋找您的端點，然後在資料表中選擇名稱。端點面板中列出的端點名稱是在部署模型時所定義的。Studio 工作區會在新索引標籤中開啟 [端點] 頁面。
選擇 [測試推論] 索引標籤。
對於「測試選項」，請選取下列其中一項：
1. 選取 [測試範例請求]，立即將要求傳送至您的端點。使用 JSON 編輯器提供 JSON 格式的範例資料，然後選擇「傳送請求」將要求提交至您的端點。提交請求後，Studio 會在 JSON 編輯器右側的卡片中顯示推論輸出。
2. 選取「使用 Python SDK 範例程式碼」，以檢視傳送要求至端點的程式碼。然後，從「示例推論請求」部分複製代碼示例，然後從測試環境中運行代碼。

卡片頂端會顯示傳送至端點的請求類型 (僅接受 JSON)。卡片會顯示下列欄位：

狀態 — 會顯示以下其中一項狀態類型：
- Success — 請求已成功。
- Failed — 請求失敗。回應會顯示在 Failure Reason (失敗原因) 底下。
- Pending — 當推論請求處於待處理時，狀態會顯示旋轉的圓形圖示。
執行長度 — 調用所花費的時間 (結束時間減去開始時間)，以毫秒為單位。
請求時間 — 自發送請求以來已過了多少分鐘。
結果時間 — 自傳回結果後已過了多少分鐘。

使用 AWS SDK for Python (Boto3)調用您的端點

將模型部署到端點後，您可以使用其中一個 AWS SDK 來檢查端點，包括. AWS SDK for Python (Boto3)若要使用此 SDK 測試您的端點，請使用下列其中一種方法：

invoke_endpoint — 將推論請求傳送至模型端點，並傳回模型產生的回應。此方法會傳回推論承載，以作為模型完成生成後的一個回應。如需更多資訊，請參閱適用於 Python 的AWS SDK (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) 產生的文字。

若要取得推論串流，您可以使用適用於 Python 的 SDK (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 運行時客戶端和模型端點的名稱進行初始化。取得推論串流之前，端點託管的模型必須支援推論串流。
在範例 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 要傳遞至模型的輸入資料。資料的格式必須與用於訓練的格式相同。此範例顯示如何將二進位資料傳送至您的端點。

如需將檔案內容傳遞至file://的參數fileb://時何時使用的詳細資訊 AWS CLI，請參閱本機檔案參數的最佳作法。

若要取得更多資訊，並查看您可以傳遞的其他參數，請參閱 AWS CLI 命令推論中的 invoke-endpoint。

如果 invoke-endpoint 命令成功，它會傳回類似以下的回應：


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

如果命令未成功，請檢查輸入承載的格式是否正確。

檢查檔案輸出檔案來檢視調用的輸出 (在此範例中為 output_file.txt)。


more output_file.txt

您的瀏覽器已停用或無法使用 Javascript。

您必須啟用 Javascript，才能使用 AWS 文件。請參閱您的瀏覽器說明頁以取得說明。

文件慣用形式

部署模型

管理端點