MLOps en la práctica: cómo diseñar aplicaciones de IA

Los modelos de IA solo son útiles cuando se ponen en acción, como en las aplicaciones. MLOps se trata de tener la configuración y los procedimientos correctos para entrenar y usar estos modelos de manera consistente y organizada. Eso significa asegurarse de que todo sea reproducible, escalable, fácil de mantener y pueda verificarse más adelante.

Junto con los ingenieros de plataforma, los ingenieros de MLOps son los encargados de construir y cuidar los sistemas y herramientas que los científicos de datos y los ingenieros de ML utilizan para crear y lanzar sus modelos. Hacen posible que el equipo trabaje sin problemas y de manera eficiente, con todo lo que necesitan al alcance de la mano.

Según mi experiencia, diseñar un sistema que equilibre los estándares de código y la flexibilidad para los científicos de datos puede ser todo un desafío. Por un lado, es crucial asegurarse de que el código se adhiera a las mejores prácticas y sea mantenible a lo largo del tiempo. Por otro lado, los científicos de datos necesitan la libertad de experimentar con diferentes tipos de modelos y enfoques para lograr los mejores resultados. Después de todo, no desea frustrar a los científicos de datos.

Como se describe en mi otra serie de publicaciones de blog, es esencial comenzar con un buen diseño. Esta publicación de blog describe una plantilla para la parte de inferencia del modelo de las aplicaciones de IA. Comienzo simple y me extiendo en cuatro pasos.

Pero primero, definamos los componentes de una aplicación que usa IA.

Al dividir la aplicación en varios componentes en lugar de diseñarla como un monolito, podemos lograr una separación clara de las preocupaciones. Este enfoque nos permite actualizar cada componente de forma independiente, lo que es muy beneficioso en términos de flexibilidad y escalabilidad. Una aplicación de IA consta de los siguientes elementos:

• Orquestación de aplicaciones: un componente que administra todas las tareas en la aplicación;
• Otra lógica de negocio: todo lo que no sea IA, por ejemplo, una API;
• Servicio de IA: todo lo relacionado con la IA, como la inferencia del modelo.

Los nuevos datos fluyen hacia la aplicación, donde la lógica empresarial los procesa y los utiliza en el servicio de IA. Las predicciones del modelo de IA se utilizan en otras partes de la aplicación, por ejemplo, expuestas a través de una API.

Acerquémonos al servicio de IA. ¿Lo que está sucediendo allí? El modelo de IA se carga durante la inicialización. Durante el tiempo de ejecución, todos los pasos del modelo se ejecutan en orden y son administrados por orquestación.

Los componentes de la aplicación se pueden construir de varias maneras. Algunas opciones son:

• Un enfoque orientado a objetos es donde se inicia una instancia de la clase de servicio y donde sus métodos se utilizan según sea necesario. Orquestación a través de una función o método.
• Un microservicio, donde la orquestación se maneja dentro del servicio de IA o a través de un servicio de orquestación (comunicación de servicio a servicio).
• Una canalización de aprendizaje automático, donde los pasos del modelo se ejecutan en trabajos. Orquestación a través de la canalización.

Si bien las herramientas pueden diferir, la configuración de alto nivel sigue siendo la misma en todas las opciones.

Usemos la primera opción como ejemplo para profundizar en los diversos componentes del servicio de IA. En cuatro pasos, agregamos funcionalidad adicional al tiempo que garantizamos la reproducibilidad y la calidad:

1. Los basicos
2. Adición de contratos de servicio para la validación de datos
3. Almacenar el artefacto del modelo fuera del servicio de IA
4. Utilice un archivo de configuración para consultar los modelos y la lógica

El código completo se puede encontrar en GitHub.

Versión 1: los fundamentos

El siguiente código muestra una clase que cubre todos los pasos del servicio de IA como en la Figura 2, aunque la «carga de datos» y la «devolución de resultados» solo se incluyen implícitamente.

