Despliegue de un modelo en la nube con Streamlit Cloud

En este tutorial veremos cómo desplegar en la nube un modelo de Machine Learning pre-entrenado usando Streamlit Cloud.

Al final de este tutorial se encuentra el enlace para descargar el código fuente.

Introducción

En un artículo anterior hablamos de las diferentes formas de realizar el despliegue de un modelo de Machine Learning, así que ahora pondremos en práctica algunos de esos conceptos.

Entonces, en este tutorial veremos cómo tomar un modelo previamente entrenado y cómo desplegarlo en la nube de manera sencilla usando Streamlit Cloud.

Para este despliegue usaremos el autoencoder entrenado anteriormente para la detección de anomalías cardiacas. Además, crearemos un sencillo aplicativo web con Streamlit y llevaremos nuestro código a GitHub para finalmente realizar el despliegue desde Streamlit Cloud.

El modelo a desplegar

Usaremos el autoencoder entrenado previamente para la detección de anomalías cardiacas.

En ese tutorial se encuentran todos los detalles del modelo implementado, aunque acá resumiremos algunas de sus principales características:

• El objetivo de este autoencoder es clasificar el electrocardiograma (ECG) de un sujeto como “Normal” o “Anormal”
• Para lo anterior contamos con un set de entrenamiento con 7.600 registros y uno de prueba con 1.900. Ambos sets contienen tanto sujetos normales como anormales
• Cada registro corresponde a un ciclo cardiaco y ha sido interpolado para tener exactamente un total de 140 muestras

Recordemos que el autoencoder es una arquitectura de Deep Learning que está entrenada para reconstruir el dato de entrada y en este caso particular nuestro modelo ha sido entrenado únicamente con registros “Normales”.

Así que si usamos el modelo entrenado para reconstruir un dato “Normal” el error en la reconstrucción será menor que aquel que obtendríamos al introducir un dato “Anormal”.

Siguiendo la lógica anterior podremos realizar la clasificación de forma sencilla. Simplemente tomamos un registro y generamos una predicción con el autoencoder entrenado. Luego calculamos la diferencia (medida a través del error medio absoluto – o MAE por sus siglas en Inglés -) entre la reconstrucción y el dato original y si este error es inferior a un cierto umbral lo clasificaremos como “Normal”. De lo contrario le asignaremos la categoría “Anormal”.

Este modelo será entonces el punto de partida para realizar el despliegue en la nube.

Estructura del proyecto

Cuando realizamos el despliegue de un modelo de Machine Learning lo más recomendable es organizar nuestro código de forma tal que la etapa de desarrollo (descrita en la sección anterior) esté separada de la etapa de producción (la que usaremos para desplegar el modelo).

Lo anterior debido a que el código no será el mismo. En la etapa de desarrollo debemos leer nuestro set de datos, crear el modelo, entrenarlo y ponerlo a prueba. Sin embargo, en la etapa de producción tan sólo debemos tomar el modelo pre-entrenado y el código necesario para realizar las predicciones, así como un aplicativo web que nos permita fácilmente cargar un dato y realizar una predicción.

Por este motivo en este tutorial organizaremos nuestro código en tres directorios:

• «dev» que contendrá todo el código de desarrollo: el Notebook de Jupyter con la implementación así como los sets de entrenamiento y prueba
• «prod» que contendrá el archivo con el modelo pre-entrenado, el código para generar las predicciones, el código para el aplicativo web y otros archivos adicionales requeridos durante el despliegue (y de los que hablaremos más adelante en este tutorial)
• «data» en donde almacenaremos los registros individuales de los ECGs que hacen parte del set de prueba y que usaremos más adelante para probar el funcionamiento del modelo desplegado

Modificaciones a la etapa de desarrollo

Antes de iniciar el despliegue debemos añadir algunas líneas de código al programa implementado en la etapa de desarrollo.

En particular debemos añadir una porción de código encargada de exportar cada dato de prueba al directorio «data», así como almacenar el modelo pre-entrenado en el directorio «prod».

Para almacenar cada dato como un archivo individual usaremos el formato «pickle» que se encuentra en la Librería Estándar de Python. Para esto necesitamos estas líneas de código:

from pickle import dump
x_test_s = [x_test_1_s, x_test_2_s, x_test_3_s, x_test_4_s, x_test_5_s]

for i, subset in enumerate(x_test_s):
    for j, registro in enumerate(subset):
        dump(registro, open(f'../data/x_test_{i+1}_{j+1}.pkl', 'wb'))

