Implementar un modelo de aprendizaje automático personalizado como punto final de SageMaker

Implementación de terminales de SageMaker

Una guía rápida y sencilla para crear un punto final de AWS SageMaker para su modelo

El desarrollo de un modelo de aprendizaje automático (ML) implica pasos clave, desde la recopilación de datos hasta la implementación del modelo. Después de perfeccionar los algoritmos y garantizar el rendimiento mediante pruebas, el último paso crucial es la implementación. Esta fase transforma la innovación en utilidad, permitiendo que otros se beneficien de las capacidades predictivas del modelo. El modelo de ML implementado cierra la brecha entre el desarrollo y el impacto en el mundo real, brindando beneficios tangibles a los usuarios y partes interesadas.

Esta guía cubre los pasos básicos necesarios para desarrollar un ML personalizado como punto final de SageMaker. En este punto, asumo que ya tiene un modelo funcional y desea exponerlo al resto del mundo a través de un punto final. La guía le ayudará a implementar un modelo basado en PyTorch que tiene como objetivo predecir anomalías en videoclips. El modelo, también conocido como VAD AIse basa en el documento “Representaciones basadas en atributos para una detección de anomalías de vídeo precisa e interpretable”y su implementación se puede encontrar en el anomalía Repositorio de GitHub por AbiertoVINO. Para leer más sobre este interesante enfoque, desplácese hasta el final de este blog en la sección Apéndice.

En este punto, quiero enfatizar que en este caso, no podemos usar el Modelo PyTorch abstracción construida específicamente para implementar modelos PyTorch por dos razones. La primera razón es que tenemos la anomalía paquete como una dependencia adicional que no está incluida en la imagen prediseñada de PyTorch Sagemaker. La segunda razón es que el modelo requiere información adicional aprendida durante el paso de entrenamiento que no forma parte de los pesos del modelo PyTorch.

A continuación se detallan los pasos para lograr este objetivo:

1. Escribir el script de entrega del modelo Sagemaker
2. Sube el modelo a S3
3. Cargue una imagen de Docker personalizada en AWS ECR
4. Crear un modelo en SageMaker
5. Crear una configuración de punto final
6. Crear un punto final
7. Invocar el punto final

Escribir el script de entrega del modelo Sagemaker

El script de entrega del modelo de Sagemaker (inference.py) es un componente importante al crear un modelo de Sagemaker. Sirve de puente entre los modelos de aprendizaje automático y los datos del mundo real. Básicamente, procesa las solicitudes entrantes, ejecuta las predicciones del modelo y devuelve los resultados. Influyendo así en el proceso de toma de decisiones de una solicitud.

El script inference.py se compone de varios métodos clave, cada uno de los cuales tiene un propósito único y, en conjunto, facilita el proceso de entrega del modelo. A continuación enumeré los cuatro principales.

1. El método model_fn tiene la tarea de cargar el modelo entrenado. Lee los artefactos del modelo que se han guardado y devuelve un objeto de modelo que puede usarse para predicciones. Este método se llama solo una vez cuando se inicia el servidor del modelo SageMaker.
2. El método input_fn toma los datos solicitados y los formatea en un formato adecuado para hacer predicciones. Por ejemplo, en el código siguiente, esta función formatea los datos de forma diferente según la fuente de los datos (bytes de imagen o lista de URI de S3) y si la lista de fotogramas debe considerarse como un clip de vídeo.
3. El método predict_fn toma los datos de la solicitud formateados y realiza inferencias contra el modelo cargado.
4. Finalmente, se utiliza el método output_fn. Toma el resultado de la predicción y lo formatea en un mensaje de respuesta. Por ejemplo, empaquetelo como un objeto JSON.

El código para el script de entrega del modelo Sagemaker se puede encontrar a continuación.

import os
import json
import joblib
import torch
from PIL import Image
import numpy as np
import io
import boto3
from enum import Enum
from urllib.parse import urlsplit
from omegaconf import OmegaConf
from anomalib.data.utils import read_image, InputNormalizationMethod, get_transforms
from anomalib.models.ai_vad.torch_model import AiVadModel

device = "cuda"


class PredictMode(Enum):
    frame = 1
    batch = 2
    clip = 3

    
