リアルタイム推論用のモデルを呼び出す

SageMaker ホスティングサービスを使用してモデルをデプロイした後、テストデータを送信することで、そのエンドポイントでモデルをテストできます。Amazon SageMaker Studio、 AWS SDKsを使用してエンドポイントをテストできます AWS CLI。

Amazon SageMaker Studio を使用してエンドポイントを呼び出す

モデルをエンドポイントにデプロイした後、Amazon SageMaker Studio を介してエンドポイントを表示し、単一の推論リクエストを送信してエンドポイントをテストできます。

カード上部に、エンドポイントに送信されたリクエストのタイプが表示されます (現在は JSON のみが受け入れられます)。カードには次のフィールドが表示されます。

AWS SDK for Python (Boto3)を使用してエンドポイントを呼び出す

モデルをエンドポイントにデプロイした後、 を含む AWS SDKsのいずれかを使用してエンドポイントを確認できます AWS SDK for Python (Boto3)。この SDK を使用してエンドポイントをテストするには、次のいずれかのメソッドを使用します。

これらのメソッドをクライアントコードで使用する前に、 SageMaker ランタイムクライアントを作成し、エンドポイントの名前を指定する必要があります。次の例で、残りの例のクライアントとエンドポイントを設定します。

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

この例では、 がモデル SageMaker に渡す Bodyフィールドに入力データを提供します。このデータは、トレーニングに使われたのと同じ形式である必要があります。例では、レスポンスを response 変数に保存しています。

response 変数は、HTTP ステータス、デプロイされたモデルの名前、および他のフィールドへのアクセスを提供します。次のスニペットは HTTPStatusCode を出力します。

print(response["HTTPStatusCode"])

推論レスポンスをストリーミングするために呼び出す

推論ストリーミングをサポートするモデルをデプロイした場合、そのモデルを呼び出して、その推論ペイロードをパートのストリームとして受け取ることができます。モデルがこれらのパートを生成すると、段階的に配信されます。推論ストリームを受信する場合、アプリケーションはモデルがレスポンスペイロード全体を生成するのを待つ必要はありません。代わりに、アプリケーションはレスポンスの一部が利用可能になるとすぐに受信し始めます。

アプリケーションで推論ストリームを使用することで、ユーザーは最初のパートをすぐに取得できるので、推論が速いと認識できるインタラクションを作成できます。例えば、大規模言語モデル (LLM) が生成するテキストを段階的に表示するチャットボットを作成できます。

推論ストリームを取得するために、SDK for Python (Boto3) invoke_endpoint_with_response_stream のメソッドを使用できます。レスポンスの本文で、SDK は推論を一連の PayloadPart オブジェクトとして出力する EventStream オブジェクトを提供します。

{'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 . . .

を使用してエンドポイントを呼び出す AWS CLI

AWS Command Line Interface () でコマンドを実行することで、エンドポイントをテストできますAWS CLI。 AWS 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 に渡す の入力データを指定します。データは、トレーニングに使われたのと同じ形式である必要があります。この例は、バイナリデータをエンドポイントに送信する方法を示しています。

ファイルの内容を のパラメータに渡すfileb://ときに file://をいつ上書きするかの詳細については AWS CLI、「ローカルファイルパラメータのベストプラクティス」を参照してください。

渡すことができる追加のパラメータの詳細については、「AWS CLI コマンドリファレンス」の「invoke-endpoint」を参照してください。

invoke-endpoint コマンドが成功すると、次のようなレスポンスが返されます。

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

コマンドが成功しない場合は、入力ペイロードが正しい形式かどうかを確認してください。

ファイル出力ファイル (この例では output_file.txt) を確認することで、呼び出しの出力を表示します。

more output_file.txt