en donde simplemente hemos creado una lista («x_test_s») con los 5 subsets de prueba («x_test_1_s» a «x_test_5_s») y luego con una sentencia for hemos tomado cada uno de los datos y lo hemos almacenado en el directorio «data» usando el método «dump» del módulo «pickle».

Ahora realizaremos la segunda modificación, que nos permitirá almacenar el modelo entrenado en el directorio «prod». Para esto necesitamos sólo una línea de código:

torch.save(autoencoder, '../prod/autoencoder.pth')

donde hemos usado el método «save» de PyTorch y «autoencoder.pth» es el nombre del archivo que contendrá nuestro modelo entrenado.

Etapa de producción: el backend

Muy bien, con el modelo ya entrenado y almacenado estamos listos para iniciar la etapa de producción.

Esta etapa de producción tendrá dos elementos:

• El backend que será simplemente el código en Python encargado de tomar un dato y generar una predicción con el modelo pre-entrenado
• El frontend que será la interfaz que permitirá la interacción del usuario y que en este caso será simplemente una página web

En esta sección nos enfocaremos en el backend y en la próxima hablaremos del frontend.

Este backend tendrá un elemento que permitirá crear el modelo y cargar los coeficientes de entrenamiento, otro elemento encargado de leer los datos y otro a cargo de las predicciones. Todos estos elementos serán almacenados en el archivo «utils.py» del directorio «prod».

Veamos entonces cada uno de estos componentes.

Lectura de un dato

La entrada al modelo será uno de los datos (en formato pickle) almacenados previamente en el directorio «data».

Para ello simplemente crearemos la función «leer_dato»:

import pickle

def leer_dato(uploaded_file):
    dato = pickle.loads(uploaded_file.getvalue())

    return dato

En este caso el argumento de entrada a la función («uploaded_file») será un objeto de Streamlit que apuntará al archivo que queramos procesar. Para acceder al contenido de este archivo usaremos el método «getvalue()» y posteriormente podremos usar el método «loads» para realizar la lectura y almacenar el resultado en la variable dato.

Creación del modelo y carga de los coeficientes del modelo entrenado

El primer paso es crear una instancia de la clase «Autoencoder» que definimos en la etapa de desarrollo y que se encarga de definir el autoencoder en PyTorch con la arquitectura deseada. Para ello simplemente copiamos y pegamos las líneas de código que ya habíamos creado previamente en la fase de desarrollo:

import torch

class Autoencoder(torch.nn.Module):
    def __init__(self):
        super(Autoencoder, self).__init__()

        #Codificador
        self.cod1 = torch.nn.Linear(in_features=140, out_features=32) # Entrada: 140 muestras
        self.cod2 = torch.nn.Linear(in_features=32, out_features=16)
        self.cod3 = torch.nn.Linear(in_features=16, out_features=8)
        
        #Decodificador
        self.dec1 = torch.nn.Linear(in_features=8, out_features=16)
        self.dec2 = torch.nn.Linear(in_features=16, out_features=32)
        self.dec3 = torch.nn.Linear(in_features=32, out_features=140)
        
    def forward(self, x):
        x = torch.nn.functional.relu(self.cod1(x))
        x = torch.nn.functional.relu(self.cod2(x))
        x = torch.nn.functional.relu(self.cod3(x))
        
        x = torch.nn.functional.relu(self.dec1(x))
        x = torch.nn.functional.relu(self.dec2(x))
        x = torch.sigmoid(self.dec3(x))
        
        return x

Esta clase simplemente nos permitirá crear la “cubierta” del modelo. El “relleno” de este modelo serán precisamente los coeficientes obtenidos durante el entrenamiento y que se encuentran almacenados en el directorio «prod» con el nombre «autoencoder.pth». Para cargar estos coeficientes en el modelo instanciado anteriormente usaremos la siguiente función:

def cargar_modelo_preentrenado(model_path):
    modelo = torch.load(model_path)
    modelo.eval()

    return modelo

En este caso el parámetro de entrada a la función será el nombre del archivo con el cual almacenamos previamente el modelo. Con esta información usaremos «modelo» (que será una instancia de la clase «Autoencoder») y los métodos «load» y «eval» de PyTorch para cargar los coeficientes.

Al finalizar la función retornará la variable «modelo» que contendrá el modelo pre-entrenado y listo para realizar las predicciones.

Clasificación con el modelo pre-entrenado