def model_fn(model_dir):
    """
    This function is the first to get executed upon a prediction request,
    it loads the model from the disk and returns the model object which will be used later for inference.
    """

    # Load the config file
    config = OmegaConf.load(os.path.join(model_dir, "ai_vad_config.yaml"))
    config_model = config.model

    # Load the model
    model = AiVadModel(
            box_score_thresh=config_model.box_score_thresh,
            persons_only=config_model.persons_only,
            min_bbox_area=config_model.min_bbox_area,
            max_bbox_overlap=config_model.max_bbox_overlap,
            enable_foreground_detections=config_model.enable_foreground_detections,
            foreground_kernel_size=config_model.foreground_kernel_size,
            foreground_binary_threshold=config_model.foreground_binary_threshold,
            n_velocity_bins=config_model.n_velocity_bins,
            use_velocity_features=config_model.use_velocity_features,
            use_pose_features=config_model.use_pose_features,
            use_deep_features=config_model.use_deep_features,
            n_components_velocity=config_model.n_components_velocity,
            n_neighbors_pose=config_model.n_neighbors_pose,
            n_neighbors_deep=config_model.n_neighbors_deep,
        )

    # Load the model weights
    model.load_state_dict(torch.load(os.path.join(model_dir, "ai_vad_weights.pth"), map_location=device), strict=False)

    # Load the memory banks
    velocity_estimator_memory_bank, pose_estimator_memory_bank, appearance_estimator_memory_bank = joblib.load(os.path.join(model_dir, "ai_vad_banks.joblib")) 
    if velocity_estimator_memory_bank is not None:
        model.density_estimator.velocity_estimator.memory_bank = velocity_estimator_memory_bank
    if pose_estimator_memory_bank is not None:
        model.density_estimator.pose_estimator.memory_bank = pose_estimator_memory_bank
    if appearance_estimator_memory_bank is not None:
        model.density_estimator.appearance_estimator.memory_bank = appearance_estimator_memory_bank
    model.density_estimator.fit()

    # Move the entire model to device
    model = model.to(device)

    # get the transforms
    transform_config = config.dataset.transform_config.eval if "transform_config" in config.dataset.keys() else None
    image_size = (config.dataset.image_size[0], config.dataset.image_size[1])
    center_crop = config.dataset.get("center_crop")
    center_crop = tuple(center_crop) if center_crop is not None else None
    normalization = InputNormalizationMethod(config.dataset.normalization)
    transform = get_transforms(config=transform_config, image_size=image_size, center_crop=center_crop, normalization=normalization)

    return model, transform


def input_fn(request_body, request_content_type):
    """
    The request_body is passed in by SageMaker and the content type is passed in 
    via an HTTP header by the client (or caller).
    """

    print("input_fn-----------------------")

    if request_content_type in ("application/x-image", "image/x-image"):
        image = Image.open(io.BytesIO(request_body)).convert("RGB")
        numpy_array = np.array(image)
        print("numpy_array.shape", numpy_array.shape)
        print("input_fn-----------------------")
        return [numpy_array], PredictMode.frame

    elif request_content_type == "application/json":
        request_body_json = json.loads(request_body)

        s3_uris = request_body_json.get("images", [])

        if len(s3_uris) == 0:
            raise ValueError(f"Images is a required key and should contain at least a list of one S3 URI")

        s3 = boto3.client("s3")
        frame_paths = []
        for s3_uri in s3_uris:
            parsed_url = urlsplit(s3_uri)
            bucket_name = parsed_url.netloc
            object_key = parsed_url.path.lstrip('/')
            local_frame_path = f"/tmp/{s3_uri.replace('/', '_')}"
            # Download the frame from S3
            s3.download_file(bucket_name, object_key, local_frame_path)
            frame_paths.append(local_frame_path)

        frames = np.stack([torch.Tensor(read_image(frame_path)) for frame_path in frame_paths], axis=0)
        
        predict_mode = PredictMode.clip if request_body_json.get("clip", False) else PredictMode.batch
        
        print("frames.shape", frames.shape)
        print("predict_mode", predict_mode)
        print("input_fn-----------------------")

        return frames, predict_mode

    # If the request_content_type is not as expected, raise an exception
    raise ValueError(f"Content type {request_content_type} is not supported")


