Rotinas de previsão personalizadas

Com as rotinas de previsão personalizadas (CPR, na sigla em inglês), é possível criar contêineres personalizados com códigos de pré/pós processamento de maneira fácil, sem precisar configurar um servidor HTTP ou criar um contêiner do zero. É possível usar o pré-processamento para normalizar/transformar as entradas ou fazer chamadas para serviços externos a fim de coletar mais dados e usar o pós-processamento para formatar a previsão do modelo ou executar a lógica de negócios.

O diagrama a seguir descreve o fluxo de trabalho do usuário com e sem rotinas de previsão personalizadas.

As principais diferenças são as seguintes:

  • Você não precisa escrever um servidor de modelo ou um Dockerfile. O servidor do modelo, que é o servidor HTTP que hospeda o modelo, é fornecido para você.

  • É possível implantar e depurar o modelo localmente, acelerando o ciclo de iteração durante o desenvolvimento.

Criar e implantar um contêiner personalizado

Nesta seção, descrevemos como usar o CPR para criar um contêiner personalizado com lógica de pré/pós processamento e implantação em um endpoint local e on-line.

Instalação

É necessário ter o SDK da Vertex AI e o Docker instalados no seu ambiente.

Gravar Predictor personalizada

Implementar a interface 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

Por exemplo, consulte Implementação da Predictor do Sklearn.

Escrever Handler personalizado (opcional)

Os gerenciadores personalizados têm acesso ao objeto bruto de solicitação e, portanto, são úteis em casos raros em que é necessário personalizar a lógica relacionada ao servidor da Web, como suporte a cabeçalhos de solicitação/resposta adicionais ou desserialização de solicitações de previsão não formatadas em JSON.

Veja um exemplo de notebook que implementa o Predictor e o Handler.

Embora não seja obrigatório, para melhorar a organização e a reutilização de código, recomendamos que você implemente a lógica do servidor da Web no gerenciador e a lógica de ML no Predictor, conforme mostrado no gerenciador padrão.

Criar contêiner personalizado

Coloque o código personalizado e um arquivo requirements.txt extra, se precisar instalar pacotes na imagem, em um diretório.

Use o SDK Vertex para criar contêineres personalizados, conforme mostrado abaixo:

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},
)

É possível inspecionar a especificação do contêiner para receber informações úteis, como o URI da imagem e as variáveis de ambiente.

local_model.get_serving_container_spec()

Execute o contêiner localmente (opcional)

Essa etapa é necessária apenas se você quer executar e testar o contêiner localmente, o que é útil para uma iteração mais rápida. No exemplo a seguir, implantamos no endpoint local e enviamos uma solicitação de previsão (formato para o corpo da solicitação).

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},
    )

Imprimir a verificação de integridade e a resposta de previsão.

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

Imprima todos os registros do contêiner.

local_endpoint.print_container_logs(show_all=True)

Fazer upload para o registro de modelo da Vertex AI

Seu modelo precisará acessar os artefatos do modelo (os arquivos de treinamento). Portanto, verifique se você fez upload deles para o Google Cloud Storage.

Enviar a imagem para o Artifact Registry.

local_model.push_image()

Em seguida, faça upload para o 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},
)

Depois de fazer upload do modelo para o Registro de modelos da Vertex AI, ele pode ser usado para receber previsões em lote ou implantado em um endpoint da Vertex AI para receber previsões on-line.

Implantar no endpoint da Vertex AI

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

Depois de implantado, é possível receber previsões on-line.

Amostras de notebook

Os exemplos mostram as diferentes maneiras de implantar um modelo com pré/pós-processamento personalizado na Vertex AI Prediction.