app.py

import streamlit as st
import pandas as pd
import streamlit as st
import pickle
import time
from typing import Tuple
from sklearn.feature_extraction.text import TfidfVectorizer
import transformers
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import f1_score
import torch
from transformers import AutoTokenizer, AutoModel
from torch.utils.data import TensorDataset, DataLoader
from sklearn.preprocessing import LabelEncoder
import re
import string
import numpy as np
import torch.nn as nn
import json
import gensim
import torch.nn.functional as F
from transformers import GPT2LMHeadModel, GPT2Tokenizer
from transformers import AutoModelForSequenceClassification



st.title('10-я неделя DS. Классификация отзывов, определение токсичности и генерация текста')

st.sidebar.header('Выберите страницу')
page = st.sidebar.radio("Выберите страницу", ["Вводная информация", "Классификация отзывов", "Зоопарк моделей и F1-score", "Определение токсичности", "Генерация текста"])

if page == "Вводная информация":
        
        st.subheader('*Задача №1*: Классификация отзывов на медицинские учреждения')
        st.write('Задача в двух словах: необходимо дать классификацию отзыва тремя моделями, время, за которое происходит классификаци отзыва, а также таблицу сравнения моделей по F-1 macro для моделей')

        st.subheader('*Задача №2*: Определение токсичности')
        st.write('Задача в двух словах: Оценка степени токсичности пользовательского сообщения ')

        st.subheader('*Задача №3*: Генерация текста')
        st.write('Задача в двух словах: Генерация текста GPT-моделью по пользовательскому prompt')

        st.subheader('☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️☀️')

        st.subheader('Выполнила команда "BERT": Алексей А., Светлана, Алиса')