def predict_fn(input_data, model):
    """
    This function takes in the input data and the model returned by the model_fn
    It gets executed after the model_fn and its output is returned as the API response.
    """

    print("predict_fn-----------------------")

    model, transform = model
    
    frames, predict_mode = input_data

    processed_data = {}
    processed_data["image"] = [transform(image=frame)["image"] for frame in frames]
    processed_data["image"] = torch.stack(processed_data["image"])

    image = processed_data["image"].to(device)
    
    # Add one more dimension for a batch size of one clip
    if predict_mode == PredictMode.clip:
        image = image.unsqueeze(0)

    print("image.shape", image.shape)

    model.eval()

    with torch.no_grad():
        boxes, anomaly_scores, image_scores = model(image)

    print("boxes_len", [len(b) for b in boxes])

    processed_data["pred_boxes"] = [box.int() for box in boxes]
    processed_data["box_scores"] = [score.to(device) for score in anomaly_scores]
    processed_data["pred_scores"] = torch.Tensor(image_scores).to(device)

    print("predict_fn-----------------------")

    return processed_data


def output_fn(prediction, accept):
    """
    Post-processing function for model predictions. It gets executed after the predict_fn.
    """

    print("output_fn-----------------------")

    # Check if accept type is JSON
    if accept != "application/json":
        raise ValueError(f"Accept type {accept} is not supported")

    # Convert PyTorch Tensors to lists so they can be JSON serializable
    for key in prediction:
        # If torch.Tensor convert it to list
        if isinstance(prediction[key], torch.Tensor):
            prediction[key] = prediction[key].tolist()
        # If list, convert every tensor in the list
        elif isinstance(prediction[key], list):
            prediction[key] = [tensor.tolist() if isinstance(tensor, torch.Tensor) else tensor for tensor in prediction[key]]

    print("output_fn-----------------------")

    return json.dumps(prediction), accept

PD: Se recomienda encarecidamente probar el script de servicio del modelo antes de pasar al siguiente paso. Esto se puede hacer fácilmente simulando la canalización de invocación como se muestra en el código siguiente.

import json
from inference import model_fn, predict_fn, input_fn, output_fn

response, accept = output_fn(
    predict_fn(
        input_fn(payload, "application/x-image"),
        model_fn("../")
    ),
    "application/json"
)
json.loads(response).keys()

Sube el modelo a S3

Para crear un punto final de SageMaker que cargue el modelo AI VAD PyTorch exactamente en el mismo estado, necesitamos los siguientes archivos:

• Pesos del modelo AI VAD PyTorch (también conocido como state_dict)
• Bancos de memoria del estimador de densidad (que no forman parte de los pesos del modelo)
• Un archivo de configuración con los hiperparámetros del modelo PyTorch.
• Un modelo de Sagemaker que sirve script (inference.py)

El siguiente código demuestra cómo organizar todos los archivos necesarios en un directorio.

PD: anulé la devolución de llamada incorporada de PyTorch ModelCheckpoint para garantizar que esos bancos de memoria se guarden como parte del guardado del punto de control (la implementación se puede encontrar aquí).

import torch
import joblib
import shutil

checkpoint = "results/ai_vad/ucsd/run/weights/lightning/model.ckpt"
config_path = "results/ai_vad/ucsd/run/config.yaml"

model_weights = torch.load(checkpoint)
model_state_dict = model_weights["state_dict"]

torch.save(model_state_dict, "../ai_vad_weights.pth")

velocity_estimator_memory_bank = None
pose_estimator_memory_bank = None
appearance_estimator_memory_bank = None
if "velocity_estimator_memory_bank" in model_weights:
    velocity_estimator_memory_bank = model_weights["velocity_estimator_memory_bank"]
if "pose_estimator_memory_bank" in model_weights:
    pose_estimator_memory_bank = model_weights["pose_estimator_memory_bank"]
if "appearance_estimator_memory_bank" in model_weights:
    appearance_estimator_memory_bank = model_weights["appearance_estimator_memory_bank"]
banks = (velocity_estimator_memory_bank, pose_estimator_memory_bank, appearance_estimator_memory_bank)

joblib.dump(banks, "../ai_vad_banks.joblib")

shutil.copyfile(config_path, "../ai_vad_config.yaml")

Luego, los cuatro archivos se comprimieron para crear tar.gz usando el siguiente comando.

