Memanggil model untuk inferensi waktu nyata

Setelah menggunakan Amazon SageMaker AI untuk menerapkan model ke titik akhir, Anda dapat berinteraksi dengan model dengan mengirimkan permintaan inferensi ke model tersebut. Untuk mengirim permintaan inferensi ke model, Anda memanggil titik akhir yang menghostingnya. Anda dapat memanggil titik akhir Anda menggunakan Amazon SageMaker Studio, the AWS SDKs, atau. AWS CLI

Panggil Model Anda Menggunakan Amazon Studio SageMaker

Setelah menerapkan model ke titik akhir, Anda dapat melihat titik akhir melalui Amazon SageMaker Studio dan menguji titik akhir dengan mengirimkan permintaan inferensi tunggal.

Bagian atas kartu menunjukkan jenis permintaan yang dikirim ke titik akhir (hanya JSON yang diterima). Kartu menunjukkan bidang-bidang berikut:

Panggil Model Anda dengan Menggunakan AWS SDK untuk Python (Boto3)

Jika Anda ingin memanggil titik akhir model dalam kode aplikasi Anda, Anda dapat menggunakan salah satunya AWSSDKs, termasuk. AWS SDK untuk Python (Boto3) Untuk memanggil endpoint Anda dengan SDK ini, Anda menggunakan salah satu metode Python berikut:

Sebelum Anda dapat menggunakan metode ini dalam kode aplikasi Anda, Anda harus menginisialisasi klien SageMaker AI Runtime, dan Anda harus menentukan nama titik akhir Anda. Contoh berikut mengatur klien dan titik akhir untuk sisa contoh berikut:

import boto3

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

endpoint_name='endpoint-name'

Memohon untuk Mendapatkan Respons Inferensi

Contoh berikut menggunakan invoke_endpoint metode untuk memanggil titik akhir dengan: AWS SDK untuk 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'))

Contoh ini memberikan data input di Body lapangan agar SageMaker AI dapat diteruskan ke model. Data ini harus dalam format yang sama dengan yang digunakan untuk pelatihan. Contoh memberikan respon terhadap response variabel.

responseVariabel menyediakan akses ke status HTTP, nama model yang digunakan, dan bidang lainnya. Cuplikan berikut mencetak kode status HTTP:

print(response["HTTPStatusCode"])

Memanggil untuk Streaming Respons Inferensi

Jika Anda menerapkan model yang mendukung streaming inferensi, Anda dapat memanggil model untuk menerima muatan inferensinya sebagai aliran suku cadang. Model memberikan bagian-bagian ini secara bertahap saat model menghasilkannya. Saat aplikasi menerima aliran inferensi, aplikasi tidak perlu menunggu model menghasilkan seluruh muatan respons. Sebagai gantinya, aplikasi segera menerima bagian dari respons saat tersedia.

Dengan menggunakan aliran inferensi dalam aplikasi Anda, Anda dapat membuat interaksi di mana pengguna Anda menganggap inferensi menjadi cepat karena mereka mendapatkan bagian pertama dengan segera. Anda dapat menerapkan streaming untuk mendukung pengalaman interaktif yang cepat, seperti chatbots, asisten virtual, dan generator musik. Misalnya, Anda dapat membuat chatbot yang secara bertahap menampilkan teks yang dihasilkan oleh model bahasa besar (LLM).

Untuk mendapatkan aliran inferensi, Anda dapat menggunakan invoke_endpoint_with_response_stream metode ini. Dalam badan respons, SDK menyediakan EventStream objek, yang memberikan inferensi sebagai serangkaian PayloadPart objek.

{'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"]}
. . .

Contoh kelas Python berikut,SmrInferenceStream, menunjukkan bagaimana Anda dapat memproses aliran inferensi yang mengirimkan data teks dalam format 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]

Contoh ini memproses aliran inferensi dengan melakukan hal berikut:

Contoh berikut membuat dan menggunakan SmrInferenceStream objek:

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

Contoh ini meneruskan badan permintaan ke stream_inference metode. Ini mengulangi respons untuk mencetak setiap bagian yang dikembalikan oleh aliran inferensi.

Contoh mengasumsikan bahwa model pada titik akhir yang ditentukan adalah LLM yang menghasilkan teks. Output dari contoh ini adalah kumpulan teks yang dihasilkan yang mencetak secara bertahap:

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

Panggil Model Anda dengan Menggunakan AWS CLI