if page == "Классификация отзывов":
    # Загрузка tf-idf модели и векторайзера
    with open('tf-idf/tf-idf.pkl', 'rb') as f:
        model_tf = pickle.load(f)

    with open('tf-idf/tf-idf_vectorizer.pkl', 'rb') as f:
        vectorizer_tf = pickle.load(f)

    # Загрузка словаря vocab_to_int и Word2Vec модели
    with open('lstm/vocab_to_int.json', 'r') as f:
        vocab_to_int = json.load(f)

    word2vec_model = gensim.models.Word2Vec.load("lstm/word2vec.model")

    stop_words = ['и', 'в', 'во', 'не', 'что', 'он', 'на', 'я', 'с', 'со', 'как', 'а', 'то', 'все', 'она', 'так', 'его', 'но', 'да', 'ты', 'к', 'у', 'же', 'вы', 'за', 'бы', 'по', 'только', 'ее', 'мне', 'было', 'вот', 'от', 'меня', 'еще', 'нет', 'о', 'из', 'ему', 'теперь', 'когда', 'даже', 'ну', 'вдруг', 'ли', 'если', 'уже', 'или', 'ни', 'быть', 'был', 'него', 'до', 'вас', 'нибудь', 'опять', 'уж', 'вам', 'ведь', 'там', 'потом', 'себя', 'ничего', 'ей', 'может', 'они', 'тут', 'где', 'есть', 'надо', 'ней', 'для', 'мы', 'тебя', 'их', 'чем', 'была', 'сам', 'чтоб', 'без', 'будто', 'чего', 'раз', 'тоже', 'себе', 'под', 'будет', 'ж', 'тогда', 'кто', 'этот', 'того', 'потому', 'этого', 'какой', 'совсем', 'ним', 'здесь', 'этом', 'один', 'почти', 'мой', 'тем', 'чтобы', 'нее', 'сейчас', 'были', 'куда', 'зачем', 'всех', 'никогда', 'можно', 'при', 'наконец', 'два', 'об', 'другой', 'хоть', 'после', 'над', 'больше', 'тот', 'через', 'эти', 'нас', 'про', 'всего', 'них', 'какая', 'много', 'разве', 'три', 'эту', 'моя', 'впрочем', 'хорошо', 'свою', 'этой', 'перед', 'иногда', 'лучше', 'чуть', 'том', 'нельзя', 'такой', 'им', 'более', 'всегда', 'конечно', 'всю', 'между']
    
    def data_preprocessing(text: str) -> str:
        text = text.lower()
        text = re.sub('<.*?>', '', text) # html tags
        text = ''.join([c for c in text if c not in string.punctuation])# Remove punctuation
        text = ' '.join([word for word in text.split() if word not in stop_words])
        text = [word for word in text.split() if not word.isdigit()]
        text = ' '.join(text)
        return text

    # Функция для предсказания класса отзыва
    def classify_review_tf(review):
        # Векторизация отзыва
        review_vector = vectorizer_tf.transform([review])
        # Предсказание
        start_time = time.time()
        prediction = model_tf.predict(review_vector)
        end_time = time.time()
        # Время предсказания
        prediction_time = end_time - start_time
        return prediction[0], prediction_time
    
    VOCAB_SIZE = len(vocab_to_int) + 1  # add 1 for the padding token
    EMBEDDING_DIM = 32
    HIDDEN_SIZE = 32
    SEQ_LEN = 100

    class BahdanauAttention(nn.Module):
        def __init__(self, hidden_size: torch.Tensor = HIDDEN_SIZE) -> None:
            super().__init__()

            self.W_q = nn.Linear(hidden_size, hidden_size)
            self.W_k = nn.Linear(hidden_size, hidden_size)
            self.V = nn.Linear(HIDDEN_SIZE, 1)

        def forward(
            self,
            keys: torch.Tensor,
            query: torch.Tensor
        ) -> Tuple[torch.Tensor, torch.Tensor]:
            query = self.W_q(query)
            keys = self.W_k(keys)

            energy = self.V(torch.tanh(query.unsqueeze(1) + keys)).squeeze(-1)
            weights = F.softmax(energy, -1)
            context = torch.bmm(weights.unsqueeze(1), keys)
            return context, weights
    
    embedding_matrix = np.zeros((VOCAB_SIZE, EMBEDDING_DIM))
    embedding_layer = torch.nn.Embedding.from_pretrained(torch.FloatTensor(embedding_matrix))

    class LSTMConcatAttention(nn.Module):
        def __init__(self) -> None:
            super().__init__()

            # self.embedding = nn.Embedding(VOCAB_SIZE, EMBEDDING_DIM)
            self.embedding = embedding_layer
            self.lstm = nn.LSTM(EMBEDDING_DIM, HIDDEN_SIZE, batch_first=True)
            self.attn = BahdanauAttention(HIDDEN_SIZE)
            self.clf = nn.Sequential(
                nn.Linear(HIDDEN_SIZE, 128),
                nn.Dropout(),
                nn.Tanh(),
                nn.Linear(128, 1)
            )
        
        def forward(self, x):
            embeddings = self.embedding(x)
            outputs, (h_n, _) = self.lstm(embeddings)
            att_hidden, att_weights = self.attn(outputs, h_n.squeeze(0))
            out = self.clf(att_hidden)
            return out, att_weights
        
    model_lstm = LSTMConcatAttention()  # Инициализируйте с теми же параметрами, что использовались при обучении
    model_lstm.load_state_dict(torch.load("lstm/lstm_model.pth"))
    model_lstm.eval()

        # Проверка и добавление токена <UNK>, если он отсутствует
    if '<UNK>' not in vocab_to_int:
        vocab_to_int['<UNK>'] = len(vocab_to_int)  # Присвоение нового уникального индекса

    # Проверка и добавление токена <PAD>, если он отсутствует
    if '<PAD>' not in vocab_to_int:
        vocab_to_int['<PAD>'] = len(vocab_to_int)  # Присвоение нового уникального индекса

    def text_to_vector(text, unknown_token_id=0):
            words = text.split()
            vector = [vocab_to_int.get(word, unknown_token_id) for word in words]  # здесь unknown_token_id - это ID для "неизвестных" слов
            return np.array(vector, dtype=np.int64)  # Убедитесь, что тип данных int64


    def classify_review_lstm(review: str, SEQ_LEN: int, model: nn.Module, threshold: float = 0.5):
        """Predict sentiment class for a review

        Args:
            review (str): Review text
            SEQ_LEN (int): sequence length
            model (nn.Module): trained model
            threshold (float): threshold for class prediction

        Returns:
            str: Predicted sentiment ('positive' or 'negative')
        """
        inp = text_to_vector(review)
        inp_tensor = torch.tensor(inp, dtype=torch.int64)
        start_time = time.time()
        with torch.inference_mode():
            pred, _ = model(inp_tensor.long().unsqueeze(0))
        end_time = time.time()
        prediction_time = end_time - start_time
        # Convert prediction to sentiment label
        sentiment = 'positive' if pred.sigmoid().item() > threshold else 'negative'

        return sentiment, prediction_time
    
    tokenizer_rubert = AutoTokenizer.from_pretrained("cointegrated/rubert-tiny2")
    model_rubert = AutoModel.from_pretrained("cointegrated/rubert-tiny2")
    clf_rubert = LogisticRegression(max_iter=1000)  # Предполагается, что ваша модель уже обучена

    with open('rubert/logistic_regression_model.pkl', 'rb') as f:
        clf_rubert = pickle.load(f)

    # Функция для предсказания
    def make_prediction(text):
        start_time = time.time()
        encoded = tokenizer_rubert(text, add_special_tokens=True, max_length=128, padding='max_length', truncation=True, return_tensors="pt")
        with torch.no_grad():
            outputs = model_rubert(**encoded)
        features = outputs.last_hidden_state[:, 0, :].numpy()
        prediction = clf_rubert.predict(features)
        end_time = time.time()
        prediction_time = end_time - start_time
        return prediction[0], prediction_time

    # Создание интерфейса Streamlit
    st.title('Классификатор отзывов на клиники')

    # Текстовое поле для ввода отзыва
    user_review = st.text_input('Введите ваш отзыв на клинику')

    if st.button('Классифицировать'):
        if user_review:
            # Классификация отзыва
            prediction_tf, pred_time_tf = classify_review_tf(user_review)
            st.write(f'Предсказанный класс TF-IDF: {prediction_tf}')
            st.write(f'Время предсказания TF-IDF: {pred_time_tf:.4f} секунд')
            prediction_lstm, pred_time_lstm = classify_review_lstm(user_review, SEQ_LEN=SEQ_LEN, model=model_lstm)
            st.write(f'Предсказанный класс LSTM: {prediction_lstm}')
            st.write(f'Время предсказания LSTM: {pred_time_lstm:.4f} секунд')
            prediction_rubert, pred_time_rubert = make_prediction(user_review)
            prediction_ru = 'negative' if prediction_rubert == 0 else 'positive'
            st.write(f'Предсказанный класс RuBERT: {prediction_ru}')
            st.write(f'Время предсказания RuBERT: {pred_time_rubert:.4f} секунд')
        else:
            st.write('Пожалуйста, введите отзыв')

