Tech Blog | Portando Flask a FastAPI para el servicio de modelos ML

Una de las dificultades comunes que he visto con los nuevos científicos de datos que llegan a la industria es conseguir que sus modelos de aprendizaje automático interactúen con los sistemas de producción, es decir, dar el salto desde su entorno de investigación (normalmente desarrollando en un portátil Jupyter o similar) a un ecosistema de código más amplio.

Flask es un marco de trabajo muy popular para la construcción de API REST en Python, y casi la mitad de los codificadores de Python informan de su uso en 2019. En particular, Flask es útil para servir a los modelos ML, en los que la simplicidad y la flexibilidad son más deseables que la funcionalidad «baterías incluidas» de otros marcos de trabajo más orientados al desarrollo general de la web. Sin embargo, desde la introducción de Flask ha habido una serie de desarrollos en el rendimiento de Python, el comportamiento de la anotación de tipos y la interfaz de programación asíncrona, lo que nos permite construir una API más rápida, más robusta y con mayor capacidad de respuesta – aquí aprenderemos cómo migrar al nuevo marco de trabajo FastAPI para aprovechar estos avances.

Este post asumirá cierta familiaridad con Flask y las peticiones HTTP, aunque primero revisaremos la construcción de la aplicación en Flask. El código completo de dockerized para este post se puede encontrar aquí:

• configurando la aplicación FastAPI y ejecutando el servidor de sincronización con el uvicornio y el gunicornio
• especificación de la ruta y puntos finales
• validación de datos
• puntos finales asincrónicos

Cosas que no cubriremos:

• unidad de prueba de su API – pero usted puede diseñar pruebas utilizando la starlette TestClient que se comportan más o menos como Flask$0027s
• ya que eso es muy específico de la organización, pero la aplicación se puede acoplar fácilmente y se ejecuta en servidores listos para la producción con gunicornio, y hemos incluido un ejemplo de configuración en el ejemplo de código completo.

un simple Flask API

En primer lugar, necesitaremos un modelo. El código de demostración enlazado más arriba proporciona un guión para entrenar un simple clasificador Bayes ingenuo en un subconjunto del conjunto de datos de los «20 grupos de noticias» disponibles en scikit-learn (usando los grupos de noticias rec.sport.hockey, sci.space, y talk.politics.misc), pero siéntete libre de construir y escabechar tu propio clasificador de texto en el mismo – cualquier paquete ML con una API razonable funcionará bien en este marco.

Comencemos con una simple aplicación Flask de un solo archivo que sirve a este modelo. Primero, el contenido del encabezado:

importarpickledeFlaskimportFlask,jsonify,requestimportnumpyasnpapp=Flask(__nombre__)conhopen(os.getenv("MODELO_PATH"), "rb")asrf:clf=pickle.load(rf)@app. route("/healthcheck",methods=["GET"])defhealthcheck():msg=("esta sentencia ya está a medio camino, ""y todavía no ha dicho nada en absoluto")returnjsonify({"message":msg})

en el que configuramos nuestro objeto de aplicación Flask y cargamos el modelo en memoria al iniciar la aplicación – los métodos de predicción reales harán referencia a este objeto de modelo como un cierre, evitando la innecesaria entrada y salida de archivos para cada llamada al modelo. (Dejaremos de lado el encontrar un mejor paradigma de carga de modelos que el pepinillo por ahora, ya que eso dependerá de las especificaciones del esquema de despliegue y del modelo involucrado).

Incluir un punto final de «chequeo de salud» que devuelva un simple estado de «estoy bien» sin computación a las solicitudes GET es útil tanto para el monitoreo del despliegue, como para asegurar que su propio entorno de desarrollo está configurado correctamente.lo anterior debería producir una API que funcione (aunque ligeramente tonta) en su entorno – pruébela ejecutando desde su shell

$ exportFLASK_APP=<path to your app file >.py$ flask run

(o utilizar la invocación dockerizada del código de ejemplo).

puntos finales de la predicción

A continuación, agreguemos alguna funcionalidad de predicción real:

@app.route("/predict",methods=["POST"])defpredict():samples=request.get_json()["samples"]data=np.array([sample["text"]forsampleinsamples])probas=clf.predict_proba(data)predictions=probas. argmax(axis=1)returnjsonify({"predicciones":(np.tile(clf.classes_,(len(predicciones),1))[np.arange(len(predicciones)),predicciones].tolist()), "probabilidades":probas[np.arange(len(predicciones)),predicciones].tolist()})

