AzulaFire/Text_Sentiment_Analysis_System

1import paddle2import numpy as np3import random4from paddlenlp.transformers import SkepTokenizer, SkepModel5import gradio as gr6from seqeval.metrics.sequence_labeling import get_entities7label_ext_path = "./data/data121190/label_ext.dict"8label_cls_path = "./data/data121242/label_cls.dict"9ext_model_path = "./best_ext.pdparams"10cls_model_path = "./best_cls.pdparams"11def set_seed(seed):12    paddle.seed(seed)13    random.seed(seed)14    np.random.seed(seed)15def format_print(results):16    for result in results:17        aspect, opinion = result[0], set(result[1:])18        print(f"aspect: {aspect}, opinion: {opinion}\n")19 20def decoding(text, tag_seq):21    assert len(text) == len(tag_seq), f"text len: {len(text)}, tag_seq len: {len(tag_seq)}"22 23    puncs = list(",.?;!，。？；！")24    splits = [idx for idx in range(len(text)) if text[idx] in puncs]25 26    prev = 027    sub_texts, sub_tag_seqs = [], []28    for i, split in enumerate(splits):29        sub_tag_seqs.append(tag_seq[prev:split])30        sub_texts.append(text[prev:split])31        prev = split32    sub_tag_seqs.append(tag_seq[prev:])33    sub_texts.append((text[prev:]))34 35    ents_list = []36    for sub_text, sub_tag_seq in zip(sub_texts, sub_tag_seqs):37        ents = get_entities(sub_tag_seq, suffix=False)38        ents_list.append((sub_text, ents))39 40    aps = []41    no_a_words = []42    for sub_tag_seq, ent_list in ents_list:43        sub_aps = []44        sub_no_a_words = []45        # print(ent_list)46        for ent in ent_list:47            ent_name, start, end = ent48            if ent_name == "Aspect":49                aspect = sub_tag_seq[start:end+1]50                sub_aps.append([aspect])51                if len(sub_no_a_words) > 0:52                    sub_aps[-1].extend(sub_no_a_words)53                    sub_no_a_words.clear()54            else:55                ent_name == "Opinion"56                opinion = sub_tag_seq[start:end + 1]57                if len(sub_aps) > 0:58                    sub_aps[-1].append(opinion)59                else:60                    sub_no_a_words.append(opinion)61 62        if sub_aps:63            aps.extend(sub_aps)64            if len(no_a_words) > 0:65                aps[-1].extend(no_a_words)66                no_a_words.clear()67        elif sub_no_a_words:68            if len(aps) > 0:69                aps[-1].extend(sub_no_a_words)70            else:71                no_a_words.extend(sub_no_a_words)72 73    if no_a_words:74        no_a_words.insert(0, "None")75        aps.append(no_a_words)76 77    return aps 78    79def is_aspect_first(text, aspect, opinion_word):80    return text.find(aspect) <= text.find(opinion_word)81 82def concate_aspect_and_opinion(text, aspect, opinion_words):83    aspect_text = ""84    for opinion_word in opinion_words:85        if is_aspect_first(text, aspect, opinion_word):86            aspect_text += aspect+opinion_word+"，"87        else:88            aspect_text += opinion_word+aspect+"，"89    aspect_text = aspect_text[:-1]90 91    return aspect_text92 93def format_print(results):94    for result in results:95        aspect, opinions, sentiment = result["aspect"], result["opinions"], result["sentiment"]96        print(f"aspect: {aspect}, opinions: {opinions}, sentiment: {sentiment}")97    print()98    return f"aspect: {aspect}, opinions: {opinions}, sentiment: {sentiment}"99 100def is_target_first(text, target, word):101    return text.find(target) <= text.find(word)102 103 104def ext_load_dict(dict_path):105    with open(dict_path, "r", encoding="utf-8") as f:106        words = [word.strip() for word in f.readlines()]107        word2id = dict(zip(words, range(len(words))))108        id2word = dict((v, k) for k, v in word2id.items())109 110        return word2id, id2word111 112 113def cls_load_dict(dict_path):114    with open(dict_path, "r", encoding="utf-8") as f:115        words = [word.strip() for word in f.readlines()]116        word2id = dict(zip(words, range(len(words))))117        id2word = dict((v, k) for k, v in word2id.items())118 119        return word2id, id2word120 121 122def read(data_path):123    with open(data_path, "r", encoding="utf-8") as f:124        for line in f.readlines():125            items = line.strip().split("\t")126            assert len(items) == 3127            example = {"label": int(128                items[0]), "target_text": items[1], "text": items[2]}129 130            yield example131 132 133def convert_example_to_feature(example, tokenizer, label2id,  max_seq_len=512, is_test=False):134    encoded_inputs = tokenizer(135        example["target_text"], text_pair=example["text"], max_seq_len=max_seq_len, return_length=True)136 137    if not is_test:138        label = example["label"]139        return encoded_inputs["input_ids"], encoded_inputs["token_type_ids"], encoded_inputs["seq_len"], label140 141    return encoded_inputs["input_ids"], encoded_inputs["token_type_ids"], encoded_inputs["seq_len"]142class SkepForTokenClassification(paddle.nn.Layer):143    def __init__(self, skep, num_classes=2, dropout=None):144        super(SkepForTokenClassification, self).__init__()145        self.num_classes = num_classes146        self.skep = skep147        self.dropout = paddle.nn.Dropout(148            dropout if dropout is not None else self.skep.config["hidden_dropout_prob"])149        self.classifier = paddle.nn.Linear(150            self.skep.config["hidden_size"], num_classes)151 152    def forward(self, input_ids, token_type_ids=None, position_ids=None, attention_mask=None):153        sequence_output, _ = self.skep(154            input_ids, token_type_ids=token_type_ids, position_ids=position_ids, attention_mask=attention_mask)155 156        sequence_output = self.dropout(sequence_output)157        logits = self.classifier(sequence_output)158        return logits159class SkepForSequenceClassification(paddle.nn.Layer):160    def __init__(self, skep, num_classes=2, dropout=None):161        super(SkepForSequenceClassification, self).__init__()162        self.num_classes = num_classes163        self.skep = skep164        self.dropout = paddle.nn.Dropout(165            dropout if dropout is not None else self.skep.config["hidden_dropout_prob"])166        self.classifier = paddle.nn.Linear(167            self.skep.config["hidden_size"], num_classes)168 169    def forward(self, input_ids, token_type_ids=None, position_ids=None, attention_mask=None):170        _, pooled_output = self.skep(input_ids, token_type_ids=token_type_ids,171                                     position_ids=position_ids, attention_mask=attention_mask)172 173        pooled_output = self.dropout(pooled_output)174        logits = self.classifier(pooled_output)175        return logits176# load dict177model_name = "skep_ernie_1.0_large_ch"178target1_dir = "./skepTokenizer"179target2_dir = "./skepModel"180ext_label2id, ext_id2label = ext_load_dict(label_ext_path)181cls_label2id, cls_id2label = cls_load_dict(label_cls_path)182tokenizer = SkepTokenizer.from_pretrained(target1_dir)183print("label dict loaded.")184 185# load ext model186ext_state_dict = paddle.load(ext_model_path)187ext_skep = SkepModel.from_pretrained(target2_dir)188ext_model = SkepForTokenClassification(ext_skep, num_classes=len(ext_label2id))189ext_model.load_dict(ext_state_dict)190print("extraction model loaded.")191 192# load cls model193cls_state_dict = paddle.load(cls_model_path)194cls_skep = ext_skep195cls_model = SkepForSequenceClassification(196    cls_skep, num_classes=len(cls_label2id))197cls_model.load_dict(cls_state_dict)198print("classification model loaded.")199def predict(input_text):200 201    ext_model.eval()202    cls_model.eval()203 204    # processing input text205    encoded_inputs = tokenizer(list(input_text), is_split_into_words=True, max_seq_len=max_seq_len,)206    input_ids = paddle.to_tensor([encoded_inputs["input_ids"]])207    token_type_ids = paddle.to_tensor([encoded_inputs["token_type_ids"]])208 209    # extract aspect and opinion words210    logits = ext_model(input_ids, token_type_ids=token_type_ids)211    predictions = logits.argmax(axis=2).numpy()[0]212    tag_seq = [ext_id2label[idx] for idx in predictions][1:-1]213    aps = decoding(input_text, tag_seq)214 215    # predict sentiment for aspect with cls_model216    results = []217    for ap in aps:218        aspect = ap[0]219        opinion_words = list(set(ap[1:]))220        aspect_text = concate_aspect_and_opinion(input_text, aspect, opinion_words)221        222        encoded_inputs = tokenizer(aspect_text, text_pair=input_text, max_seq_len=max_seq_len, return_length=True)223        input_ids = paddle.to_tensor([encoded_inputs["input_ids"]])224        token_type_ids = paddle.to_tensor([encoded_inputs["token_type_ids"]])225 226        logits = cls_model(input_ids, token_type_ids=token_type_ids)227        prediction = logits.argmax(axis=1).numpy()[0]228 229        result = {"aspect": aspect, "opinions": opinion_words, "sentiment": cls_id2label[prediction]}230        results.append(result)231 232    # print results233    return format_print(results)234max_seq_len = 1024235gr.Interface(inputs=["text"],outputs=["text"],fn= predict).launch()