Teniendo el dato de entrada y el modelo pre-entrenado ya podremos realizar la clasificación, que consta de dos fases: la predicción y la definición de la categoría a la que pertenece el dato (“Normal” o “Anormal”).

La función «predecir» permitirá tomar el «modelo», el «dato» y el «umbral» (definido en la etapa de desarrollo) y a la salida generará un dato de tipo booleano indicando si la reconstrucción entregada por el modelo se encuentra por debajo («True») o por encima («False») del umbral establecido:

def predecir(modelo, dato, umbral):
    fn_perdida = torch.nn.L1Loss(reduction='none')
    reconstrucciones = modelo(torch.from_numpy(datos).float())
    perdida = fn_perdida(reconstrucciones, torch.from_numpy(datos).float()).mean()

    return torch.lt(perdida, umbral)

Entendamos cada línea de código de esta función:

• «fn_perdida = torch.nn.L1Loss(reduction=’none’)»: acá simplemente definiremos una instancia (variable «fn_perdida») del error medio absoluto («torch.nn.L1Loss»), que es la métrica que usaremos para comparar el dato original con la reconstrucción entregada por el modelo.
• En la segunda línea obtendremos como tal la reconstrucción («reconstrucciones») del dato de entrada al modelo. Para ello introducimos el dato al modelo («modelo(torch.from_numpy(datos).float())»), teniendo en cuenta que debe estar en formato de tensor de PyTorch y almacenado en punto flotante (por lo cual usamos los métodos «from_numpy()» y «float()»)
• Teniendo el dato original («datos») y la reconstrucción («reconstrucciones»), en la tercera línea calculamos como tal el error («perdida») a través precisamente de la variable «fn_perdida» creada inicialmente.

En la última parte de la función lo que haremos será comparar este error con el umbral establecido previamente. Esta comparación la realizamos con la función lt de PyTorch que:

• Arroja un valor verdadero («True») si «perdida<umbral»
• O un valor falso («False») en caso contrario

Con el resultado de esta comparación ya podemos definir la categoría a la que pertenece el dato, a través de la función «obtener_categoria»:

def obtener_categoria(comparaciones):
    if comparaciones.item():
        categoria = 'Normal'
    else:
        categoria = 'Anormal'

    return categoria

En este caso tomamos simplemente el resultado de la predicción («comparaciones») y extraemos su contenido (con el método «item()»). Si dicho contenido es de tipo «True» («if comparaciones.item()») clasificaremos el dato en la «categoria = ‘Normal'», de lo contrario lo clasificaremos como «Anormal».

Muy bien, y con esto ya tenemos listo el backend. Recordemos que todas estas funciones estarán almacenadas en el archivo «utils.py».

Ya estamos listos para implementar el frontend.

Etapa de producción: el frontend

Como lo mencioné anteriormente, el frontend será la interfaz que conectará el núcleo de la aplicación (el backend) con el usuario final y que en este caso será simplemente una página web que permitirá cargar un dato y luego mostrar el resultado de la clasificación en la misma página.

Para implementar este frontend usaremos Streamlit, una librería de Python que, como lo veremos en un momento, permite desarrollar aplicativos web de manera muy intuitiva y con muy pocas líneas de código.

Antes de comenzar con la implementación vale la pena aclarar que en esta etapa toda la implementación será local (es decir en nuestro computador). En la última parte del tutorial veremos cómo llevar este código a la nube para el despliegue final del modelo.

Implementación en Streamlit

Comencemos entonces esta implementación creando el archivo «app.py» en nuestro directorio «prod». Lo primero que haremos será importar el módulo «streamlit» junto con la totalidad del código que creamos en la sección anterior para el backend:

import streamlit as st
from utils import *

Además, definiremos inicialmente el valor del umbral a usar durante la clasificación (y el cual obtuvimos en la etapa de desarrollo):

UMBRAL = 0.089

Y ahora sí comencemos a crear nuestra página web con Streamlit. En primer lugar definamos algunas características básicas:

st.set_page_config(page_icon="📊", page_title="Detección de anomalías cardiacas", layout="wide")
st.image("https://www.codificandobits.com/img/cb-logo.png", width=200)
st.title("Detección de anomalías cardiacas con autoencoders")
c29, c30, c31 = st.columns([1, 6, 1]) # 3 columnas: 10%, 60%, 10%

donde hemos usado:

• «set_page_config» para definir el ícono («page_icon») y el título («page_title») que tendrá la página en la pestaña de nuestro navegador
• «image» para agregar un logo
• «title» para definir el título que verá el usuario una vez abra la página

Además, con «st.columns([1,6,1])» le hemos indicado a Streamlit que nuestra página estará dividida en tres columnas con un ancho equivalente al 10%, 60% y 10% del ancho total de la pantalla.

Ahora sí definimos como tal el contenido de la página, que contendrá un cuadro de diálogo para abrir el archivo y que realizará la clasificación siempre y cuando hayamos cargado un archivo con formato válido. Todo esto lo podemos implementar con estas líneas de código:

with c30:
    # Bloque 1
    uploaded_file = st.file_uploader(
        "", type = 'pkl',
        key="1",
    )

    # Bloque 2
    if uploaded_file is not None:
        # Bloque 2a
        info_box_wait = st.info(
            f"""
                Realizando la clasificación...
                """)

        # Bloque 2b: Acá viene la predicción con el modelo
        dato = leer_dato(uploaded_file)
        autoencoder = Autoencoder()
        autoencoder = cargar_modelo_preentrenado('autoencoder.pth')
        prediccion = predecir(autoencoder, dato, UMBRAL)
        categoria = obtener_categoria(prediccion)

        # Bloque 2c: Y mostrar el resultado
        info_box_result = st.info(f"""
            El dato analizado corresponde a un sujeto: {categoria}
            """)

    else:
        # Bloque 2d
        st.info(
            f"""
                👆 Debe cargar primero un dato con extensión .pkl
                """
        )
        st.stop()

Veamos qué hace cada uno de los bloques de procesamiento en este código:

• Bloque 1: permite realizar la carga de un archivo en formato Pickle («type = pkl») usando el método «file_uploader». El resultado es almacenado en «uploaded_file»
• Bloque 2: el bloque tiene un condicional («if…else»). Si realmente el usuario carga un archivo («if uploaded_file is not None:») se ejecutarán los bloques 2a, 2b y 2c; de lo contrario se ejecutará el bloque 2d.
• Bloque 2a: muestra un cuadro de diálogo en la página web indicando al usuario que se está realizando la clasificación del dato
• Bloque 2b: en este bloque se da como tal la conexión con el backend. Acá simplemente llamaremos en el orden requerido las funciones almacenadas en «utils.py» explicadas anteriormente. Al culminar la ejecución de este bloque tendremos la categoría del dato almacenada en un string
• Bloque 2c: al finalizar la clasificación mostraremos en pantalla un cuadro de diálogo con la categoría predicha por el modelo («Normal» o «Anormal»)
• Bloque 2d: sólo se ejecuta si no hemos cargado un dato con la extensión requerida. En este caso se muestra en pantalla un mensaje con las instrucción respectiva

Prueba del aplicativo en el entorno local

Perfecto. Con el backend y el frontend ya implementados, lo que nos resta en esta fase es ponerlos a prueba. Es decir abrir nuestra página web y realizar la clasificación de un dato.

Esto es muy simple, basta con abrir el Terminal (en Mac) o la ventana de comandos (en Windows) y desde el directorio «prod» escribir:

streamlit run app.py

¡y automáticamente se abrirá nuestro navegador de Internet y nos mostrará el aplicativo que acabamos de implementar!

Vale la pena aclarar nuevamente que lo que estamos viendo es un despliegue local, es decir que la página web que acabamos de abrir se está ejecutando desde nuestro computador, no desde un servidor en Internet. Para verificar esto podemos ver en la barra de direcciones algo como «http://localhost:8501/»

Simplemente damos click en el botón Browse files, cargamos el dato que deseemos clasificar y veremos en pantalla uno de estos dos mensajes: El dato analizado corresponde a un sujeto: Normal ó El dato analizado corresponde a un sujeto: Anormal.

¡Y listo, ya tenemos desplegado localmente nuestro modelo!

El siguiente paso es realizar el despliegue en la nube, para lo cual haremos uso de GitHub junto con el servicio Streamlit Cloud. Veamos en detalle cómo hacerlo.

Despliegue en la nube

Este despliegue es tal vez el paso más sencillo de todos. Simplemente lo que haremos será tomar el contenido de nuestro directorio de producción («prod») y crear un repositorio en GitHub. Después iremos a Streamlit Cloud y allí realizaremos la conexión con este repositorio.

Veamos en detalle todos los pasos para realizar este despliegue.

Archivo «requirements.txt»

