Desarrollo de microservicio de inferencia en API

Un microservicio de inferencia en API se refiere a un tipo de microservicio que proporciona un modelo de inteligencia artificial entrenado para hacer predicciones o inferencias en los datos de entrada.

Estos microservicios se utilizan para proporcionar una API que le permita a los desarrolladores consumir los modelos de inferencia como un servicio.

En general, los microservicios de inferencia en API se utilizan en aplicaciones de inteligencia artificial, como la clasificación de imágenes, el procesamiento del lenguaje natural o la detección de objetos.

Por ejemplo, una aplicación de clasificación de imágenes podría utilizar un microservicio de inferencia en API para predecir las etiquetas de las imágenes basadas en un modelo de aprendizaje automático entrenado previamente.

Desarrollo de microservicio de inferencia en API

Tenemos un proyecto con la siguiente estructura:

En esta estructura distinguimos los siguientes componentes:

• Dockerfile: aquí definimos la imagen base que usaremos y cómo empaquetamos el proyecto.
• Requirements.txt: especificación de dependencias a instalar en el microservicio.
• App: aplicación de inferencia online en FastAPI.
  • main.py: punto de entrada para la ejecución de la aplicación.
  • models: en este encontramos la definición de los esquemas que usaremos dentro de la aplicación.
  • services: en este módulo incluiremos la implementación de nuestra clase de inferencia.
  • api/routes: módulo en el que definiremos los diferentes endpoints que tendrá la API.
  • api/core: módulo en el que habrá funcionalidades comunes del servicio y configuraciones.

#microservicio de inferencia en API
%cd /content /twitter-sentiment-analysis-online/

#microservicio de inferencia en API
import os

os.environ ["DEFAULT_MODEL_PATH"] = "/twitter-sentiment-analysis-online/"

#microservicio de inferencia en API
! gsutil -m cp \
  "gs: //$BUCKET_NAME/ models / model.h5" \
  "gs: //$BUCKET_NAME/ models / tokenizer.pkl" \

Para probar el correcto funcionamiento del servicio en local:

#microservicio de inferencia en API
import nest_asyncio
from pyngrok import ngrok, conf

conf.get:default ().auth_token = "25TtT902JbR7LogkJKxk70jhCz6_4Z6JeyChZAau4gkr2th42"

ngrok_tunnel = ngrok.connect (8000)
print ('Public URL:', ngrok_tunnel.public_url)
nest_asyncio.apply ()

#microservicio de inferencia en API
! uvicorn app.main:app --port 8000

En el siguiente paso lo que haremos será rellenar la aplicación. Para ello, crearemos un documento denominado main.py (en caso de que no esté creado aún) y escribimos lo siguiente:

#microservicio de inferencia en API
from fastapi import FastAPI

from app.api.routes.router import api_router
from app.core.config import (
         API_PREFIX,
         APP_NAME,
         APP_VERSION,
         IS_DEBUG,
)
from app.core.event_handlers import start_app_handler, stop_app_handler

def get_app ( ) -> FastAPI:
       fast_app = FastAPI (title = APP_NAME, version = APP_VERSION, debug = IS_DEBUG)
       fast_app.include_router (api_router, prefix = API_PREFIX)

       fast_app.add _event_handler ("startup", start_app_handler (fast_app))
       fast_app.add _event_handler ("shutdown", stop_app_handler (fast_app))
       
       return fast_app

app = get_app ()

Ahora, crearemos una carpeta en la siguiente ruta:

Ahí escribiremos lo siguiente:

from fastapi import APIRouter

from app.api.routes import heartbeat, prediction

api_router = APIRouter ()
api_router.include_router (heartbeat.router, tags = ["health"], prefix = "/health")
api_router.include_router (prediction.router, tags = ["prediction"], prefix = "/model")

Ahora nos iremos al archivo predict.py y escribiremos esto:

#microservicio de inferencia en API
from fastapi import APIRouter, Depends
from starlette.requests import Request

from app.models.payload import TextPayLoad
from app.models.prediction import SentimentPredictionResult
from app.services.models import SentimentAnalysisModel

router = APIRouter ()

@router.post ("/predict", response_model = SentimentPredictionResult, name = "predict")
def post_predict (
       request: Request,
       data: TextPayLoad = None,
)   -> SentimentPredictionResult:
       model: SentimentAnalysisModel = request.app.state.model
       prediction: SentimentPredictionResult = model.predict (data)

       return prediction

En el archivo payload.py escribimos lo siguiente:

from pydantic import BaseModel

classTextPayload (BaseModel):
      text: str


def payload_to_text (payload: TextPayload) -> str:
       return payload.text

Ahora tendremos que insertar información en otro archivo de salida denominado prediction.py (diferente al prediction.py de más arriba):

from pydantic import BaseModel

from app.core.enums import Sentiment

class SentimentPredictionResult (BaseModel):
       label: Sentiment
       score: float
       elapsed_time: float

En el archivo enums.py escribiremos:

from enum import Enum

class Sentiment (Enum):
      NEGATIVE = "NEGATIVE"
      NEUTRAL = "NEUTRAL"
      POSITIVE = "POSITIVE"

Ahora modificaremos nuestro archivo models.py:

import os
import time
import tempfile

import pickle
import tensorflow as tf
from loguru import logger
from tensorflow.keras. preprocessing.sequence import pad_squences


from app.core.messages import NO_VALID_PAYLOAD
from app.models.payload import TextPayload, payload_to_text
from app.models.prediction import SentimentPredictionResult
from app.core.enums import Sentiment
from app.core.config import (
        KERAS_MODEL,
        TOKENIZER_MODEL,
        SEQUENCE_LENGTH,
        SENTIMENT_THRESHOLD,
)

class SentimentAnalysisModel:
        def __init__ (
                self, model_dir,
        ):
                self.model_dir = model_dir
                self._load_local_model ()
        
        def _ load_local_model (self):
                kears_model_path = os.path.join (self.model_dir, KERAS_MODEL)
                with tempfile.NamedTemporaryFile (suffix = ".h5") as local_file:
                        with tf.io.gfile.GFile (keras_model_path, mode = "rb") as gcs_file:
                                local_file.write (gcs_file.read ())
                                self.model = tf.keras.models.load_model (local_file.name, compile = False)

        tokenizer_path = os.path.join (self.model_dir, TOKENIZER_MODEL)
        self.tokenizer = pickle.load (tf.io.gfile.GFile (tokenizer_path, mode = "rb"))

        def _decode_sentiment (self, score: float, include_neutral = True) -> str:
                if include_neutral:
                        label = Sentiment.NEUTRAL.value

Si quieres seguir formándote en alguna de las numerosas y demandadas temáticas del mundo del Big Data, en KeepCoding podrás acceder a una formación de alta intensidad en la que contarás con la guía de grandes expertos a través de la teoría y la práctica para que, en unos pocos meses, te conviertas en un gran profesional IT.

No te pierdas el Big Data, Inteligencia Artificial & Machine Learning Full Stack Bootcamp, la formación intensiva e íntegra que catapultará tu carrera. ¡Solicita ahora mismo más información y da el salto que transformará tu vida!