[PYTHON] Configuration du serveur d'API Web d'inférence de modèle d'apprentissage automatique [Exemple d'implémentation d'API rapide disponible]

Objectif de cet article

Présentation d'une configuration typique de l'API Web d'inférence de machine learning. Je pense que le contenu peut être lu sans forcément avoir connaissance du WEB ou de l'apprentissage automatique. (Hors exemples d'implémentation) La composition à introduire provient de l'expérience de création d'API Web d'inférence pour certains modèles d'apprentissage automatique en entreprise, mais comme c'est mon opinion personnelle, s'il y a quelque chose de mieux, veuillez me le faire savoir dans les commentaires. Je suis heureux. Dans l'exemple d'implémentation, le framework Web utilise l'API Fast du point de vue de la facilité de gestion du traitement asynchrone et de la simplicité de la mise en œuvre.

table des matières

1. Configuration de l'API Web d'inférence de machine learning
2. Exemple de mise en œuvre

1. Configuration de l'API Web d'inférence de machine learning

Dans cet article, je présenterai deux modèles.

Note) Tout d'abord, je vais vous expliquer les parties communes. Les connaissances en machine learning ne sont fondamentalement requises que pour les parties communes. Si vous n'êtes pas familiarisé avec l'apprentissage automatique ou le Web, vous pouvez répartir les rôles entre la partie commune et la partie décrite plus loin, afin de pouvoir le laisser couler.

API d'inférence (partie commune)

Si vous souhaitez que le modèle entraîné en déduit, vous construirez généralement l'API d'inférence de modèle d'apprentissage automatique suivante. Même si vous ne développez que sur un PC local ou sur Jupyter Notebook, je pense que vous allez créer une telle API (pipeline).

Je vais omettre les détails, mais pour la commodité de la distribution de la charge et de la gestion des modèles, je pense que vous ne pouvez supprimer que l'API qui utilise le modèle d'apprentissage automatique pour le serveur sur le cloud (Référence: [GCP AI platform Prediction](https: /) /cloud.google.com/ai-platform/prediction/docs)). Dans le cas d'un modèle lourd où des problèmes de performances surviennent à moins que le GPU ne soit utilisé non seulement pour la charge mais aussi pour l'inférence, il n'est pas possible de le gérer avec un serveur pour les applications WEB courantes, donc je pense qu'il est plus flexible de pouvoir l'isoler. De plus, je pense que la même configuration sera utilisée lors de l'utilisation d'un service externe qui utilise un modèle entraîné.

À mesure que la quantité de données augmente, je pense qu'il sera nécessaire de prendre des mesures telles que le remplacement du prétraitement par un moteur de traitement de données à grande échelle tel que Google Cloud Dataflow.

Lors de la création d'une API Web basée sur l'API d'inférence développée sur un PC local ou sur Jupyter Notebook comme décrit ci-dessus, il existe principalement deux types de modèles qui peuvent être pris en compte. Ceux-ci traitent les données d'entrée et de sortie différemment.

--1.1. Prédiction en ligne (également appelée prédiction HTTP) --1.2. Prédiction par lots

