<repo_name>monisoi/nlp_api_for_aituber<file_sep>/README.md
# NLP API for AItuber
AItuber向けの様々な自然言語処理APIを追加していく予定です。

## 動作確認環境
- keras v2.3.0
- tensorflow v1.14.0

## /analyze_sentiment
感情分析です。  
以下の手法を使わせていただいています。  
[Twitter Sentiment Analysis](https://www.kaggle.com/paoloripamonti/twitter-sentiment-analysis)  
学習させたファイルを `/models` に格納してください。<file_sep>/conversation_word_level.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import numpy as np
import re
import pickle
from tensorflow.keras.models import Model, load_model
from tensorflow.keras.layers import Input
from tensorflow.keras.preprocessing.sequence import pad_sequences


SEQUENCE_LENGTH = 30
MAX_OUTPUT_LENGTH = 100


def clean_text(text):
    text = text.lower()
    text = re.sub(r"i'm", "i am", text)
    text = re.sub(r"he's", "he is", text)
    text = re.sub(r"she's", "she is", text)
    text = re.sub(r"it's", "it is", text)
    text = re.sub(r"that's", "that is", text)
    text = re.sub(r"what's", "that is", text)
    text = re.sub(r"where's", "where is", text)
    text = re.sub(r"how's", "how is", text)
    text = re.sub(r"\'ll", " will", text)
    text = re.sub(r"\'ve", " have", text)
    text = re.sub(r"\'re", " are", text)
    text = re.sub(r"\'d", " would", text)
    text = re.sub(r"\'re", " are", text)
    text = re.sub(r"won't", "will not", text)
    text = re.sub(r"can't", "cannot", text)
    text = re.sub(r"n't", " not", text)
    text = re.sub(r"n'", "ng", text)
    text = re.sub(r"'bout", "about", text)
    text = re.sub(r"'til", "until", text)
    text = re.sub(r"[-()\"'#/@;:<>{}`+=~|.!?,]", "", text)
    return text


class Conversation():
    def __init__(self, tokenizer, model):
        self.tokenizer, self.word2index, self.index2word = self.load_tokenizer(tokenizer)
        self.model = load_model(model)
        self.encoder_model, self.decoder_model = self.create_models()

    def load_tokenizer(self, file_name):
        with open(file_name, 'rb') as handle:
            tokenizer = pickle.load(handle)
            word2index = tokenizer.word_index
            index2word = dict(map(reversed, word2index.items()))
            return tokenizer, word2index, index2word

    def create_models(self):
        HIDDEN_DIM = 256

        # encoder
        encoder_inputs = self.model.input[0]
        _, *encoder_states = self.model.layers[4].output
        encoder_model = Model(encoder_inputs, encoder_states)

        # decoder
        decoder_inputs = Input(shape=(1,))
        decoder_embedding_layer = self.model.layers[3]
        decoder_embedded = decoder_embedding_layer(decoder_inputs)
        decoder_lstm_layer = self.model.layers[5]
        decoder_states_inputs = [Input(shape=(HIDDEN_DIM,)), Input(shape=(HIDDEN_DIM,))]
        decoder_lstm, *decoder_states = decoder_lstm_layer(
            decoder_embedded,
            initial_state=decoder_states_inputs)
        decoder_dense_layer = self.model.layers[6]
        decoder_outputs = decoder_dense_layer(decoder_lstm)
        decoder_model = Model(
            [decoder_inputs] + decoder_states_inputs,
            [decoder_outputs] + decoder_states)

        return encoder_model, decoder_model

    def decode_sequence(self, input_seq):
        formated_input_seq = '<s> {} </s>'.format(clean_text(input_seq))
        tokenized_input_seq = pad_sequences(
            self.tokenizer.texts_to_sequences([formated_input_seq]),
            padding='post',
            maxlen=SEQUENCE_LENGTH)
        bos = [self.word2index['<s>']]
        eos = [self.word2index['</s>']]
        states = self.encoder_model.predict(tokenized_input_seq)
        target = np.array(bos)
        output_seq = [bos[0]]
        for i in range(MAX_OUTPUT_LENGTH):
            tokens, *states = self.decoder_model.predict([target] + states)
            output_index = [np.argmax(tokens[0, -1, :])]
            output_seq += output_index
            if output_index == eos:
                break
            target = np.array(output_index)
        output_seq = ' '.join([self.index2word[i] for i in output_seq if i not in bos + eos])
        return output_seq

    def reply(self, input_seq):
        output_seq = self.decode_sequence(input_seq)
        response = {
            "input_seq": input_seq,
            "output_seq": output_seq,
        }
        return response
<file_sep>/conversation_char_level.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pandas as pd
import numpy as np
from keras.layers import Dense, LSTM, Input
from keras.models import Model
from keras.models import load_model


latent_dim = 256
num_encoder_tokens = 37
num_decoder_tokens = 39
max_decoder_seq_length = 301

input_token_index = {' ': 0, '0': 1, '1': 2, '2': 3, '3': 4, '4': 5, '5': 6, '6': 7, '7': 8,
                     '8': 9, '9': 10, 'a': 11, 'b': 12, 'c': 13, 'd': 14, 'e': 15, 'f': 16,
                     'g': 17, 'h': 18, 'i': 19, 'j': 20, 'k': 21, 'l': 22, 'm': 23, 'n': 24,
                     'o': 25, 'p': 26, 'q': 27, 'r': 28, 's': 29, 't': 30, 'u': 31, 'v': 32,
                     'w': 33, 'x': 34, 'y': 35, 'z': 36}
target_token_index = {'\t': 0, '\n': 1, ' ': 2, '0': 3, '1': 4, '2': 5, '3': 6, '4': 7, '5': 8,
                      '6': 9, '7': 10, '8': 11, '9': 12, 'a': 13, 'b': 14, 'c': 15, 'd': 16,
                      'e': 17, 'f': 18, 'g': 19, 'h': 20, 'i': 21, 'j': 22, 'k': 23, 'l': 24,
                      'm': 25, 'n': 26, 'o': 27, 'p': 28, 'q': 29, 'r': 30, 's': 31, 't': 32,
                      'u': 33, 'v': 34, 'w': 35, 'x': 36, 'y': 37, 'z': 38}
reverse_input_char_index = dict(
    (i, char) for char, i in input_token_index.items())
reverse_target_char_index = dict(
    (i, char) for char, i in target_token_index.items())


class Conversation():
    def __init__(self, model):
        self.model = load_model(model)
        self.encoder_model = self.create_encoder()
        self.decoder_model = self.create_decoder()

    def create_encoder(self):
        encoder_inputs = self.model.input[0]  # input_1
        encoder_outputs, state_h, state_c = self.model.layers[2].output
        encoder_states = [state_h, state_c]
        return Model(encoder_inputs, encoder_states)

    def create_decoder(self):
        decoder_inputs = self.model.input[1]  # input_2
        decoder_state_input_h = Input(shape=(latent_dim,), name='input_3')
        decoder_state_input_c = Input(shape=(latent_dim,), name='input_4')
        decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c]
        decoder_lstm = self.model.layers[3]
        decoder_outputs, state_h, state_c = decoder_lstm(
            decoder_inputs, initial_state=decoder_states_inputs)
        decoder_states = [state_h, state_c]
        decoder_dense = self.model.layers[4]
        decoder_outputs = decoder_dense(decoder_outputs)
        return Model(
            [decoder_inputs] + decoder_states_inputs,
            [decoder_outputs] + decoder_states)

    def decode_sequence(self, input_seq):
        states_value = self.encoder_model.predict(input_seq)
        target_seq = np.zeros((1, 1, num_decoder_tokens))
        target_seq[0, 0, target_token_index['\t']] = 1.
        stop_condition = False
        decoded_sentence = ''
        while not stop_condition:
            output_tokens, h, c = self.decoder_model.predict(
                [target_seq] + states_value)
            sampled_token_index = np.argmax(output_tokens[0, -1, :])
            sampled_char = reverse_target_char_index[sampled_token_index]
            decoded_sentence += sampled_char
            if (sampled_char == '\n' or
                    len(decoded_sentence) > max_decoder_seq_length):
                stop_condition = True
            target_seq = np.zeros((1, 1, num_decoder_tokens))
            target_seq[0, 0, sampled_token_index] = 1.
            states_value = [h, c]
        return decoded_sentence

    def reply(self, input_seq):
        onehot_input_seq = np.zeros(
            (1, len(input_seq), num_encoder_tokens), dtype='float32')
        for i, char in enumerate(input_seq):
            onehot_input_seq[0, i, input_token_index[char]] = 1.
        response = {
            "input_seq": input_seq,
            "output_seq": self.decode_sequence(onehot_input_seq),
        }
        return response