Anda dapat memanggil titik akhir model Anda dengan menjalankan perintah dengan AWS Command Line Interface ()AWS CLI. AWS CLIMendukung permintaan inferensi standar dengan invoke-endpoint perintah, dan mendukung permintaan inferensi asinkron dengan perintah. invoke-endpoint-async

Contoh berikut menggunakan invoke-endpoint perintah untuk mengirim permintaan inferensi ke titik akhir model:

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

Untuk --endpoint-name parameter, berikan nama titik akhir yang Anda tentukan saat Anda membuat titik akhir. Untuk --body parameter, berikan data input agar SageMaker AI dapat diteruskan ke model. Data harus dalam format yang sama dengan yang digunakan untuk pelatihan. Contoh ini menunjukkan cara mengirim data biner ke titik akhir Anda.

Untuk informasi selengkapnya tentang fileb:// kapan harus menggunakan file:// over saat meneruskan konten file ke parameterAWS CLI, lihat Praktik Terbaik untuk Parameter File Lokal.

Untuk informasi selengkapnya, dan untuk melihat parameter tambahan yang dapat Anda lewati, lihat invoke-endpointdi AWS CLICommand Reference.

Jika invoke-endpoint perintah berhasil mengembalikan respon seperti berikut:

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

Jika perintah tidak berhasil, periksa apakah muatan input dalam format yang benar.

Lihat output dari pemanggilan dengan memeriksa file output file (output_file.txtdalam contoh ini).

more output_file.txt

Memanggil Model Anda dengan Menggunakan AWS SDK untuk Python

Memanggil untuk Streaming Dua Arah Permintaan Inferensi dan Respons

Jika Anda ingin menjalankan endpoint model dalam kode aplikasi Anda untuk mendukung streaming dua arah, Anda dapat menggunakan SDK eksperimental baru untuk Python yang mendukung kemampuan streaming dua arah dengan dukungan HTTP/2. SDK ini memungkinkan komunikasi dua arah real-time antara aplikasi klien Anda dan SageMaker titik akhir, memungkinkan Anda mengirim permintaan inferensi secara bertahap sambil secara bersamaan menerima respons streaming saat model menghasilkannya. Ini sangat berguna untuk aplikasi interaktif di mana klien dan server perlu bertukar data terus menerus melalui koneksi persisten.

Untuk memanggil titik akhir Anda dengan streaming dua arah, gunakan metode ini. invoke_endpoint_with_bidirectional_stream Metode ini membuat koneksi persisten yang memungkinkan Anda mengalirkan beberapa potongan payload ke model Anda sambil menerima respons secara real-time saat model memproses data. Koneksi tetap terbuka sampai Anda secara eksplisit menutup aliran input atau titik akhir menutup koneksi, mendukung hingga 30 menit waktu koneksi.

Prasyarat

Sebelum Anda dapat menggunakan streaming dua arah dalam kode aplikasi Anda, Anda harus:

Siapkan klien streaming dua arah

Contoh berikut menunjukkan cara menginisialisasi komponen yang diperlukan untuk streaming dua arah:

from sagemaker_runtime_http2.client import SageMakerRuntimeHTTP2Client
from sagemaker_runtime_http2.config import Config, HTTPAuthSchemeResolver
from smithy_aws_core.identity import EnvironmentCredentialsResolver
from smithy_aws_core.auth.sigv4 import SigV4AuthScheme

# Configuration
AWS_REGION = "us-west-2"
BIDI_ENDPOINT = f"https://runtime.sagemaker.{AWS_REGION}.amazonaws.com:8443"
ENDPOINT_NAME = "your-endpoint-name"

# Initialize the client configuration
config = Config(
    endpoint_uri=BIDI_ENDPOINT,
    region=AWS_REGION,
    aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
    auth_scheme_resolver=HTTPAuthSchemeResolver(),
    auth_schemes={"aws.auth#sigv4": SigV4AuthScheme(service="sagemaker")}
)

# Create the SageMaker Runtime HTTP/2 client
client = SageMakerRuntimeHTTP2Client(config=config)

Klien streaming dua arah lengkap

Contoh berikut menunjukkan cara membuat klien streaming dua arah yang mengirimkan beberapa muatan teks ke SageMaker titik akhir dan memproses respons secara real-time:

