실시간 추론을 위한 모델 호출

Amazon SageMaker AI를 사용하여 모델을 엔드포인트에 배포한 후 추론 요청을 전송하여 모델과 상호 작용할 수 있습니다. 모델에 추론 요청을 보내려면 추론 요청을 호스팅하는 엔드포인트를 호출합니다. 사용자는 Amazon SageMaker Studio, AWS , SDK 또는 AWS CLI를 사용하여 엔드포인트를 테스트할 수 있습니다.

Amazon SageMaker 스튜디오를 사용하여 모델 호출

모델을 엔드포인트에 배포한 후 Amazon SageMaker Studio를 통해 엔드포인트를 보고 단일 추론 요청을 전송하여 엔드포인트를 테스트할 수 있습니다.

카드 상단에는 엔드포인트로 전송된 요청 유형이 표시됩니다 (JSON만 허용). 카드에는 다음 필드가 표시됩니다.

AWS SDK for Python (Boto3)을 사용하여 모델 호출

애플리케이션 코드에서 모델 엔드포인트를 호출하려는 경우를 포함한 AWS SDKs 중 하나를 사용할 수 있습니다 AWS SDK for Python (Boto3). 이 SDK로 엔드포인트를 호출하려면 다음 Python 메서드 중 하나를 사용합니다.

애플리케이션 코드에서 이러한 메서드를 사용하려면 먼저 SageMaker AI 런타임 클라이언트를 초기화하고 엔드포인트 이름을 지정해야 합니다. 다음 예제는 이어지는 나머지 예제를 위해 클라이언트와 엔드포인트를 설정합니다.

import boto3

sagemaker_runtime = boto3.client(
    "sagemaker-runtime", region_name='aws_region')

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 AI가 모델에 전달할 Body 필드에 입력 데이터를 제공합니다. 이 데이터는 훈련에 사용한 것과 동일한 형식이어야 합니다. 이 예제에서는 response 변수에 응답을 할당합니다.

response 변수는 HTTP 상태, 배포된 모델 이름 및 기타 필드에 대한 액세스를 제공합니다. 다음 스니펫은 HTTP 상태 코드를 인쇄합니다.

print(response["HTTPStatusCode"])

호출하여 추론 응답을 스트리밍합니다.

추론 스트리밍을 지원하는 모델을 배포한 경우 사용자는 모델을 호출하여 추론 페이로드를 파트 스트림으로 수신할 수 있습니다. 모델은 모델이 이러한 부분을 생성할 때 이러한 부분을 점진적으로 제공합니다. 애플리케이션이 추론 스트림을 수신하면 애플리케이션이 모델이 전체 응답 페이로드를 생성할 때까지 기다릴 필요가 없습니다. 대신 애플리케이션은 응답이 제공되는 즉시 일부 응답을 받습니다.

애플리케이션에서 추론 스트림을 사용하면 사용자가 첫 번째 부분을 즉시 받기 때문에 추론이 빠르다고 인식할 수 있는 상호 작용을 만들 수 있습니다. 사용자는 스트리밍을 구현하여 챗봇, 가상 어시스턴트, 음악 생성기와 같은 빠른 대화형 환경을 지원할 수 있습니다. 예를 들어 대규모 언어 모델(LLM)에서 생성된 텍스트를 점진적으로 표시하는 챗봇을 만들 수 있습니다.

추론 스트림을 가져오려면 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 . . .

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 파라미터에는 엔드포인트를 생성할 때 지정한 엔드포인트 이름을 제공합니다. --body 파라미터의 경우 SageMaker AI가 모델에 전달할 입력 데이터를 제공합니다. 데이터는 훈련에 사용한 것과 동일한 형식이어야 합니다. 이 예제는 엔드포인트로 바이너리 데이터를 보내는 방법을 보여줍니다.

파일의 콘텐츠를 파라미터에 전달할 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