if page == "Зоопарк моделей и F1-score":
# Создание данных для таблицы
    data = {
        "Название модели": ["TF-IDF", "LSTM", "RuBert tiny-2"],
        "F-1 macro score": ["0,94", "0,89", "0,90"]
    }

    # Создание DataFrame
    df = pd.DataFrame(data)

    # Отображение таблицы в Streamlit
    st.table(df)


if page == "Определение токсичности":

    # Функция для загрузки обученной модели
    def load_model(model_path):
        with open(model_path, 'rb') as file:
            model = pickle.load(file)
        return model

    # Загрузка обученной модели
    clf_c = load_model('toxic/logistic_regression_model_toxic.pkl')  # Укажите путь к файлу модели

    # Загрузка токенизатора и модели BERT
    tokenizer_c = AutoTokenizer.from_pretrained("cointegrated/rubert-tiny-toxicity")
    model_c = AutoModel.from_pretrained("cointegrated/rubert-tiny-toxicity")

    # Функция для предсказания токсичности сообщения
    def predict_toxicity(text):
        encoded = tokenizer_c(text, return_tensors="pt", padding=True, truncation=True, max_length=512)
        with torch.no_grad():
            outputs = model_c(**encoded)
        features = outputs.last_hidden_state[:, 0, :].numpy()
        prediction = clf_c.predict_proba(features)
        return prediction[0]

    model_checkpoint = 'cointegrated/rubert-tiny-toxicity'
    tokenizer_b = AutoTokenizer.from_pretrained(model_checkpoint)
    model_b = AutoModelForSequenceClassification.from_pretrained(model_checkpoint)

    def text2toxicity(text):
        with torch.no_grad():
            inputs = tokenizer_b(text, return_tensors='pt', truncation=True, padding=True)
            proba = torch.sigmoid(model_b(**inputs).logits).cpu().numpy()
            return proba[0][1]

    # Создание интерфейса Streamlit
    st.title("Оценка токсичности сообщения")

    # Текстовое поле для ввода сообщения
    user_input = st.text_area("Введите сообщение для оценки")

    if st.button("Оценить токсичность сообщения кастомизированной моделью"):
        if user_input:
            # Оценка токсичности сообщения
            prediction = predict_toxicity(user_input)[1]
            st.write(f'Вероятность токсичности согласно кастомизированной модели: {prediction:.4f}')
        else:
            st.write("Пожалуйста, введите сообщение")
    
    if st.button('Определить токсичность базовой моделью'):
        if user_input:
            # Определение токсичности сообщения
            proba_toxicity = text2toxicity(user_input)
            st.write(f'Вероятность токсичности rubert-tiny-toxicity.pretrained: {proba_toxicity:.4f}')
        else:
            st.write('Пожалуйста, введите сообщение')

    
    
