Spaces:
Running
Running
File size: 4,107 Bytes
82c750f 25eda0a 82c750f 82becce 82c750f 82becce |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
# Built-in
import re
import joblib
from pathlib import Path
import os
import uvicorn
# Dependencies for FastAPI
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import keras
import tensorflow as tf
# Set Environment
os.environ["KERAS_BACKEND"] = "tensorflow"
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '1'
# Setup Paths
lr_model_path = Path('./prod_models/emotion_classifier_pipe_lr.pkl')
keras_model_path = Path('./prod_models/emo_modelV2_tf')
# Class for Text Body
class Paragraph(BaseModel):
input: str
# Classes
classes = ['admiration', 'amusement', 'anger', 'annoyance', 'approval', 'caring', 'confusion', 'curiosity', 'desire', 'disappointment', 'disapproval', 'disgust', 'embarrassment', 'excitement', 'fear', 'gratitude', 'grief', 'joy', 'love', 'nervousness', 'optimism', 'pride', 'realization', 'relief', 'remorse', 'sadness', 'surprise', 'neutral']
# Load the Logistic Regression Model
with open(lr_model_path, 'rb') as f:
lr_model = joblib.load(f)
# Load the Keras Model
tfsmlayer = keras.layers.TFSMLayer(str(keras_model_path), call_endpoint="serving_default")
inputs = keras.Input(shape=(1,), dtype=tf.string)
outputs = tfsmlayer(inputs)
keras_model = keras.Model(inputs, outputs)
# Start the app
app = FastAPI()
# Setup CORS policy
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# APIs
@app.get("/")
async def docs():
return RedirectResponse(url="/docs")
@app.post("/logistic-regression")
async def predict_emotions_lr(paragraph : Paragraph):
# Split the huge chunk of text into a list of strings
text_list = [text.strip() for text in re.split(r'[.!?;\n]', paragraph.input) if text.strip()]
# Create a list to store predictions per text
predictions_per_text = []
for text in text_list:
emotion = [{'label': label, 'score': score} for label, score in zip(lr_model.classes_, lr_model.predict_proba([text])[0])]
predictions_per_text.append(emotion)
# Create a dictionary to aggregate scores for each label
total = {}
# Iterate over each list and aggregate the scores
for prediction in predictions_per_text:
for emotion_dict in prediction:
label = emotion_dict['label']
score = emotion_dict['score']
total[label] = total.get(label, 0) + score
# Convert the dictionary to a list of dictionaries
result = [{"label": label, "score": score} for label, score in total.items()]
# Sort the result in descending order based on score
sorted_result = sorted(result, key=lambda x: x['score'], reverse=True)
return {"predictions": sorted_result}
@app.post("/keras")
async def predict_emotions_keras(paragraph : Paragraph):
# Split the huge chunk of text into a list of strings
text_list = [text.strip() for text in re.split(r'[.!?;\n]', paragraph.input) if text.strip()]
# Create a list to store predictions per text
predictions_per_text = []
for text in text_list:
scores = keras_model(tf.constant([text]))['dense_1'][0]
emotion = [{'label': label, 'score': score} for label, score in zip(classes, scores.numpy())]
predictions_per_text.append(emotion)
# Create a dictionary to aggregate scores for each label
total = {}
# Iterate over each list and aggregate the scores
for prediction in predictions_per_text:
for emotion_dict in prediction:
label = emotion_dict['label']
score = emotion_dict['score']
total[label] = total.get(label, 0) + score
# Convert the dictionary to a list of dictionaries
result = [{"label": label, "score": score} for label, score in total.items()]
# Sort the result in descending order based on score
sorted_result = sorted(result, key=lambda x: x['score'], reverse=True)
return {"predictions": sorted_result}
# if __name__ == "__main__":
# uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True) |