Este es el típico patrón de diseño que usamos para servir las predicciones – los clientes POST introducen los datos al punto final y reciben las predicciones de vuelta en un formato fijo, con el modelo en sí llamado vía cierre.Esto espera un cuerpo de solicitud del formulario

{"muestras":[{"texto": "este es un texto sobre hockey"}]}

Así que le pasamos una serie de muestras, cada una de las cuales contiene un conjunto de valores clave de los datos de entrada de texto que esperamos (para modelos con múltiples características distintas, pasar cada característica como un par de valores clave). Esto asegura que estamos accediendo explícitamente a los datos en la forma que queremos, juega bien con los pandas y otras herramientas de datos tabulares, y refleja la forma de los puntos de datos individuales que podríamos esperar de una fuente de datos de producción. Nuestro método predict() lo desempaqueta en una matriz, lo pasa a través del objeto de tubería del clasificador scikit-learn, y devuelve la etiqueta de predicción (mapeada al nombre de la clase) y su correspondiente probabilidad.nótese que podríamos haber llamado simplemente al método de predicción del clasificador en lugar de mapear las salidas de probabilidad predict_proba de vuelta a las etiquetas de clase, pero algún simple álgebra numérica debería ser generalmente sustancialmente más rápida que ejecutar la inferencia por segunda vez.

@app.route("/predict/<label>",methods=["POST"])defpredict_label(label):samples=request.get_json()["samples"]data=np.array([sample["text"]forsampleinsamples])probas=clf. predict_proba(data)target_idx=clf.classes_.tolist().index(label)returnjsonify({"label":label, "probabilities":probas[:,target_idx].tolist()})

De manera similar, podemos solicitar la probabilidad específica de una etiqueta dada para cada muestra – este tipo de parametrización se hace fácilmente a través del propio camino. ¡Y ahí lo tenemos!

Esta API funcionará para servir predicciones (¡pruébela usted mismo!)… pero es muy probable que se caiga si intenta ponerla en producción. En particular, carece de cualquier validación de datos – cualquier información faltante, mal nombrada o mal tecleada en la solicitud entrante da lugar a una excepción no manejada, devolviendo un error 500 y una respuesta singularmente poco útil de Flask (junto con algún HTML bastante intimidante en modo de depuración), mientras que potencialmente dispara alertas en sus sistemas de monitoreo y despierta sus DevOps.

A menudo, el manejo de errores en Flask termina con una frágil y enmarañada mezcla de try-catches y acceso protegido a dict. Los mejores enfoques utilizarán un paquete como pydantic o malvavisco para lograr una validación de datos más programática. Afortunadamente, FastAPI incluye la validación pydantic fuera de la caja.

introduzca FastAPI

FastAPI es un moderno marco web de Python diseñado para:

• proporcionar un micro-marco ligero con un sistema de enrutamiento intuitivo, similar al de un frasco
• Aprovechar el soporte de anotación de tipos en Python 3.6+ para la validación de datos y el soporte de editor
• utilizar las mejoras en el soporte de asincronía de Python y el desarrollo de la especificación ASGI para facilitar las APIs asincrónicas
• Generar automáticamente documentación útil de la API usando OpenAPI y JSON Schema

Bajo el capó, FastAPI está usando pydantic para la validación de datos y starlette para sus herramientas web, lo que lo hace ridículamente rápido comparado con marcos como Flask y da un rendimiento comparable a las APIs web de alta velocidad en Node o Go.

Afortunadamente, como FastAPI se inspiró explícitamente en Flask para inspirar su especificación de ruta, la transición a su uso es bastante rápida – empecemos a portar la funcionalidad de nuestra aplicación de servicio sobre el contenido de la cabecera correspondiente:

importosimportpicklefromfastapiimportFastAPIimportnumpyasnpapp=FastAPI()withopen(os.getenv("MODEL_PATH"), "rb")asrf:clf=pickle.load(rf)@app. get("/healthcheck")defhealthcheck():msg=("esta frase ya está a medio camino, ""y todavía no ha dicho nada")return{"message":msg}

Fácil hasta ahora – sólo hemos tenido que hacer pequeños cambios semánticos:

• instanciar el objeto de nivel superior Flask(__nombre__) se convierte en instanciar un objeto FastAPI()
• las rutas se siguen especificando con un decorador, pero con el método HTTP en el propio decorador en lugar de un argumento – @app.route(…, methods=[«GET»]) se convierte en @app.get(…)