import pandas as pd
from sklearn.externals import joblib
class AIService:
"""A class for performing inference with a pre-trained machine
learning model."""
def __init__(self, model_path: str) -> None:
"""Initialize the AIService object with a pre-trained machine
learning model."""
self.model = joblib.load(model_path)
def preprocess(self, input_data: pd.DataFrame) -> pd.DataFrame:
"""Preprocess the input data."""
return input_data.drop(["id"], axis=1)
def inference(self, preprocessed_data: pd.DataFrame) -> pd.DataFrame:
"""Perform inference with the preprocessed data."""
return self.model.predict(preprocessed_data)
def postprocess(self, input_data, predictions: pd.DataFrame) -> pd.DataFrame:
"""Postprocess the model's predictions."""
return pd.DataFrame(
{
"id": input_data["id"],
"prediction": predictions,
}
)
def orchestrate(self, input_data: pd.DataFrame) -> pd.DataFrame:
"""Run the entire inference pipeline on the input data."""
preprocessed_data = self.preprocess(input_data)
predictions = self.inference(preprocessed_data)
return self.postprocess(input_data, predictions)
my_model_service = AIService("my_model.joblib")
result = my_model_service.orchestrate(input_data)

Tenga en cuenta que nuestra clase no cubre otras tareas, como las interacciones de almacenamiento de datos. Según mi experiencia, durante las primeras etapas de desarrollo, la infraestructura de almacenamiento de datos sufre cambios frecuentes. Por lo tanto, al separar el almacenamiento de datos de otros servicios, logramos un acoplamiento flexible.

En este enfoque, el preprocesamiento y la inferencia están estrechamente relacionados, ya que la inferencia se basa en gran medida en el preprocesamiento. Diferentes tipos de modelos o versiones pueden requerir diferentes características. Por lo tanto, están estrechamente acoplados en nuestro enfoque.

Si el preprocesamiento lleva demasiado tiempo durante el proceso de inferencia, como en los casos en que la aplicación se expone como una API o cuando se usa en un diseño de transmisión, podemos considerar una tienda de funciones en línea. Mediante el uso de una tienda de funciones en línea, podemos precalcular las funciones y recuperarlas durante la inferencia, lo que reduce el tiempo de inferencia.

Versión 2: agregar contratos de servicio para la validación de datos

Para mantener una estructura de datos coherente entre el servicio de orquestación y el servicio de IA, es importante validar los datos entrantes y salientes. Al hacerlo, podemos asegurarnos de que nuestros servicios de IA se puedan actualizar independientemente de otros componentes sin romper la aplicación.

La forma en que construyas esto depende de la implementación de tu aplicación. Para nuestro ejemplo, usamos pydantic para validar los tipos de datos entrantes y salientes. La estructura de nuestro servicio de IA queda de la siguiente manera:

La validación de datos implementada en el enfoque basado en clases es la siguiente, donde se agrega lo siguiente:

• pydantic BaseModel clases para validar datos de entrada y datos de salida;
• Un método validate_dataframe;
• un ajustado orchestrate método que incluye dos nuevas referencias a la validate_dataframe método.

import pandas as pd
from pydantic import BaseModel, validator
import joblib
from fastapi.encoders import jsonable_encoder
class InputData(BaseModel):
id: intpython
feature1: float
feature2: float
@validator("id")
def id_must_be_positive(cls, v):
if v <= 0:
raise ValueError("id must be a positive integer")
return v
class OutputData(BaseModel):
id: int
prediction: int
@validator("id")
def id_must_be_positive(cls, v):
if v <= 0:
raise ValueError("id must be a positive integer")
return v
class AIService:
"""A class for performing inference with a pre-trained machine
learning model."""
def __init__(self, model_path: str) -> None:
"""Initialize the AIService object with a pre-trained machine
learning model."""
self.model = joblib.load(model_path)
def validate_dataframe(
self, schema: BaseModel, dataframe: pd.DataFrame
) -> pd.DataFrame:
list_validated_data = [schema(**row) for _, row in dataframe.iterrows()]
return pd.DataFrame(jsonable_encoder(list_validated_data))
def preprocess(self, input_data: pd.DataFrame) -> pd.DataFrame:
"""Preprocess the input data."""
return input_data.drop(["id"], axis=1)
def inference(self, preprocessed_data: pd.DataFrame) -> pd.DataFrame:
"""Perform inference with the preprocessed data."""
return self.model.predict(preprocessed_data)
def postprocess(self, input_data, predictions: pd.DataFrame) -> pd.DataFrame:
"""Postprocess the model's predictions."""
return pd.DataFrame({"id": input_data["id"], "prediction": predictions})
def orchestrate(self, input_data: pd.DataFrame) -> pd.DataFrame:
"""Run the entire inference pipeline on the input data."""
validated_input = self.validate_dataframe(InputData, input_data)
preprocessed_data = self.preprocess(validated_input)
predictions = self.inference(preprocessed_data)
postprocessed_data = self.postprocess(input_data, predictions)
return self.validate_dataframe(OutputData, postprocessed_data)
my_model_service = AIService("my_model.joblib")
result = my_model_service.orchestrate(input_data)
print(result)