if page == "Генерация текста":

    # Путь к вашим весам модели
    model_weights_path = 'gpt-2/model.pt'

    # Загружаем токенизатор от GPT-2
    tokenizer = GPT2Tokenizer.from_pretrained("sberbank-ai/rugpt3small_based_on_gpt2")

    # Создаем экземпляр модели с архитектурой GPT-2, но без предварительно обученных весов
    model = GPT2LMHeadModel.from_pretrained('sberbank-ai/rugpt3small_based_on_gpt2')

    # Загружаем веса вашей модели
    model.load_state_dict(torch.load(model_weights_path, map_location='cpu'))

    # Переносим модель на устройство (GPU или CPU)
    device = 'cpu'
    model.to(device)
    model.eval()

    def main():
        st.title("Генератор плохих отзывов больниц от ruGPT3")

        # Ввод текста от пользователя
        user_prompt = st.text_area("Введите текст-промпт:", "Я была в этой клинике")

        # Виджеты для динамической регуляции параметров
        max_length = st.slider("Выберите max_length:", 10, 300, 100)
        temperature = st.slider("Выберите temperature:", 1.0, 10.0, step=0.2)
        top_k = st.slider("Выберите top_k:", 100, 500, 50)
        top_p = st.slider("Выберите top_p:", 0.1, 1.0, 0.95, step=0.05)
        num_beams = st.slider('Выберите num_beams:', 5, 40, step=1)

        # Генерация текста при нажатии на кнопку
        if st.button("Сгенерировать текст"):
            with torch.no_grad():
                prompt = tokenizer.encode(user_prompt, return_tensors='pt').to(device)
                out = model.generate(
                    input_ids=prompt,
                    max_length=max_length,
                    num_beams=num_beams,
                    temperature=temperature,
                    top_k=top_k,
                    top_p=top_p,
                    no_repeat_ngram_size=2,
                ).cpu().numpy()
                generated_text = tokenizer.decode(out[0], skip_special_tokens=True)
                st.subheader("Сгенерированный текст:")
                st.write(generated_text)

    if __name__ == "__main__":
        main()