Podemos invocar esto como el lanzamiento de la aplicación de flask, aunque no incluye un servidor web incorporado – en su lugar, lanzaremos directamente el servidor de uvicornio,

$ uvicornio --reload <path to app file>:app

que nos da una simple configuración de un solo trabajador (o, de nuevo, sólo usar la invocación dockerizada en el código de ejemplo).

validación de datos

Para especificar el punto final de la predicción, tenemos que introducir un cambio importante en nuestra forma de pensar. En las simples validaciones mencionadas anteriormente, cosas como las capturas de prueba y la protección de los dictados están tratando esencialmente de abofetear las cosas que no queremos con nuestros datos. ¿Qué pasa si, en cambio, simplemente especificamos lo que nosotros queremos y dejamos que la aplicación se encargue del resto? Esto es exactamente lo que hacen las herramientas de validación como marshmallow o pydantic – para pydantic en FastAPI, simplemente especificamos el esquema usando las nuevas anotaciones de tipo de Python (3.6+) y lo pasamos como argumento a la función de ruta.FastAPI sabe qué hacer con la validación gracias a la anotación de tipo de python, lo que significa que sólo necesitamos especificar de forma muy natural lo que esperamos para las entradas y dejar que el resto ocurra bajo el capó.Para nuestro punto final de predict(), esto parece

fromtypingimportListfrompydanticimportBaseModelclassTextSample(BaseModel):text:strclassRequestBody(BaseModel):samples:List[Sample]

Simplemente especificamos la forma de los datos esperados (usando anotaciones de tipo Python base aquí, aunque pydantic soporta comprobaciones de tipo extendidas para cosas como validaciones de cadena/correo electrónico o dimensiones y normas de matriz) en una clase infantil de pydantic.BaseModel. Mientras que pydantic construye su funcionalidad de comprobación de tipos en esta clase, podemos seguir usándola como un objeto Python ordinario – subclasificando o añadiendo funciones adicionales funciona como se espera. Por ejemplo, podríamos construir la funcionalidad para desempaquetar el contenido de nuestras muestras en un array para su inferencia en la clase, como

classRequestBody(BaseModel):samples:List[Sample]defto_array(self):return[sample.textforsampleinself.samples]

reemplazando la comprensión de la lista usada anteriormente y garantizando el formato correcto de la matriz. En el punto final, pasamos los datos de entrada como un argumento de función:

@app.post("/predicto")defpredict(body:RequestBody):data=np.array(body.to_array())probas=clf.predict_proba(data)predictions=probas.argmax(axis=1)return{"predicciones":(np. tile(clf.classes_,(len(predicciones),1))[np.arange(len(predicciones)),predicciones].tolist()), "probabilidades":probas[np.arange(len(predicciones)),predicciones].tolist(),}

FastAPI manejará esto inteligentemente, encontrando primero los parámetros de ruta por nombre, luego empaquetando cuerpos de solicitud (para solicitudes POST) o parámetros de consulta (para GET) en los argumentos de la función. Los campos de datos o métodos resultantes pueden entonces ser accedidos a través de la típica sintaxis de atributo/método. En el caso de datos de entrada malformados, pydantic levantará un error de validación – FastAPI maneja esto internamente, devolviendo un código de error 422 con un cuerpo JSON que contiene información útil sobre el error.

valores enumerados & parámetros del camino

También podemos usar valores enumerados en nuestra validación de datos – por ejemplo, en el punto final predict_label() podemos manejar nombres de objetivos válidos con

fromenumimportEnumclassResponseValues(str,Enum):hockey="rec.sport.hockey "space="sci.space "politics="talk.politics.misc"

Pasar esto para validar el parámetro de la ruta manejará limpiamente los errores por malos nombres de objetivos, lo que ahogaría el intento de encontrar un índice del valor correspondiente en clf.classes_.En el punto final, entonces tenemos

@app.post("/predict/{label}")defpredict_label(label:ResponseValues,body:RequestBody):data=np.array(body.to_array())probas=clf.predict_proba(data)target_idx=clf.classes_.tolist().index(label.value)return{"label":label.value, "probabilities":probas[:,target_idx].tolist()}

modelos de respuesta y documentación

Hasta ahora, sólo hemos añadido validación de datos en nuestras entradas, pero FastAPI nos permite declarar un esquema para la salida también – definimos

classResponseBody(BaseModel):predicciones:List[str]probabilities:List[float]classLabelResponseBody(BaseModel):label:strprobabilities:List[float]