Antes de crear nuestro repositorio debemos configurar el archivo «requirements.txt», en donde especificaremos las librerías que usamos en nuestro backend y frontend. Este archivo le permitirá posteriormente a Streamlit Cloud saber cuáles librerías debe instalar en el servidor en la nube donde haremos el despliegue.

En nuestro proyecto las únicas librerías que necesitamos son Streamlit y Pytorch, así que el contenido del archivo será este:

streamlit
torch==1.12.1

en donde hemos especificado la versión 1.12.1 de Pytorch que es la que he usado para el despliegue local. Este archivo «requirements.txt» lo almacenamos también en el directorio de producción.

Repositorio en GitHub

Una vez creado este archivo local podemos crear el repositorio, para lo cual lo inicializamos localmente usando «git init» en el Terminal (de Mac) o en la ventana de comandos (de Windows).

A continuación creamos el repositorio remoto en GitHub, que en este caso tendrá el nombre «deteccion-anomalias-cardiacas-despliegue-streamlit».

Después enlazamos el repositorio local con el remoto escribiendo (también en el Terminal o en la ventana de comandos):

• «git remote add origin deteccion-anomalias-cardiacas-despliege-streamlit.git»
• «git branch -M main»
• «git push -u origin main»

Y listo, ya tenemos nuestro código para el despliegue en el repositorio de GitHub.

Nos queda un paso más y tendremos nuestro modelo accesible desde la web.

Despliegue con Streamlit Cloud

En primer lugar nos dirigimos al sitio web de Streamlit Cloud, creamos nuestra cuenta de usuario (Sign up) o si ya tenemos una simplemente accedemos con nuestro nombre de usuario y contraseña (Sign in).

Al ingresar veremos un panel de usuario similar al de la figura:

En este panel están listados todos los proyectos que hemos desplegado. Pero antes de este despliegue debemos configurar el acceso a GitHub. Para ello:

• Vamos a la pestaña Settings (configuración), ubicada en la esquina superior derecha del panel
• En el ítem Source control oprimimos el botón Sign in with GitHub.
• Especificamos los datos de acceso a nuestra cuenta en GitHub y listo, ya hemos enlazado Streamlit Cloud con nuestra cuenta en GitHub

Ahora volvemos al panel del usuario y oprimimos el botón New app (nueva aplicación). Veremos entonces un panel de configuración de nuestra nueva aplicación similar a este:

En este panel debemos simplemente completar los tres campos que allí aparecen:

• En Repository seleccionamos del menú desplegable el repositorio de GitHub que contiene el código fuente de nuestro proyecto
• En Branch dejamos el valor main
• Y en Main file path escribimos el nombre del archivo de nuestro código fuente que contiene el aplicativo web: app.py

Y ahora sí estamos literalmente a un click del despliegue. Simplemente oprimimos el botón Deploy (desplegar), esperamos un par de minutos (mientras nuestra aplicación se despliega en los servidores web de Streamlit Cloud)… ¡y listo, ya tenemos nuestro modelo de Machine Learning desplegado en la web!

Vemos que automáticamente Streamlit Cloud genera una dirección web (una URL), que en este caso es https://codificandobits-deteccion-anomalias-cardiacas-despli-app-7h9fdq.streamlitapp.com/.

La forma de uso de este despliegue es idéntica al que hicimos localmente: simplemente debemos cargar un dato desde nuestro computador y automáticamente el dato será enviado al servidor de Streamlit Cloud donde tenemos alojado nuestro modelo, el código para realizar la predicción se ejecutará y luego veremos en la página web el resultado de la clasificación: “Normal” o “Anormal” dependiendo del dato que hayamos introducido.

Enlace de descarga del código fuente

▼ El código fuente de este tutorial

Conclusión

Muy bien, en este tutorial acabamos de ver que el despliegue de un modelo en la nube usando Streamlit Cloud es realmente sencillo: basta con organizar nuestro código previamente desarrollado, crear el aplicativo web con Streamlit, llevar esto a un repositorio en GitHub y luego enlazar este repositorio con Streamlit Cloud.

Como se menciona en el artículo anterior, cuando hablamos del despliegue en el Machine Learning, este tipo de despliegue resulta muy útil si queremos tener rápidamente un servicio en la nube y si la capacidad de procesamiento o el número de usuarios de la aplicación no son elevados, pero si necesitamos más capacidad debemos explorar otras alternativas (como por ejemplo Google Cloud, Microsoft Azure o Amazon Web Services).