<file_sep>/main.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

from flask import Flask, request, jsonify
from sentiment_analysis import SentimentAnalysis
# from conversation_char_level import Conversation
from conversation_word_level import Conversation

SA_TOKENIZER = './models/sa_tokenizer.pkl'
SA_MODEL = './models/sa_model.h5'
CONV_TOKENIZER = './models/conv_tokenizer.pkl'
CONV_MODEL = './models/conv_model.h5'

app = Flask(__name__)


@app.before_first_request
def _load_model():
    global sentiment_analysis
    global conversation

    sentiment_analysis = SentimentAnalysis(SA_TOKENIZER, SA_MODEL)
    conversation = Conversation(CONV_TOKENIZER, CONV_MODEL)


@app.route('/')
def hello():
    """Return a friendly HTTP greeting."""
    return 'Hello World!'


@app.route('/analyze_sentiment')
def analyze_sentiment():
    if not sentiment_analysis:
        _load_model()
        if not sentiment_analysis:
            return 'Sentiment Analysis Model not found.'

    text = request.args.get('text')

    result = sentiment_analysis.predict(text)
    # result example
    # {'label': 'NEGATIVE',
    # 'score': 0.010753681883215904,
    # 'elapsed_time': 0.26644086837768555}
    return jsonify({
        'status': 'OK',
        'result': result
    })