y reemplazar los decoradores de ruta con

@app.post("/predict",response_model=ResponseBody)...@app.post("/predict/{label}",response_model=LabelResponseBody)...

Puede parecer extraño añadir validación de datos a nuestros resultados – después de todo, si lo que estamos devolviendo no se ajusta al esquema, eso indica un problema más profundo en nuestro código, Más importante para nosotros es que estos esquemas se incorporan a la documentación autogenerada de FastAPI – cuando la API se está ejecutando, al pulsar los puntos finales {api url}/docs o {api url}/redoc se cargará la documentación generada por OpenAPI que detalla los puntos finales disponibles y la estructura esperada de sus entradas y salidas (que se derivan de nuestros esquemas). Podemos incluso añadir anotaciones a la API y sus puntos finales:- las rutas individuales pueden ser etiquetadas para su categorización (útil para el versionado de la API) – las cadenas de documentación de las funciones de ruta serán introducidas en la documentación de la API para la descripción de los puntos finales – el propio objeto FastAPI puede recibir argumentos de título, descripción y palabras clave de la versión, que se rellenan en la documentación

puntos finales asincrónicos

Francamente, esto no es poco común entre los científicos de datos e ingenieros de ML – tanto entrenamiento e inferencia de modelos está limitado por el procesador que el código asíncrono no aparece mucho, comparado con (por ejemplo) el desarrollo web donde es mucho más común. Para profundizar más en esto, mira la propia documentación de async de Python, aunque en realidad encuentro que la propia explicación de FastAPI es mucho más intuitiva para captar las diferencias entre el código concurrente y el paralelo.

En resumen, la idea de la ejecución asíncrona (o de la concurrencia, si se prefiere) se refiere a que el proceso dispare el trabajo a un recurso externo, como una solicitud a una API externa o a un almacén de datos. En el código síncrono, el proceso se bloquea hasta que se completa ese trabajo; en el caso de una solicitud lenta, eso significa que todo el proceso está inactivo hasta que recibe un valor de retorno. La ejecución asíncrona permite al proceso cambiar de contexto y trabajar en algo no relacionado hasta que se indica que el trabajo solicitado se ha completado, en cuyo momento se reanuda. (Por el contrario, la ejecución de código paralelo tendría múltiples líneas de trabajo, todas ellas ejecutando ese código potencialmente bloqueador de forma independiente unas de otras)

En el aprendizaje automático, la ejecución suele estar vinculada al procesador, es decir, el trabajo se suscribe constantemente a la capacidad de procesamiento de uno o varios núcleos de la CPU. La ejecución asíncrona no es particularmente útil en este caso, ya que no hay realmente situaciones en las que el procesador pueda dejar de trabajar para esperar un resultado mientras se ejecuta otra cosa (por el contrario, hay amplios escenarios en los que el trabajo de ML se puede computar en paralelo). Sin embargo, todavía podemos imaginar situaciones en las que nuestro modelo que sirve la API podría estar esperando un recurso externo en lugar de hacer el trabajo pesado de computación por su cuenta. Por ejemplo, supongamos que nuestra API debe solicitar información de una base de datos o una caché en memoria basada en su computación, o que nuestra API es un intermediario ligero que realiza la validación o el preprocesamiento antes de realizar el trabajo en una API separada que sirve el flujo tensorial y que se ejecuta en una instancia de la GPU, el manejo adecuado del procesamiento asíncrono puede dar a nuestro trabajo un importante impulso de rendimiento a un bajo costo.

Históricamente, el trabajo de asincronía en Python no ha sido trivial (aunque su API ha mejorado rápidamente desde Python 3.4), particularmente con Flask.esencialmente, Flask (en la mayoría de los servidores WSGI) está bloqueando por defecto – el trabajo desencadenado por una solicitud a un punto final particular mantendrá al servidor completamente hasta que esa solicitud sea completada. En cambio, Flask (o mejor dicho, el servidor WSGI que lo ejecuta, como gunicornio o uWSGI) logra el escalamiento ejecutando múltiples instancias de trabajadores de la aplicación en paralelo, de tal manera que las solicitudes pueden ser cultivadas a otros trabajadores mientras uno está ocupado. Dentro de un solo trabajador, el trabajo asíncrono puede ser envuelto en una llamada de bloqueo (la función de ruta en sí sigue bloqueando), enhebrado (en las versiones más recientes de Flask), o enviado a un gestor de colas como Celery – pero no hay una sola historia consistente en la que las rutas puedan manejar limpiamente las peticiones asíncronas sin herramientas adicionales.