tar -czvf ../ai_vad_model.tar.gz -C ../ ai_vad_weights.pth ai_vad_banks.joblib ai_vad_config.yaml inference.py

Por último, el archivo se cargó en S3 usando boto3.

import boto3
from datetime import datetime

current_datetime = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')

s3 = boto3.resource('s3')
s3.meta.client.upload_file("../ai_vad_model.tar.gz", "ai-vad", f"{current_datetime}/ai_vad_model.tar.gz")

Cargue una imagen de Docker personalizada en AWS ECR

Como se mencionó anteriormente, dado que tenemos una dependencia adicional que no está incluida en la imagen prediseñada de PyTorch Sagemaker (es decir, anomalía paquete), creamos una nueva imagen de Docker para ese propósito. Antes de crear la imagen de Docker personalizada, se requiere autenticación en el repositorio de Amazon ECR.

REGION=<my_aws_region>
ACCOUNT=<my_aws_account>
 
# Authenticate Docker to an Amazon ECR registry
aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin <docker_registry_url>.dkr.ecr.$REGION.amazonaws.com

# Loging to your private Amazon ECR registry
aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin $ACCOUNT.dkr.ecr.$REGION.amazonaws.com

El Dockerfile se puede encontrar a continuación y se pueden encontrar las diferentes rutas de registro de Docker. aquí. Asegúrese de seleccionar la ruta de registro correcta según las necesidades del modelo (CPU/GPU, versión de Python, etc.) y su región de AWS. Por ejemplo, si la región es us-east-1, la ruta completa del registro de Docker debería ser similar a esta:
763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-inference:2.0.0-gpu-py310

# Use the SageMaker PyTorch image as the base image
FROM <docker_registry_url>.dkr.ecr.<my_aws_region>.amazonaws.com/pytorch-inference:2.0.0-gpu-py310

# Install the additional dependency
RUN pip install "git+https://github.com/hairozen/anomalib.git@ai-vad-inference-improvements"

Ahora, podemos ejecutar el comando de compilación clásico de Docker para crear esta imagen personalizada.

docker build -t ai-vad-image .

El siguiente paso es crear el repositorio de AWS ECR para la nueva imagen que creamos, etiquetarla y enviar la imagen al repositorio de AWS ECR.

# Create the AWS ECR repository
aws ecr create-repository --repository-name ai-vad-image

# Tag the image
docker tag ai-vad-image:latest $ACCOUNT.dkr.ecr.$REGION.amazonaws.com/ai-vad-image:latest

# Push the tagged image to the AWS ECR repository
docker push $ACCOUNT.dkr.ecr.$REGION.amazonaws.com/ai-vad-image:latest

Crear un modelo en SageMaker

Este paso es bastante sencillo. Código a continuación.

import boto3
import sagemaker

sagemaker_client = boto3.client(service_name="sagemaker")
role = sagemaker.get_execution_role()

model_name = f"ai-vad-model-{current_datetime}"

primary_container = {
    "Image": f"{my_aws_account}.dkr.ecr.{my_aws_region}.amazonaws.com/ai-vad-image:latest",
    "ModelDataUrl": f"s3://ai-vad/{current_datetime}/ai_vad_model.tar.gz"
}

create_model_response = sagemaker_client.create_model(
    ModelName=model_name,
    ExecutionRoleArn=role,
    PrimaryContainer=primary_container)

Crear una configuración de punto final

El siguiente paso es crear una configuración de punto final. A continuación puedes encontrar uno básico.

endpoint_config_name = f"ai-vad-model-config-{current_datetime}"

sagemaker_client.create_endpoint_config(
    EndpointConfigName=endpoint_config_name,
    ProductionVariants=[{
        "InstanceType": "ml.g5.xlarge",
        "InitialVariantWeight": 1,
        "InitialInstanceCount": 1,
        "ModelName": model_name,
        "VariantName": "AllTraffic"}])

Crear un punto final

Ahora estamos listos para crear el punto final.

endpoint_name = f"ai-vad-model-endpoint-{current_datetime}"

sagemaker_client.create_endpoint(
    EndpointName=endpoint_name,
    EndpointConfigName=endpoint_config_name)

