[프로그래머스 스쿨 AI] Weak 4 딥러닝 모델 서빙 및 flask 로 보내기

1. 딥러닝 모델 만들어 보기

1. 코드

import torch
import itertools
from utils import clean_text


class ModelHandler:
    def __init__(self):
        self.id2label = {0: 'negative', 1: 'positive'}

    def _clean_text(self, text):
        model_input = []
        if isinstance(text, str):
            cleaned_text = clean_text(text)
            model_input.append(cleaned_text)
        elif isinstance(text, (list, tuple)) and len(text) > 0 and (all(isinstance(t, str) for t in text)):
            cleaned_text = itertools.chain((clean_text(t) for t in text))
            model_input.extend(cleaned_text)
        else:
            model_input.append('')
        return model_input

class DLModelHandler(ModelHandler):
    def __init__(self):
        super().__init__()
        self.initialize()

    def initialize(self, ):
        from transformers import AutoTokenizer, AutoModelForSequenceClassification
        self.model__name_or_path = 'sackoh/bert-base-multilingual-cased-nsmc'
        self.tokenizer = AutoTokenizer.from_pretrained(self.model__name_or_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(self.model__name_or_path)
        self.model.to('cpu')
        pass

    def preprocess(self, data):
        model_input = self._clean_text(data)
        model_input = self.tokenizer(data, return_tensors='pt', padding=True)
        return model_input

    def inference(self, model_input):
        with torch.no_grad():
            model_output = self.model(**model_input)[0].cpu()
            model_output = 1.0 / (1.0 + torch.exp(-model_output))
            model_output = model_output.numpy().astype('float')
        return model_output
        
    def postprocess(self, model_output):
        # process predictions to predicted label and output format
        predicted_probabilities = model_output.max(axis=1)
        predicted_ids = model_output.argmax(axis=1)
        predicted_lables = [self.id2label[id_] for id_ in predicted_ids]
        return predicted_lables, predicted_probabilities

    def handle(self, data):
        # 코드는 topknell.tistory.com/47 
        model_input = self.preprocess(data)
        model_output = self.inference(model_input)
        return self.postprocess(model_output)

 

알고리즘에 따른 실행 속도에 대하여 보여줬다

이건 테스트 케이스 코드다

#file# text_model_handler.py

import unittest
from model import MLModelHandler, DLModelHandler

text = ['음악이 주가 된, 최고의 음악영화',
        '발연기 도저히 못보겠다 진짜 이렇게 연기를 못할거라곤 상상도 못했네',
        '실화라서더욱아름답고찡하네요...많이울었어요벌써4년이란시간이흘렀네요']


class TestModelHandler(unittest.TestCase):
    def test_ml_model_handler(self):
        predicted = ['positive', 'negative', 'negative']
        ml_handler = MLModelHandler()
        results = ml_handler.handle(text)
        for label, pred in zip(results[0], predicted):
            self.assertEqual(label, pred)

    def test_dl_model_handler(self):
        predicted = ['positive', 'negative', 'positive']
        dl_handler = DLModelHandler()
        results = dl_handler.handle(text)
        for label, pred in zip(results[0], predicted):
            self.assertEqual(label, pred)


if __name__ == '__main__':
    unittest.main()

2.  모델 실행해 보기

#터미널
conda activate pytorch_p36
python -m unittest -v test_model_handler.py

 

해보면된다

2. Flask 보내기

1. 코드 만들기

from flask import Flask, request, json
from model import MLModelHandler, DLModelHandler

app = Flask(__name__)

# assign model handler as global variable [2 LINES]
ml_handler = MLModelHandler()
dl_handler = DLModelHandler()

@app.route("/predict", methods=["POST"])
def predict():
    # handle request and body
    body = request.get_json()
    text = body.get('text', '')
    text = [text] if isinstance(text, str) else text
    do_fast = body.get('do_fast', True)

    # model inference [2 LINES]
    if do_fast:
        predictions = ml_handler.handle(text)
    else:
        predictions = dl_handler.handle(text)

    # response
    result = json.dumps({str(i): {'text': t, 'label': l, 'confidence': c}
                         for i, (t, l, c) in enumerate(zip(text, predictions[0], predictions[1]))})
    return result


if __name__ == "__main__":
    app.run(host='0.0.0.0', port='5000', debug=True)

 

2. 실행해 보기

#aws code 상 터미널 
flask run

# 컴퓨터 터널 (안됨)
curl -X POST \
-H "Content-Type: application/json; charset=utf-8 \
-d '{"text" : ["영화 오랜만에 봤는데 괜찮은 영화였어", "정말지루한영화였어"], "use_fast": false}'\
http://13.125.81.163:5000/predict



# 파이썬 3
import requests
url = "http://13.125.81.163:5000/predict"
data = {"text":["영화 오랜만에 봤는데 정말 재미있었어", "영화가 정말 지루했었어"], "use_fast" : False}
reponse = requests.post(url, json=data)
print(reponse)

 

3.영상 기록