async con FastAPI

Por el contrario, FastAPI está diseñado desde el principio para funcionar asincrónicamente – gracias a su marco ASGI de starlette subyacente, las funciones de ruta por defecto a ejecutarse dentro de un bucle de eventos asíncronos. Con un buen servidor ASGI (FastAPI está diseñado para acoplarse al uvicornio, ejecutándose encima del uvloop) esto puede conseguirnos un rendimiento a la par que los servidores web asíncronos rápidos en Go o Node, sin perder los beneficios del más amplio ecosistema de aprendizaje de máquinas de Python.

A diferencia de lo que ocurre con los hilos o las colas de Celery para conseguir una ejecución asíncrona en Flask, ejecutar un punto final de forma asíncrona es muy sencillo en FastAPI: simplemente declaramos la función de ruta como asíncrona (con async def) y ¡ya estamos listos! Podemos hacerlo incluso si la función de ruta no es convencionalmente asíncrona, es decir, no tenemos ninguna llamada en espera (como si el punto final está ejecutando la inferencia contra un modelo ML). De hecho, a menos que el punto final esté realizando específicamente una operación IO de bloqueo (a una base de datos, por ejemplo), es mejor declarar la función con async def (ya que las funciones de bloqueo son en realidad punteadas a un threadpool externo y luego esperadas de cualquier manera).

Para nuestras funciones de predicción de ML arriba, podemos declarar los puntos finales con async def, aunque eso no hace realmente ningún cambio interesante en nuestro código. ¿Pero qué pasa si necesitáramos hacer algo verdaderamente asíncrono, como solicitar (y esperar) un recurso de una API externa? Desafortunadamente, nuestro paquete de peticiones convencionales en Python se bloquea, por lo que no podemos usarlo para hacer peticiones HTTP de forma asíncrona – en su lugar, usaremos la funcionalidad de petición en el excelente paquete aiohttp.

solicitudes de asincronización

En primer lugar, tendremos que configurar una sesión de cliente – esto mantendrá un pool persistente funcionando para esperar peticiones de, en lugar de crear una nueva sesión para cada petición (que es en realidad lo que hacen las peticiones si se llaman como las típicas requests.get, requests.post, etc.):

app=FastAPI()...client_session=aiohttp.ClientSession()

También tendremos que asegurarnos de que esta sesión se cierra correctamente – afortunadamente, FastAPI nos da un decorador fácil para declarar estas operaciones:

@app.on_event("shutdown")asyncdefcleanup():awaitclient_session.close()

Esto ejecutará cualquier cosa llamada dentro de la función (esperando el cierre limpio de la sesión del cliente aiohttp, aquí) cuando la aplicación FastAPI se apague. Para la petición externa, envolvemos una llamada en espera en la función de ruta:

@app.get("/cat-facts",response_model=TextSample)asyncdefcat_facts():asyncwithclient_session.get(url)asresp:response=awaitresp.json()returnresponse

colocando la petición dentro de un bloque de contexto asíncrono, y luego esperando una respuesta que pueda ser analizada. Aquí hemos hecho uso de nuestros modelos de respuesta para restringir los valores de retorno – la respuesta de la API «cat facts» devuelve un montón de metadatos adicionales sobre el hecho, pero sólo queremos devolver el texto del hecho. En lugar de manipular la respuesta antes de devolverla, podemos simplemente reutilizar nuestro actual esquema TextSample para empaquetarla en la respuesta y confiar en pydantic para que se encargue del filtrado, por lo que nuestra respuesta se parece a

y eso es todo! Podemos usar esta construcción para cualquier llamada asincrónica a recursos externos que podamos necesitar, como recuperar datos de un almacén de datos o disparar trabajos de inferencia para tensorizar el flujo que se ejecuta en los recursos de la GPU.

envolviendo

En este post, hemos caminado a través de un diseño común y simple para poner de pie sus modelos de aprendizaje de máquina detrás de una API REST, permitiendo las predicciones sobre la marcha de una manera consumible que debería interactuar limpiamente con una variedad de entornos de código. Aunque Flask es un marco de trabajo muy común para estas tareas, podemos aprovechar las mejoras en la comprobación de tipos de Python y el soporte asíncrono al migrar al nuevo marco de trabajo FastAPI – afortunadamente, ¡portar a FastAPI desde Flask es sencillo!

Categorías: technicalTags: python, machine learning