Versión 3: almacene el artefacto modelo fuera del servicio AI

Anteriormente, supusimos que el artefacto del modelo se almacenaría dentro del servicio del modelo. Sin embargo, un enfoque alternativo es separar el almacenamiento del código y los artefactos del modelo mediante la introducción de un registro modelo. Esto permite actualizaciones independientes del modelo a la última versión sin afectar el servicio directamente.

Para implementar el registro modelo en nuestro ejemplo, utilicé MLflow. El modelo se carga desde el registro durante el proceso de inicialización. El código actualizado se presenta a continuación. En aras de la claridad, excluí los contratos de datos del ejemplo y solo incluí el modificado __init__ método.

import pandas as pd
from sklearn.linear_model import LogisticRegression
import mlflow
class AIService:
"""A class for performing inference with a pre-trained machine
learning model."""
def __init__(self, model_name, model_version):
"""Initialize the MyModel object with the configuration file path."""
model_version = None if model_version == "latest" else int(model_version)
model_uri = f"models:/{model_name}/{model_version}"
self.model = mlflow.sklearn.load_model(model_uri)
my_model_service = AIService("sklearn-linear-regression","latest")
result = my_model_service.orchestrate(input_data)
print(result)

Versión 4: use un archivo de configuración para referirse a modelos y lógica

El paso final que explicaré es la utilización de un archivo de configuración. En lugar de especificar directamente el nombre y la versión del modelo durante la inicialización del servicio, podemos pasar un archivo de configuración que contiene toda la información necesaria. Este enfoque proporciona una forma más optimizada y organizada de administrar la configuración del servicio.

Otro caso de uso para el archivo de configuración es hacer referencia a la lógica de preprocesamiento. Como se mencionó anteriormente, el preprocesamiento y el modelo están estrechamente relacionados. Por ejemplo, una versión más nueva de un modelo puede requerir una nueva característica. Por lo tanto, necesitamos flexibilidad para actualizar la lógica de preprocesamiento independientemente de otras partes del servicio. Con la ayuda de un archivo de configuración, podemos ajustar fácilmente la lógica de preprocesamiento sin afectar el resto del servicio.

En el código, se ve de la siguiente manera. Un archivo de configuración config/config.yaml:

model:
name: sklearn-linear-regression
version: latest
preprocessing:
function_name: preprocess
module_path: src.sklearn_linear_regression.preprocessing

Lógica de preprocesamiento almacenada en src/preprocessing.py:

def preprocess(input_data):
"""Preprocess the input data."""
return input_data.drop(["id"], axis=1)

Tenga en cuenta que esta función reemplaza lapreprocessing método en la clase. A continuación se muestra la actualización __init__ método de la clase:

class AIService:
"""A class for performing inference with a pre-trained machine
learning model."""
def __init__(self, config_path):
"""Initialize the MyModel object with the configuration file path."""
with open(config_path) as f:
config = yaml.safe_load(f)
# Load the model
model_version = config["model"]["version"]
model_version = None if model_version == "latest" else int(model_version)
model_uri = f"models:/{config['model']['name']}/{model_version}"
self.model = mlflow.sklearn.load_model(model_uri)
# Load preprocessing logic
preprocess_module = importlib.import_module(
config["preprocessing"]["module_path"], "src"
)
self.preprocess = getattr(
preprocess_module, config["preprocessing"]["function_name"]
)
config_path = "config/config.yaml"
my_model_service = AIService(config_path)
result = my_model_service.orchestrate(input_data)

El servicio de IA que hemos definido permite ejecutar diferentes modelos en la aplicación con facilidad. Para ser más específicos, podemos convertir el servicio de IA en un servicio de scikit-learn y crear varias instancias del mismo, como una para la regresión logística y otra para un modelo de bosque aleatorio. Además, podemos crear un servicio de TensorFlow para ejecutar uno o más modelos de TensorFlow. La orquestación de aplicaciones maneja la comunicación con estos servicios.