import asyncio
import logging
from sagemaker_runtime_http2.client import SageMakerRuntimeHTTP2Client
from sagemaker_runtime_http2.config import Config, HTTPAuthSchemeResolver
from sagemaker_runtime_http2.models import (
    InvokeEndpointWithBidirectionalStreamInput, 
    RequestStreamEventPayloadPart, 
    RequestPayloadPart
)
from smithy_aws_core.identity import EnvironmentCredentialsResolver
from smithy_aws_core.auth.sigv4 import SigV4AuthScheme

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SageMakerBidirectionalClient:
    
    def __init__(self, endpoint_name, region="us-west-2"):
        self.endpoint_name = endpoint_name
        self.region = region
        self.client = None
        self.stream = None
        self.response_task = None
        self.is_active = False
        
    def _initialize_client(self):
        bidi_endpoint = f"runtime.sagemaker.{self.region}.amazonaws.com:8443"
        config = Config(
            endpoint_uri=bidi_endpoint,
            region=self.region,
            aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
            auth_scheme_resolver=HTTPAuthSchemeResolver(),
            auth_schemes={"aws.auth#sigv4": SigV4AuthScheme(service="sagemaker")}
        )
        self.client = SageMakerRuntimeHTTP2Client(config=config)
    
    async def start_session(self):
        """Establish a bidirectional streaming connection with the endpoint."""
        if not self.client:
            self._initialize_client()
            
        logger.info(f"Starting session with endpoint: {self.endpoint_name}")
        self.stream = await self.client.invoke_endpoint_with_bidirectional_stream(
            InvokeEndpointWithBidirectionalStreamInput(endpoint_name=self.endpoint_name)
        )
        self.is_active = True
        
        # Start processing responses concurrently
        self.response_task = asyncio.create_task(self._process_responses())
    
    async def send_message(self, message):
        """Send a single message to the endpoint."""
        if not self.is_active:
            raise RuntimeError("Session not active. Call start_session() first.")
            
        logger.info(f"Sending message: {message}")
        payload = RequestPayloadPart(bytes_=message.encode('utf-8'))
        event = RequestStreamEventPayloadPart(value=payload)
        await self.stream.input_stream.send(event)
    
    async def send_multiple_messages(self, messages, delay=1.0):
        """Send multiple messages with a delay between each."""
        for message in messages:
            await self.send_message(message)
            await asyncio.sleep(delay)
    
    async def end_session(self):
        """Close the bidirectional streaming connection."""
        if not self.is_active:
            return
            
        await self.stream.input_stream.close()
        self.is_active = False
        logger.info("Stream closed")
        
        # Cancel the response processing task
        if self.response_task and not self.response_task.done():
            self.response_task.cancel()
    
    async def _process_responses(self):
        """Process incoming responses from the endpoint."""
        try:
            output = await self.stream.await_output()
            output_stream = output[1]
            
            while self.is_active:
                result = await output_stream.receive()
                
                if result is None:
                    logger.info("No more responses")
                    break
                
                if result.value and result.value.bytes_:
                    response_data = result.value.bytes_.decode('utf-8')
                    logger.info(f"Received: {response_data}")
                    
        except Exception as e:
            logger.error(f"Error processing responses: {e}")

# Example usage
async def run_bidirectional_client():
    client = SageMakerBidirectionalClient(endpoint_name="your-endpoint-name")
    
    try:
        # Start the session
        await client.start_session()
        
        # Send multiple messages
        messages = [
            "I need help with", 
            "my account balance", 
            "I can help with that", 
            "and recent charges"
        ]
        await client.send_multiple_messages(messages)
        
        # Wait for responses to be processed
        await asyncio.sleep(2)
        
        # End the session
        await client.end_session()
        logger.info("Session ended successfully")
        
    except Exception as e:
        logger.error(f"Client error: {e}")
        await client.end_session()

if __name__ == "__main__":
    asyncio.run(run_bidirectional_client())

Klien menginisialisasi klien SageMaker Runtime HTTP/2 dengan URI endpoint regional pada port 8443, yang diperlukan untuk koneksi streaming dua arah. session()Metode start_ memanggil invoke_endpoint_with_bidirectional_stream() untuk membuat koneksi persisten dan membuat tugas asinkron untuk memproses respons yang masuk secara bersamaan.

send_event()Metode ini membungkus data payload dalam objek permintaan yang sesuai dan mengirimkannya melalui aliran input, sementara _process_responses() metode terus mendengarkan dan memproses respons dari titik akhir saat mereka tiba. Pendekatan dua arah ini memungkinkan interaksi real-time di mana baik mengirim permintaan dan menerima tanggapan terjadi secara bersamaan melalui koneksi yang sama.