Richiama modelli per l'inferenza in tempo reale

Dopo aver distribuito il modello utilizzando i servizi di SageMaker hosting, puoi testarlo su quell'endpoint inviandogli i dati di test. Puoi testare i tuoi endpoint utilizzando Amazon SageMaker Studio, gli AWS SDK o il. AWS CLI

Richiama il tuo endpoint utilizzando Amazon Studio SageMaker

Dopo aver distribuito il modello su un endpoint, puoi visualizzare l'endpoint tramite Amazon SageMaker Studio e testarlo inviando singole richieste di inferenza.

Nota

SageMaker supporta solo i test degli endpoint in Studio per endpoint in tempo reale.

Per inviare una richiesta di inferenza di test al tuo endpoint

1. Avvia Amazon SageMaker Studio.

2. Nel riquadro di navigazione a sinistra, scegli Implementazioni.

3. Dal menu a discesa, scegli Endpoint.

4. Cerca il tuo endpoint per nome e scegli il nome nella tabella. I nomi degli endpoint elencati nel pannello Endpoint vengono definiti quando si distribuisce un modello. L'area di lavoro di Studio apre la pagina Endpoint in una nuova scheda.

5. Scegli la scheda Test inference.

6. Per Opzioni di test, selezionate una delle seguenti opzioni:

   1. Seleziona Verifica la richiesta di esempio per inviare immediatamente una richiesta al tuo endpoint. Utilizza l'editor JSON per fornire dati di esempio in formato JSON e scegli Invia richiesta per inviare la richiesta al tuo endpoint. Dopo aver inviato la richiesta, Studio mostra l'output dell'inferenza in una scheda a destra dell'editor JSON.

   2. Seleziona Usa codice di esempio Python SDK per visualizzare il codice per l'invio di una richiesta all'endpoint. Quindi, copia l'esempio di codice dalla sezione Richiesta di inferenza di esempio ed esegui il codice dal tuo ambiente di test.

La parte superiore della scheda mostra il tipo di richiesta che è stata inviata all'endpoint (è accettato solo JSON). La scheda mostra i seguenti campi:

• Stato: visualizza uno dei seguenti tipi di stato:

  • Success: la richiesta è riuscita.

  • Failed: esito negativo della richiesta. Una risposta viene visualizzata in Motivo dell'errore.

  • Pending: mentre la richiesta di inferenza è in sospeso, lo stato mostra un'icona circolare rotante.

• Durata dell'esecuzione: quanto tempo è durato l'invocazione (ora di fine meno l'ora di inizio) in millisecondi.

• Ora della richiesta: quanti minuti sono trascorsi dall'invio della richiesta.

• Ora del risultato: quanti minuti sono trascorsi dalla restituzione del risultato.

Richiama il tuo endpoint utilizzando il AWS SDK for Python (Boto3)

Dopo aver distribuito il modello su un endpoint, puoi controllarlo utilizzando uno degli AWS SDK, incluso come. AWS SDK for Python (Boto3) Per testare l'endpoint con questo SDK, utilizza uno dei seguenti metodi:

• invoke_endpoint: invia una richiesta di inferenza a un endpoint del modello e restituisce la risposta generata dal modello. Questo metodo restituisce il payload di inferenza come una risposta dopo che il modello ha terminato la sua generazione. Per ulteriori informazioni, consulta invoke_endpoint in AWS SDK for Python (Boto) API Reference.

• invoke_endpoint_with_response_stream: invia una richiesta di inferenza a un endpoint del modello e trasmette la risposta in parti incrementali mentre il modello genera l'inferenza. Con questo metodo, l'applicazione client inizia immediatamente a ricevere parti della risposta non appena le parti diventano disponibili. Il cliente non deve attendere che il modello generi l'intero payload di risposta. Puoi implementare lo streaming per supportare esperienze interattive veloci, come chatbot, assistenti virtuali e generatori di musica.

  Utilizza questo metodo solo per richiamare modelli che supportano lo streaming inferenziale.

  Quando un container gestisce una richiesta di inferenza in streaming, restituisce l'inferenza del modello come una serie di parti in modo incrementale man mano che il modello le genera. Le applicazioni client iniziano a ricevere risposte immediatamente quando sono disponibili. Non è necessario attendere che il modello generi l'intera risposta. Puoi implementare lo streaming per supportare esperienze interattive veloci, come chatbot, assistenti virtuali e generatori di musica.

Prima di poter utilizzare questi metodi nel codice client, è necessario creare un client SageMaker Runtime e specificare il nome dell'endpoint. L'esempio seguente configura il client e l'endpoint per il resto degli esempi seguenti:

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'

Richiamare per ottenere una risposta di inferenza

L'esempio seguente utilizza il metodo invoke_endpoint per richiamare un endpoint con 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'))

Questo esempio fornisce dati di input nel Body campo SageMaker da passare al modello. Questi dati devono essere nello stesso formato utilizzato per l'addestramento. L'esempio memorizza la risposta nella variabile response.

La variabile response fornisce l'accesso allo stato HTTP, al nome del modello distribuito e ad altri campi. Il seguente snippet stampa HTTPStatusCode:

print(response["HTTPStatusCode"])

Richiamare per trasmettere una risposta di inferenza

Se hai distribuito un modello che supporta lo streaming di inferenza, puoi richiamare il modello per ricevere il payload di inferenza sotto forma di flusso di parti. Il modello fornisce queste parti in modo incrementale man mano che il modello le genera. Quando un'applicazione riceve un flusso di inferenza, non deve attendere che il modello generi l'intero payload di risposta. Invece, l'applicazione inizia immediatamente a ricevere parti della risposta non appena diventano disponibili.