Tenga en cuenta que pueden pasar algunos minutos hasta que el estado del punto final cambie de “Creando” a “En servicio”. El estado actual se puede comprobar como se muestra a continuación.

response = sagemaker_client.describe_endpoint(EndpointName=endpoint_name)
response["EndpointStatus"]

Invocar el punto final

Ha llegado la hora del dinero. Ahora es el momento de invocar el punto final para probar que todo funciona como se esperaba.

with open(file_name, "rb") as f:
    payload = f.read()

predictor = sagemaker.predictor.Predictor(endpoint_name=endpoint_name)
predictor.serializer = DataSerializer(content_type="image/x-image")
predictor.predict(payload)

Entonces, esta es una buena verificación, pero debes tener en cuenta que la función predictor.predict no ejecuta la canalización de invocación completa desde el script de publicación de SageMaker que incluye:
salida_fn(predict_fn(input_fn(input_data, model_fn(model_dir)),aceptar)

Para probarlo también, invoquemos el modelo mediante una llamada API.

with open(file_name, "rb") as f:
    payload = f.read()

sagemaker_runtime = boto3.client("runtime.sagemaker")
response = sagemaker_runtime.invoke_endpoint(
    EndpointName=endpoint_name,
    ContentType="image/x-image",
    Body=payload
)

response = json.loads(response["Body"].read().decode())

Usando la gran visualización anomalía proporciona, podemos dibujar los cuadros y sus etiquetas para un cuadro determinado del conjunto de datos UCSDped2.

Conclusión

Bien, resumamos rápidamente lo que cubrimos aquí. La implementación de un modelo de SageMaker para servicio requiere una serie de pasos.

Primero, se debe escribir el script de entrega del modelo de Sagemaker para definir la funcionalidad y el comportamiento del modelo.

Luego, el modelo se carga en Amazon S3 para su almacenamiento y recuperación.
Además, se carga una imagen de Docker personalizada en AWS Elastic Container Registry (ECR) para contener el modelo y sus dependencias.
El siguiente paso implica crear un modelo en SageMaker, que asocia los artefactos del modelo almacenados en S3 con la imagen de Docker almacenada en ECR.

Luego se crea una configuración de punto final, que define la cantidad y el tipo de instancias que se utilizarán para alojar el modelo.

Finalmente, se crea un punto final para establecer una conexión en vivo entre el modelo implementado y las aplicaciones cliente, permitiéndoles invocar el punto final y hacer predicciones en tiempo real.

A través de estos pasos, la implementación de un modelo de SageMaker se convierte en un proceso simplificado que garantiza un servicio de modelo eficiente y confiable.

Apéndice

El Representaciones basadas en atributos para una detección de anomalías de vídeo precisa e interpretable artículo publicado en 2023 por Reiss et al. que propone un método simple pero altamente efectivo para la detección de anomalías de video (VAD) utilizando representaciones basadas en atributos.

El artículo sostiene que los métodos VAD tradicionales, que a menudo se basan en el aprendizaje profundo, suelen ser difíciles de interpretar, lo que dificulta que los usuarios comprendan por qué el sistema marca ciertos fotogramas u objetos como anómalos.

Para abordar este problema, los autores proponen un método que representa cada objeto en un vídeo por su velocidad, pose y profundidad. Estos atributos son fáciles de entender e interpretar y se pueden utilizar para calcular puntuaciones de anomalías mediante un enfoque basado en la densidad.

El documento muestra que esta representación simple es suficiente para lograr un rendimiento de vanguardia en varios conjuntos de datos VAD desafiantes, incluido ShanghaiTech, el conjunto de datos VAD más grande y complejo.

Además de ser preciso, los autores también demuestran que su método es interpretable. Por ejemplo, pueden proporcionar a los usuarios una lista de los objetos de un vídeo que más contribuyen a su puntuación de anomalía, junto con su velocidad, pose e información profunda. Esto puede ayudar a los usuarios a comprender por qué el sistema marca el vídeo como anómalo.

En general, este artículo es una contribución significativa al campo de VAD. Propone un método simple, preciso e interpretable para VAD que puede usarse en una variedad de aplicaciones.

Implementar un modelo de aprendizaje automático personalizado como punto final de SageMaker fue publicado originalmente en Hacia la ciencia de datos en Medium, donde las personas continúan la conversación resaltando y respondiendo a esta historia.