(Le nom utilisé dans la plate-forme IA de GCP est utilisé. Référence: [Prédiction en ligne vs prédiction par lots](https://cloud.google.com/ml-engine/docs/tensorflow/online-vs-batch- prédiction? hl = ja))

1.1. Prévisions en ligne

Lorsqu'une requête http arrive, la fonction ML est activée et la sortie est immédiatement renvoyée par la réponse http. Chargez le poids une seule fois au démarrage du serveur. Lors du chargement des poids, il sera plus facile de changer de modèle si vous obtenez les poids du stockage cloud (stockage google, etc.).

avantage

• Déplacez simplement la fonction d'inférence qui fonctionne localement dans le cadre Web et cela fonctionnera.
• Étant donné que le résultat de l'inférence est renvoyé simplement en appuyant sur une API, il devient une API facile à atteindre. --Réponse rapide lorsque le modèle est petit et que les données sont petites

Désavantage

• L'API Web est souvent configurée pour expirer en dizaines de secondes à minutes du point de vue de la répartition de la charge, donc si cela prend beaucoup de temps à déduire, le processus échouera. Il ne convient donc pas aux modèles lourds ou au traitement de grandes quantités de données à la fois.

1.2. Prédiction par lots

Si la réponse ne peut pas être renvoyée immédiatement ou n'a pas besoin d'être renvoyée, le résultat de l'inférence de l'API ML est stocké dans un stockage sans répondre directement comme indiqué ci-dessous. Le processus peut être divisé en trois étapes comme indiqué ci-dessous. (Il est préférable que 2 et 3 soient séparés. L'API d'importation peut être intégrée à l'API ML.)

1. API de téléchargement: stockez les données à saisir dans le stockage (base de données, stockage en nuage, etc.)
2. API ML (exécution asynchrone): récupérez les données du stockage, exécutez les fonctions ML et enregistrez les résultats dans le stockage. Cependant, la réponse est renvoyée avant la fin du traitement.
3. télécharger l'API: obtenir les résultats du stockage et les renvoyer

Chaque API peut être faiblement couplée. Par conséquent, la mise en œuvre de l'API de téléchargement et de l'API de téléchargement est assez flexible. Il existe différentes manières de l'utiliser comme suit.

--Accumuler les données d'entrée pendant une certaine période de temps et en déduire à la fois à la fin de la journée --Inférence utilisant un modèle compliqué qui expire

• Mettez en cache le résultat de l'inférence et n'inférez pas à plusieurs reprises pour la même entrée
• Tel

En outre, l'implémentation des API de téléchargement et de téléchargement est correcte dans des langages autres que Python, et les API peuvent être sur différents serveurs tant qu'elles peuvent lire et écrire sur le même stockage. Vous pouvez lire et écrire dans Storage directement depuis le frontal sans passer par l'API. Surtout lorsque l'entrée / sortie est une image, il s'agit d'un flux plus simple pour gérer directement le stockage cloud.

avantage

• Plus de pannes dues au timeout --Haute degré de liberté
• L'API d'apprentissage peut être implémentée avec la même configuration.

Désavantage

――Il est difficile à utiliser car il est plus compliqué que la prédiction en ligne.

• Le traitement prend plus de temps que la prédiction en ligne

2. Exemple de mise en œuvre

Implémentons les API de prédiction en ligne et de prédiction par lots avec l'API Fast. En regardant l'exemple ci-dessous, je pense que si vous faites fonctionner correctement le pipeline d'inférence localement, vous sentirez que l'obstacle pour en faire une API Web est assez faible.

Ce qu'il ne faut pas faire

Cet article ne couvre pas les éléments suivants:

• security
• deploy

Qu'est-ce que FastAPI

Le framework Web de Python, qui est un microframework comme Flask. Ses atouts incluent des performances élevées, une facilité d'écriture, une conception fortement centrée sur les opérations de production et des fonctions modernes. En particulier, le traitement asynchrone est facile à gérer.

Ce qui suit est basé sur les connaissances de base de Fast API. Si vous souhaitez connaître les détails, veuillez vous référer aux éléments suivants, le cas échéant.

• Document officiel
• [FastAPI] Premiers pas avec le framework Web Python ASGI FastAPI

API d'inférence (partie commune)

Pour être polyvalent, nous définissons une simulation très approximative. Cela ne veut rien dire, mais c'est facile, alors je l'appellerai une tâche d'analyse des émotions pour le traitement du langage naturel.

Les fonctions requises sont les suivantes. Cependant, si seul le modèle est découpé sur un autre serveur, il n'est pas nécessaire de conserver la charge et le modèle.

• Chargement du modèle instantané en utilisant la charge de poids, le joblib, le cornichon, etc.
• Rétention de modèle
• Pipeline d'inférence

Cette fois, nous utiliserons un modèle qui renvoie des émotions aléatoires avec prédire. Je veux rendre le temps de traitement réel, donc je le fige pendant 20 secondes lors du chargement et le fige pendant 10 secondes lors de la prédiction.

ml.py

from random import choice
from time import sleep

class MockMLAPI:
    def __init__(self):
        # model instanse
        self.model = None

    def load(self, filepath=''):
        """
        when server is activated, load weight or use joblib or pickle for performance improvement.
        then, assign pretrained model instance to self.model.
        """
        sleep(20)
        pass

    def predict(self, x):
        """implement followings
        - Load data
        - Preprocess
        - Prediction using self.model
        - Post-process
        """
        sleep(10)
        preds = [choice(['happy', 'sad', 'angry']) for i in range(len(x))]
        out = [{'text': t.text, 'sentiment': s} for t, s in zip(x, preds)]
        return out

Format des données de demande / réponse

Définit le format des données de la demande. Essayons de prendre en charge plusieurs entrées comme indiqué ci-dessous.

{
  "data": [
    {"text": "hogehoge"},
    {"text": "fugafuga"}
  ]
}

Les données de réponse doivent être dans un format qui ajoute le résultat de l'inférence à l'entrée et le renvoie.

{
  "prediction": [
    {"text": "hogehoge", "sentiment": "angry"},
    {"text": "fugafuga", "sentiment": "sad"}
  ]
}

Alors, définissez le schéma comme suit.

schemas.py

from pydantic import BaseModel
from typing import List

# request
class Text(BaseModel):
    text: str

class Data(BaseModel):
    data: List[Text]

# response
class Output(Text):
    sentiment: str

class Pred(BaseModel):
    prediction: List[Output]

2.1. Prévisions en ligne

Implémentez une API Web pour la prédiction en ligne en utilisant les parties communes mentionnées ci-dessus. Tout ce dont tu as besoin c'est

--Chargez le modèle d'apprentissage automatique formé au démarrage du serveur --Recevoir des données, déduire avec l'API ML, renvoyer le résultat

est. L'API minimale est complétée par l'implémentation comme suit.

main.py

from fastapi import FastAPI
from ml_api import schemas
from ml_api.ml import MockMLAPI

app = FastAPI()
ml = MockMLAPI()
ml.load() # load weight or model instanse using joblib or pickle

@app.post('/prediction/online', response_model=schemas.Pred)
async def online_prediction(data: schemas.Data):
    preds = ml.predict(data.data)
    return {"prediction": preds}

Contrôle de fonctionnement

Vérifiez le fonctionnement localement. Publiez l'échantillon d'entrée dans CuRL. Ensuite, vous pouvez confirmer que la sortie attendue est renvoyée. De plus, comme il a fallu 10 secondes pour que la réponse soit renvoyée, vous pouvez voir que cela n'a pris presque que le temps de traitement prévu.

$ curl -X POST "http://localhost:8000/prediction/online" -H "accept: application/json" -H "Content-Type: application/json" -d "{\"data\":[{\"text\":\"hogehoge\"},{\"text\":\"fugafuga\"}]}" -w  "\nelapsed time: %{time_starttransfer} s\n"

{"prediction":[{"text":"hogehoge","sentiment":"angry"},{"text":"fugafuga","sentiment":"happy"}]}
elapsed time: 10.012029 s

2.1. Prédiction par lots

Implémentez une API Web pour la prédiction par lots en utilisant les parties communes mentionnées ci-dessus.

1. API de téléchargement: stockez les données à saisir dans le stockage (base de données, stockage en nuage, etc.)
2. API ML (exécution asynchrone): récupérez les données du stockage, exécutez les fonctions ML et enregistrez les résultats dans le stockage. Cependant, la réponse est renvoyée avant la fin du traitement.
3. télécharger l'API: obtenir les résultats du stockage et les renvoyer

Input/Output Normalement, vous devez enregistrer les données dans un stockage en nuage ou une base de données, mais par souci de simplicité, dans cet article, nous enregistrerons les données au format csv dans le stockage local. Tout d'abord, définissez une fonction de lecture et d'écriture. Lors de l'enregistrement des données d'entrée, un nom de fichier est créé avec une chaîne de caractères aléatoires, et une série de prédictions par lots est effectuée en échangeant la chaîne de caractères aléatoires avec l'api. La mise en œuvre peut sembler longue, mais en réalité il n'y a que trois choses à faire:

--Lecture et écriture de csv

• Ajustez le chemin du fichier --Génération d'une chaîne de caractères aléatoire

io.py

import os
import csv
from random import choice
import string
from typing import List
from ml_api import schemas

storage = os.path.join(os.path.dirname(__file__), 'local_storage')

def save_csv(data, filepath: str, fieldnames=None):
    with open(filepath, 'w') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)

        writer.writeheader()
        for f in data:
            writer.writerow(f)

def load_csv(filepath: str):
    with open(filepath, 'r') as f:
        reader = csv.DictReader(f)
        out = list(reader)
    return out

def save_inputs(data: schemas.Data, length=8):
    letters = string.ascii_lowercase
    filename = ''.join(choice(letters) for i in range(length)) + '.csv'
    filepath = os.path.join(storage, 'inputs', filename)
    save_csv(data=data.dict()['data'], filepath=filepath, fieldnames=['text'])
    return filename

def load_inputs(filename: str):
    filepath = os.path.join(storage, 'inputs', filename)
    texts = load_csv(filepath=filepath)
    texts = [schemas.Text(**f) for f in texts]
    return texts

def save_outputs(preds: List[str], filename):
    filepath = os.path.join(storage, 'outputs', filename)
    save_csv(data=preds, filepath=filepath, fieldnames=['text', 'sentiment'])
    return filename

def load_outputs(filename: str):
    filepath = os.path.join(storage, 'outputs', filename)
    return load_csv(filepath=filepath)

def check_outputs(filename: str):
    filepath = os.path.join(storage, 'outputs', filename)
    return os.path.exists(filepath)

web API Créez trois API: téléchargement, inférence et téléchargement. Notez que l'inférence par lots ne renvoie pas de réponse immédiate, donc chargez le modèle chaque fois que l'API est touchée.

Ici, BackgourndTasks de FastAPI est utilisé pour traiter l'inférence de modèle de manière asynchrone. L'inférence peut être traitée en arrière-plan et la réponse peut être renvoyée en premier sans attendre la fin.

main.py

from fastapi import FastAPI
from fastapi import BackgroundTasks
from fastapi import HTTPException
from ml_api import schemas, io
from ml_api.ml import MockBatchMLAPI

app = FastAPI()

@app.post('/upload')
async def upload(data: schemas.Data):
    filename = io.save_inputs(data)
    return {"filename": filename}

def batch_predict(filename: str):
    """batch predict method for background process"""
    ml = MockMLAPI()
    ml.load()
    data = io.load_inputs(filename)
    pred = ml.predict(data)
    io.save_outputs(pred, filename)
    print('finished prediction')

@app.get('/prediction/batch')
async def batch_prediction(filename: str, background_tasks: BackgroundTasks):
    if io.check_outputs(filename):
        raise HTTPException(status_code=404, detail="the result of prediction already exists")

    background_tasks.add_task(ml.batch_predict, filename)
    return {}

@app.get('/download', response_model=schemas.Pred)
async def download(filename: str):
    if not io.check_outputs(filename):
        raise HTTPException(status_code=404, detail="the result of prediction does not exist")

    preds = io.load_outputs(filename)
    return {"prediction": preds}

Contrôle de fonctionnement

Vérifiez l'opération de la même manière que la prédiction en ligne. Publiez l'échantillon d'entrée dans CuRL. Ensuite, vous pouvez confirmer que la sortie attendue est renvoyée. Il attend également 30 secondes avant de frapper l'API de téléchargement. Cependant, vous pouvez voir que chaque réponse est renvoyée très rapidement.

$ curl -X POST "http://localhost:8000/upload" -H "accept: application/json" -H "Content-Type: application/json" -d "{\"data\":[{\"text\":\"hogehoge\"},{\"text\":\"fugafuga\"}]}" -w  "\nelapsed time: %{time_starttransfer} s\n"
{"filename":"fdlelteb.csv"}
elapsed time: 0.010242 s

$ curl -X GET "http://localhost:8000/prediction/batch?filename=fdlelteb.csv" -w  "\nelapsed time: %{time_starttransfer} s\n"
{}
elapsed time: 0.007223 s

$ curl -X GET "http://localhost:8000/download?filename=fdlelteb.csv" -w  "\nelapsed time: %{time_starttransfer} s\n"   [12:58:27]
{"prediction":[{"text":"hogehoge","sentiment":"happy"},{"text":"fugafuga","sentiment":"sad"}]}
elapsed time: 0.008825 s

en conclusion

Nous avons introduit deux configurations typiques des API Web d'inférence de machine learning: la prédiction en ligne et la prédiction par lots. Cela nécessite une petite torsion de la configuration générale de l'API Web, mais j'ai également présenté un exemple de mise en œuvre qui se construit simplement à l'aide de l'API Fast. Ce serait formidable si vous pouviez sentir que l'obstacle pour en faire une API Web est faible si vous faites fonctionner correctement le pipeline d'inférence localement. L'excitation de l'apprentissage automatique est sans fin, mais je pense qu'il y a encore peu d'informations telles que la configuration de l'API web ~~ (cela semble assez probable. J'ai ajouté une collection de liens). Je pense que la configuration présentée dans cet article est également approximative. J'apprécierais si vous pouviez commenter les améliorations!

Liens connexes

Il s'agit d'un lien qui n'a pas pu être traité dans cet article.

• N choses à garder à l'esprit lors de la mise en production de la plate-forme d'apprentissage automatique pour le raisonnement
• Publiez le modèle de conception du système d'apprentissage automatique. (Blog d'ingénierie Mercari)
• Exemple de configuration du système d'apprentissage automatique (référence)