Routine di previsione personalizzate

Le routine di previsione personalizzate (CPR) consentono di creare facilmente container personalizzati con codice di pre/post-elaborazione, senza dover gestire i dettagli della configurazione di un server HTTP o della creazione di un container da zero. Puoi utilizzare la pre-elaborazione per normalizzare/trasformare gli input o effettuare chiamate a servizi esterni per ottenere dati aggiuntivi e utilizzare la post-elaborazione per formattare la previsione del modello o eseguire la logica di business.

Il seguente diagramma illustra il flusso di lavoro dell'utente con e senza routine di previsione personalizzate.

Le principali differenze sono:

  • Non è necessario scrivere un server del modello o un Dockerfile. Ti viene fornito il server del modello, ovvero il server HTTP che ospita il modello.

  • Puoi eseguire il deployment del modello ed eseguirne il debug localmente, accelerando il ciclo di iterazione durante lo sviluppo.

Crea ed esegui il deployment di un container personalizzato

Questa sezione descrive come utilizzare CPR per creare un container personalizzato con logica di pre/post-elaborazione e implementarlo su un endpoint locale e online.

Configurazione

Devi aver installato l'SDK Vertex AI e il Docker nel tuo ambiente.

Scrivi Predictor personalizzato

Implementa l'interfaccia di Predictor.

class Predictor(ABC):
    """Interface of the Predictor class for Custom Prediction Routines.
    The Predictor is responsible for the ML logic for processing a prediction request.
    Specifically, the Predictor must define:
    (1) How to load all model artifacts used during prediction into memory.
    (2) The logic that should be executed at predict time.
    When using the default PredictionHandler, the Predictor will be invoked as follows:
      predictor.postprocess(predictor.predict(predictor.preprocess(prediction_input)))
    """

    @abstractmethod
    def load(self, artifacts_uri: str) -> None:
        """Loads the model artifact.
        Args:
            artifacts_uri (str):
                Required. The value of the environment variable AIP_STORAGE_URI.
        """
        pass

    def preprocess(self, prediction_input: Any) -> Any:
        """Preprocesses the prediction input before doing the prediction.
        Args:
            prediction_input (Any):
                Required. The prediction input that needs to be preprocessed.
        Returns:
            The preprocessed prediction input.
        """
        return prediction_input

    @abstractmethod
    def predict(self, instances: Any) -> Any:
        """Performs prediction.
        Args:
            instances (Any):
                Required. The instance(s) used for performing prediction.
        Returns:
            Prediction results.
        """
        pass

    def postprocess(self, prediction_results: Any) -> Any:
        """Postprocesses the prediction results.
        Args:
            prediction_results (Any):
                Required. The prediction results.
        Returns:
            The postprocessed prediction results.
        """
        return prediction_results

Ad esempio, vedi Implementazione di Predictor di Sklearn.

Scrivi Handler personalizzato (facoltativo)

I gestori personalizzati hanno accesso all'oggetto della richiesta non elaborata e quindi sono utili nei rari casi in cui sia necessario personalizzare la logica correlata al server web, ad esempio il supporto di intestazioni di richiesta/risposta aggiuntive o la deserializzazione delle richieste di previsione in formato non JSON.

Ecco un blocco note di esempio che implementa sia il predittore sia il gestore.

Sebbene non sia obbligatorio, per una migliore organizzazione e riusabilità del codice, consigliamo di implementare la logica del server web nel gestore e la logica ML nel predittore, come mostrato nel gestore predefinito.

Crea container personalizzato

Inserisci il codice personalizzato e un file requirements.txt aggiuntivo, se devi installare pacchetti nelle immagini, in una directory.

Utilizza l'SDK Vertex per creare container personalizzati come mostrato di seguito:

from google.cloud.aiplatform.prediction import LocalModel

# {import your predictor and handler}

local_model = LocalModel.build_cpr_model(
    {PATH_TO_THE_SOURCE_DIR},
    f"{REGION}-docker.pkg.dev/{PROJECT_ID}/{REPOSITORY}/{IMAGE}",
    predictor={PREDICTOR_CLASS},
    handler={HANDLER_CLASS},
    requirements_path={PATH_TO_REQUIREMENTS_TXT},
)

Puoi esaminare le specifiche del container per ottenere informazioni utili come l'URI dell'immagine e le variabili di ambiente.

local_model.get_serving_container_spec()

Esegui il container localmente (facoltativo)

Questo passaggio è obbligatorio solo se vuoi eseguire e testare il container localmente, il che è utile per un'iterazione più rapida. Nell'esempio seguente, eseguiamo il deployment su un endpoint locale e inviamo una richiesta di previsione (formato per il corpo della richiesta).

with local_model.deploy_to_local_endpoint(
    artifact_uri={GCS_PATH_TO_MODEL_ARTIFACTS},
    credential_path={PATH_TO_CREDENTIALS},
) as local_endpoint:
    health_check_response = local_endpoint.run_health_check()
    predict_response = local_endpoint.predict(
        request_file={PATH_TO_INPUT_FILE},
        headers={ANY_NEEDED_HEADERS},
    )

Stampa il controllo di integrità e la risposta della previsione.

print(health_check_response, health_check_response.content)
print(predict_response, predict_response.content)

Stampa tutti i log del container.

local_endpoint.print_container_logs(show_all=True)

Carica su Vertex AI Model Registry

Il modello dovrà accedere agli artefatti del modello (i file dell'addestramento), quindi assicurati di averli caricati in Google Cloud Storage.

Esegui il push dell'immagine ad Artifact Registry.

local_model.push_image()

Quindi, caricalo su Model Registry.

from google.cloud import aiplatform

model = aiplatform.Model.upload(
    local_model=local_model,
    display_name={MODEL_DISPLAY_NAME},
    artifact_uri={GCS_PATH_TO_MODEL_ARTIFACTS},
)

Una volta caricato il modello in Vertex AI Model Registry, puoi utilizzarlo per ottenere previsioni batch o eseguirne il deployment su un endpoint Vertex AI per ottenere previsioni online.

Esegui il deployment nell'endpoint Vertex AI

endpoint = model.deploy(machine_type="n1-standard-4")

Una volta eseguito il deployment, puoi ricevere previsioni online.

Esempi di blocco note

Gli esempi mostrano i diversi modi in cui puoi eseguire il deployment di un modello con pre/post-elaborazione personalizzata su Vertex AI Prediction.