@app.route('/talk')
def talk():
    if not conversation:
        _load_model()
        if not conversation:
            return 'Conversation Model not found.'

    text = request.args.get('text')

    result = conversation.reply(text)
    # result example
    # {'input_seq': 'how are you',
    # 'output_seq': 'i am fine'}
    return jsonify({
        'status': 'OK',
        'result': result
    })


if __name__ == '__main__':
    # to solve a problem. see this issue.
    # https://github.com/keras-team/keras/issues/13353
    app.run(host='0.0.0.0', port=5000, threaded=False)
<file_sep>/sentiment_analysis.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.models import load_model
import time
import pickle

SEQUENCE_LENGTH = 300
POSITIVE = "POSITIVE"
NEGATIVE = "NEGATIVE"
NEUTRAL = "NEUTRAL"
SENTIMENT_THRESHOLDS = (0.4, 0.7)


class SentimentAnalysis():
    def __init__(self, tokenizer, model):
        with open(tokenizer, 'rb') as handle:
            self.tokenizer = pickle.load(handle)
        self.model = load_model(model)

    def decode_sentiment(self, score):
        label = NEUTRAL
        if score <= SENTIMENT_THRESHOLDS[0]:
            label = NEGATIVE
        elif score >= SENTIMENT_THRESHOLDS[1]:
            label = POSITIVE
        return label

    def predict(self, text):
        start_at = time.time()
        x_test = pad_sequences(self.tokenizer.texts_to_sequences(
            [text]), maxlen=SEQUENCE_LENGTH)
        score = self.model.predict([x_test])[0]
        label = self.decode_sentiment(score)
        return {"label": label, "score": float(score),
                "elapsed_time": time.time() - start_at}