Utilizzando un flusso di inferenza nell'applicazione, è possibile creare interazioni in cui gli utenti percepiscono l'inferenza come rapida perché ottengono immediatamente la prima parte. Ad esempio, è possibile creare un chatbot che mostri in modo incrementale il testo generato da un modello di linguaggio di grandi dimensioni (LLM).

Per ottenere un flusso di inference, puoi utilizzare il metodo invoke_endpoint_with_response_stream nell'SDK per Python (Boto3). Nel corpo della risposta, l'SDK fornisce un oggetto EventStream, che fornisce l'inferenza sotto forma di una serie di oggetti PayloadPart.

Esempio Flusso di inferenza

Di seguito è riportato un esempio di un flusso di oggetti PayloadPart:

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

In ogni parte del payload, il campo Bytes fornisce una parte della risposta di inferenza del modello. Questa parte può essere qualsiasi tipo di contenuto generato da un modello, ad esempio testo, immagini o dati audio. In questo esempio, le porzioni sono oggetti JSON che contengono testo generato da un LLM.

Di solito, la parte del payload contiene una parte discreta di dati del modello. In questo esempio i blocchi discreti sono oggetti JSON interi. Occasionalmente, la risposta in streaming divide i blocchi su più parti del payload oppure combina più blocchi in un'unica parte del payload. L'esempio seguente mostra un blocco di dati in formato JSON suddiviso in due parti del payload:

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

Quando scrivi codice applicativo che elabora un flusso di inferenza, includi la logica che gestisca queste suddivisioni e combinazioni occasionali di dati. Come strategia, potete scrivere codice che concatena i contenuti di Bytes mentre l'applicazione riceve le parti del payload. Concatenando qui i dati JSON di esempio, puoi combinare i dati in un corpo JSON delimitato da una nuova riga. Quindi, il codice potrebbe elaborare lo stream analizzando l'intero oggetto JSON su ogni riga.

L'esempio seguente mostra il codice JSON delimitato da nuove righe che creereste quando concatenate il contenuto di esempio di Bytes:

{"outputs": [" a"]}
{"outputs": [" challenging"]}
{"outputs": [" problem"]}
. . .

Esempio Codice per elaborare un flusso di inferenza

Il seguente esempio di classe Python, SmrInferenceStream, dimostra come è possibile elaborare un flusso di inferenza che invia dati di testo in formato 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]

Questo esempio elabora il flusso di inferenza effettuando le seguenti operazioni:

• Viene inizializzato con un client SageMaker Runtime e il nome di un endpoint del modello. Prima di poter ottenere un flusso di inferenza, il modello ospitato dall'endpoint deve supportare lo streaming di inferenza.

• Nel metodo stream_inference di esempio, riceve il corpo della richiesta e lo passa al metodo invoke_endpoint_with_response_stream dell'SDK.

• Itera su ogni evento nell'oggetto EventStream restituito dall'SDK.

• Da ogni evento, ottiene il contenuto dell'oggetto Bytes nell'oggetto PayloadPart.

• Nel metodo _write di esempio, scrive in un buffer per concatenare il contenuto degli oggetti Bytes. I contenuti combinati formano un corpo JSON delimitato da una nuova riga.

• Utilizza il metodo _readlines di esempio per ottenere una serie iterabile di oggetti JSON.

• In ogni oggetto JSON, ottiene una parte dell'inferenza.

• Con l'espressione yield, restituisce i pezzi in modo incrementale.

L'esempio seguente crea e utilizza un oggetto 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='')

Questo esempio passa un corpo della richiesta al metodo stream_inference. Esegue un'iterazione sulla risposta per stampare ogni pezzo restituito dal flusso di inferenza.

L'esempio presuppone che il modello nell'endpoint specificato sia un LLM che genera testo. L'output di questo esempio è un corpo di testo generato che viene stampato in modo incrementale:

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

Invoca il tuo endpoint utilizzando il AWS CLI

Puoi testare il tuo endpoint eseguendo comandi con AWS Command Line Interface ()AWS CLI. AWS CLI supporta richieste di inferenza standard con il comando invoke-endpoint e supporta richieste di inferenza asincrone con il comando invoke-endpoint-async.

Nota

AWS CLI Non supporta le richieste di inferenza in streaming.

L'esempio seguente utilizza il comando invoke-endpoint per inviare una richiesta di inferenza a un endpoint del modello:

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

Per il parametro --endpoint-name, fornisci il nome che hai specificato per EndpointName quando hai creato l'endpoint con CreateEndpoint. Per il --body parametro, fornite i dati di input SageMaker da passare al modello. I dati devono essere nello stesso formato utilizzato per l'addestramento. Questo esempio mostra come inviare dati binari all'endpoint.

Per ulteriori informazioni su quando utilizzare file:// over fileb:// quando si passa il contenuto di un file a un parametro di AWS CLI, consultate Best Practices for Local File Parameters.

Per ulteriori informazioni e per visualizzare i parametri aggiuntivi che è possibile passare, consulta invoke-endpoint nel Riferimento al comando AWS CLI .

Se il comando invoke-endpoint ha esito positivo, restituisce una risposta come la seguente:

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

Se il comando non ha esito positivo, controlla se il payload di input è nel formato corretto.

Visualizza l'output della chiamata controllando il file di output del file (output_file.txt in questo esempio).

more output_file.txt