text_plain_output_1.png
from copy import deepcopy

from copy import deepcopy

from datasets import list_metrics,load_metric

from random import randint

from random import randint,shuffle

from sentence_transformers import SentenceTransformer, util

from sklearn.metrics import confusion_matrix

import numpy as np

import pandas as pd

import pandas as pd

import plotly.express as px

def performance(y_ture,y_pred):
    f1_metric = load_metric("f1")
    re_metric = load_metric("recall")
    pre_metric = load_metric("precision")
    type_c_int = list(set(np.concatenate([y_ture, y_pred])))
    type_c = [str(i) for i in type_c_int]
    
    f1_m_list = []
    re_m_list = []
    pre_m_list = []
    
    for i in type_c_int:
        bi_ture = list(y_ture == i)
        bi_pred = list(y_pred == i)
        f1_m_results = f1_metric.compute(predictions=bi_pred, references=bi_ture, average="macro")
        re_m_results = re_metric.compute(predictions=bi_pred, references=bi_ture, average="macro")
        pre_m_results = pre_metric.compute(predictions=bi_pred, references=bi_ture, average="macro")
        
        f1_m_list.append(f1_m_results["f1"])
        re_m_list.append(re_m_results["recall"])
        pre_m_list.append(pre_m_results["precision"])
        
    data = {'Class_type':type_c_int,'F1-macro':f1_m_list,'Recall-macro':re_m_list,'Precision-macro':pre_m_list}
    df = pd.DataFrame(data)
    display(df)
    
    
    z = confusion_matrix(y_ture, y_pred)
    x_lab = type_c

    fig = px.imshow(z, 
                    text_auto=True,
                    labels=dict(x="True label", y="Predicted label", color="times"),
                    x=x_lab,
                    y=x_lab)
#     fig.show()
    Accuarcy = sum([1 for i in range(len(y_ture)) if y_pred[i] == y_ture[i]])/len(y_ture)
    print("Accuarcy is", Accuarcy)
    return z,fig

cf_matrix_test, figure_test = performance([1, 3, 1, 2, 2, 1], [2, 3, 1, 3, 3, 2])

def read_and_split_the_excel(QA_path):
    """
    :func: 根据xlsx文件获取问题list和答案list（需要更新openyxl）
    :param path: 文件路径
    :return: 问题list，答案list
    """
    df1 = pd.read_excel(QA_path)
    question_list = df1.iloc[:, 0].tolist()
    answer_list = df1.iloc[:, 1].tolist()
    return (question_list, answer_list)

def read_and_split_the_01(zero_one_path):
    """
    :func: 根据xlsx文件获取原始list和测试list和label
    :param path: 文件路径
    :return: 问题list，答案list
    """
    df1 = pd.read_csv(zero_one_path)
    Sen1_list = df1.iloc[:, 0].tolist()
    Sen2_list = df1.iloc[:, 1].tolist()
    label_list = df1.iloc[:, 2].tolist()
    return (Sen1_list, Sen2_list, label_list)

def shuffle_without_repeated(list_):
    temp_list = deepcopy(list_)
    m = len(temp_list)
    m = m - 1
    for i_current in range(m, 1, -1):
        rest = i_current - 1
        i_replace = randint(0, rest)
        temp_list[i_current], temp_list[i_replace] = (temp_list[i_replace], temp_list[i_current])
    return temp_list
def obtain_shuffle_01(ori_list):
    shuffle_q_list = shuffle_without_repeated(ori_list)
    shuffle_label_list = [0] * len(shuffle_q_list)
    return (ori_list, shuffle_q_list, shuffle_label_list)

question_list = ['The cat sits outside', 'A man is playing guitar', 'The new movie is awesome', 'The new opera is nice']
obtain_shuffle_01(question_list)

def read_qa_and_expand_training_set(QA_path, zero_one_path):
    question_list, answer_list = read_and_split_the_excel(QA_path)
    Sen1_list, Sen2_list, label_list = read_and_split_the_01(zero_one_path)
    return (question_list, answer_list, Sen1_list, Sen2_list, label_list)

QA_path = '../input/uic-cn-admission/CN_QA_dataset_all.xlsx'
zero_one_path = '/kaggle/input/01-uic-rm-dup/df_test.csv'
question_list, answer_list, Sen1_list, Sen1_list_index, Sen2_list, label_list = read_qa_and_expand_training_set(QA_path, zero_one_path)

from sentence_transformers import SentenceTransformer, util
import pandas as pd
from copy import deepcopy
from random import randint
from termcolor import colored
def SBERT_get_reply(model, query, question_list, answer_list, question_list_emb, topk_SBERT, threshold_SBERT):
    queries = [query]
    query_embeddings = model.encode(queries, convert_to_tensor=True)
    index_ranked = []
    tensor_scores = []
    cosine_scores = util.pytorch_cos_sim(query_embeddings, question_list_emb)[0]
    results = zip(range(len(cosine_scores)), cosine_scores)
    results = sorted(results, key=lambda x: x[1], reverse=True)
    for index, tensor_score in results:
        index_ranked.append(question_list[index])
        tensor_scores.append(tensor_score)
    topk_idx_SBERT = index_ranked[:topk_SBERT]
    return (topk_idx_SBERT, tensor_scores)
def use_model_qa(model_path, QA_path, zero_one_path):
    model = SentenceTransformer(model_path, device='cuda')
    topk_SBERT = 3
    threshold_SBERT = 0.6
    question_list, answer_list = read_and_split_the_excel(QA_path)
    question_embeddings = model.encode(question_list, convert_to_tensor=True)
    question_list, answer_list, Sen1_list, Sen2_list, label_list = read_qa_and_expand_training_set(QA_path, zero_one_path)
    predict_result = []
    df = pd.read_csv(zero_one_path)
    match = []
    score = []
    for index, test_query in enumerate(Sen2_list):
        topk_idx_SBERT, tensor_scores = SBERT_get_reply(model, test_query, question_list, answer_list, question_embeddings, topk_SBERT, threshold_SBERT)
        match.append(topk_idx_SBERT[0])
        score.append(tensor_scores[0].cpu().numpy())
        if topk_idx_SBERT[0] == Sen1_list[index]:
            prediction = 1
        else:
            prediction = 0
        predict_result.append(prediction)
    new = pd.DataFrame({'roberta_fine_tune_match': match, 'roberta_fine_tune_score': score, 'roberta_fine_tune_label': predict_result})
    merged_df = pd.concat([df, new], axis=1)
    merged_df.to_excel('UIC问题匹配结果比较roberta.xlsx', index=None)
    cf_matrix_test, figure_test = performance(label_list, predict_result)
    return cf_matrix_test
def SBERT_QA_test(model_path):
    QA_path = '../input/uic-cn-admission/CN_QA_dataset_all.xlsx'
    zero_one_path = '/kaggle/input/01-uic-rm-dup/df_test.csv'

model_path = '/kaggle/input/sbert-fine-tune/roberta_fine_tune'
SBERT_QA_test(model_path)