Bilstm中文微博多情感分析

数据

我的数据是来自github的一个项目：ChineseNlpCorpus 里面收集了蛮多用于自然语言处理的中文数据集/语料。

下载地址： 百度网盘
数据概览： 36 万多条，带情感标注 新浪微博，包含 4 种情感，其中喜悦约 20 万条，愤怒、厌恶、低落各约 5 万条
数据来源： 新浪微博
原数据集： 微博情感分析数据集，网上搜集，具体作者、来源不详

预处理

划分训练集和测试集

将表中的各类情感的数据各抽取前10000条做测试集，剩余的用作训练集

import pandas as pd import openpyxl from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE def test(): file='simplifyweibo_4_moods.csv' data=pd.read_csv(file) label=data.get('label') review = data.get('review') train_review=[] train_label=[] test_review=[] test_label=[] n1=n2=n3=n4=0 for i in range(len(review)): lab=int(label[i]) line=str(review[i]) line=ILLEGAL_CHARACTERS_RE.sub(r'', line) if int(lab)==0: if n1<10000: n1+=1 test_label.append(lab) test_review.append(line) else: train_label.append(lab) train_review.append(line) elif int(lab)==1: if n2<10000: n2+=1 test_label.append(lab) test_review.append(line) else: train_label.append(lab) train_review.append(line) elif int(lab)==2: if n3<10000: n3+=1 test_label.append(lab) test_review.append(line) else: train_label.append(lab) train_review.append(line) elif int(lab)==3: if n4<10000: n4+=1 test_label.append(lab) test_review.append(line) else: train_label.append(lab) train_review.append(line) import openpyxl import xlsxwriter xl = openpyxl.Workbook() # 调用对象的add_sheet方法
    sheet1 = xl.create_sheet(index=0) sheet1.cell(1, 1, "label") sheet1.cell(1, 2, "review") for i in range(0, len(train_review)): sheet1.cell(i + 2, 1, train_label[i]) sheet1.cell(i + 2, 2, train_review[i]) xl.save("train.xlsx") if __name__ == '__main__': test()

得到对应的文件

 

 

生成字典

数据中的一些标点符号、特殊符号、英文字母、数字等对于我们的实验都是没有用处的，所以我们需要将他们过滤掉。

去除停用词

def tokenlize(sentence): """ 进行文本分词 :param sentence: str :return: [str,str,str] """ fileters = ['!', '"', '#', '$', '%', '&', '\(', '\)', '\*', '\+', ',', '-', '\.', '/', ':', ';', '<', '=', '>', '\?', '@', '\[', '\\', '\]', '^', '_', '`', '\{', '\|', '\}', '~', '\t', '\n', '\x97', '\x96', '”', '“', ] sentence = re.sub("|".join(fileters), "", sentence) sentence=jieba.cut(sentence,cut_all=False) sentence=' '.join(sentence) result = [i for i in sentence.split(" ") if len(i) > 0] result=movestopwords(result) return result def stopwordslist(filepath): stopwords = [line.strip() for line in open(filepath, 'r', encoding='utf-8').readlines()] return stopwords # 对句子去除停用词
def movestopwords(sentence): stopwords = stopwordslist('data/stopwords.txt')  # 这里加载停用词的路径
    outstr = [] for word in sentence: if word not in stopwords: if word != '\t' and '\n': outstr.append(word) # outstr += " "
    return outstr

设计字典类

""" 文本序列化 """


class Vocab: UNK_TAG = "<UNK>"  # 表示未知字符
    PAD_TAG = "<PAD>"  # 填充符
    PAD = 0 UNK = 1

    def __init__(self): self.dict = {  # 保存词语和对应的数字
 self.UNK_TAG: self.UNK, self.PAD_TAG: self.PAD } self.count = {}  # 统计词频的

    def fit(self, sentence): """ 接受句子，统计词频 :param sentence:[str,str,str] :return:None """
        for word in sentence: self.count[word] = self.count.get(word, 0) + 1  # 所有的句子fit之后，self.count就有了所有词语的词频

    def build_vocab(self, min_count=1, max_count=None, max_features=None): """ 根据条件构造 词典 :param min_count:最小词频 :param max_count: 最大词频 :param max_features: 最大词语数 :return: """
        if min_count is not None: self.count = {word: count for word, count in self.count.items() if count >= min_count} if max_count is not None: self.count = {word: count for word, count in self.count.items() if count <= max_count} if max_features is not None: # [(k,v),(k,v)....] --->{k:v,k:v}
            self.count = dict(sorted(self.count.items(), lambda x: x[-1], reverse=True)[:max_features]) for word in self.count: self.dict[word] = len(self.dict)  # 每次word对应一个数字

        # 把dict进行翻转
        self.inverse_dict = dict(zip(self.dict.values(), self.dict.keys())) def transform(self, sentence, max_len=None): """ 把句子转化为数字序列 :param sentence:[str,str,str] :return: [int,int,int] """
        if len(sentence) > max_len: sentence = sentence[:max_len] else: sentence = sentence + [self.PAD_TAG] * (max_len - len(sentence))  # 填充PAD

        return [self.dict.get(i, 1) for i in sentence] def inverse_transform(self, incides): """ 把数字序列转化为字符 :param incides: [int,int,int] :return: [str,str,str] """
        return [self.inverse_dict.get(i, "<UNK>") for i in incides] def __len__(self): return len(self.dict) # # 以下是调试代码 # if __name__ == '__main__': # sentences = [["今天", "天气", "很", "好"], # ["今天", "去", "吃", "什么"]] # ws = Vocab() # for sentence in sentences: # # 统计词频 # ws.fit(sentence) # # 构造词典 # ws.build_vocab(min_count=1) # print(ws.dict) # # 把句子转换成数字序列 # ret = ws.transform(["好", "好", "好", "好", "好", "好", "好", "热", "呀"], max_len=13) # print(ret) # # 把数字序列转换成句子 # ret = ws.inverse_transform(ret) # print(ret) # pass

dataset

# -*-coding:utf-8-*-
import os import pickle import re import zipfile import jieba from torch.utils.data import Dataset, DataLoader from tqdm import tqdm import pandas as pd class ImdbDataset(Dataset): def __init__(self, train=True): if train == True: url = 'data/train.xlsx'
        else: url = "data/test.xlsx" data = pd.read_excel(url) sentence = data.get('review') label = data.get('label') self.sentence_list=sentence self.label_list=label def __getitem__(self, idx): line_text=self.sentence_list[idx] # 从txt获取评论并分词
        review = tokenlize(str(line_text)) # 获取评论对应的label
        label = int(self.label_list[idx]) return review, label def __len__(self): return len(self.sentence_list) def tokenlize(sentence): """ 进行文本分词 :param sentence: str :return: [str,str,str] """ fileters = ['!', '"', '#', '$', '%', '&', '\(', '\)', '\*', '\+', ',', '-', '\.', '/', ':', ';', '<', '=', '>', '\?', '@', '\[', '\\', '\]', '^', '_', '`', '\{', '\|', '\}', '~', '\t', '\n', '\x97', '\x96', '”', '“', ] sentence = re.sub("|".join(fileters), "", sentence) sentence=jieba.cut(sentence,cut_all=False) sentence=' '.join(sentence) result = [i for i in sentence.split(" ") if len(i) > 0] result=movestopwords(result) return result def stopwordslist(filepath): stopwords = [line.strip() for line in open(filepath, 'r', encoding='utf-8').readlines()] return stopwords # 对句子去除停用词
def movestopwords(sentence): stopwords = stopwordslist('data/stopwords.txt')  # 这里加载停用词的路径
    outstr = [] for word in sentence: if word not in stopwords: if word != '\t' and '\n': outstr.append(word) # outstr += " "
    return outstr # 以下为调试代码
def collate_fn(batch): """ 对batch数据进行处理 :param batch: [一个getitem的结果，getitem的结果,getitem的结果] :return: 元组 """ reviews, labels = zip(*batch) return reviews, labels if __name__ == "__main__": from 情感分析.imdb_sentiment.vocab import Vocab imdb_dataset = ImdbDataset(True) my_dataloader = DataLoader(imdb_dataset, batch_size=2, shuffle=True, collate_fn=collate_fn) for review,label in my_dataloader: vocab_model = pickle.load(open("./models/vocab.pkl", "rb")) print(review[0]) result = vocab_model.transform(review[0], 30) print(result) break

构建字典

# -*-coding:utf-8-*-
import pickle from tqdm import tqdm from 情感分析.weibo_many_emotion import dataset # from 情感分析.imdb_sentiment.vocab import Vocab
from torch.utils.data import DataLoader class Vocab: UNK_TAG = "<UNK>"  # 表示未知字符
    PAD_TAG = "<PAD>"  # 填充符
    PAD = 0 UNK = 1

    def __init__(self): self.dict = {  # 保存词语和对应的数字
 self.UNK_TAG: self.UNK, self.PAD_TAG: self.PAD } self.count = {}  # 统计词频的

    def fit(self, sentence): """ 接受句子，统计词频 :param sentence:[str,str,str] :return:None """
        for word in sentence: self.count[word] = self.count.get(word, 0) + 1  # 所有的句子fit之后，self.count就有了所有词语的词频

    def build_vocab(self, min_count=1, max_count=None, max_features=None): """ 根据条件构造 词典 :param min_count:最小词频 :param max_count: 最大词频 :param max_features: 最大词语数 :return: """
        if min_count is not None: self.count = {word: count for word, count in self.count.items() if count >= min_count} if max_count is not None: self.count = {word: count for word, count in self.count.items() if count <= max_count} if max_features is not None: # [(k,v),(k,v)....] --->{k:v,k:v}
            self.count = dict(sorted(self.count.items(), lambda x: x[-1], reverse=True)[:max_features]) for word in self.count: self.dict[word] = len(self.dict)  # 每次word对应一个数字

        # 把dict进行翻转
        self.inverse_dict = dict(zip(self.dict.values(), self.dict.keys())) def transform(self, sentence, max_len=None): """ 把句子转化为数字序列 :param sentence:[str,str,str] :return: [int,int,int] """
        if len(sentence) > max_len: sentence = sentence[:max_len] else: sentence = sentence + [self.PAD_TAG] * (max_len - len(sentence))  # 填充PAD

        return [self.dict.get(i, 1) for i in sentence] def inverse_transform(self, incides): """ 把数字序列转化为字符 :param incides: [int,int,int] :return: [str,str,str] """
        return [self.inverse_dict.get(i, "<UNK>") for i in incides] def __len__(self): return len(self.dict) def collate_fn(batch): """ 对batch数据进行处理 :param batch: [一个getitem的结果，getitem的结果,getitem的结果] :return: 元组 """ reviews, labels = zip(*batch) return reviews, labels def get_dataloader(train=True): imdb_dataset = dataset.ImdbDataset(train) my_dataloader = DataLoader(imdb_dataset, batch_size=200, shuffle=True, collate_fn=collate_fn) return my_dataloader if __name__ == '__main__': ws = Vocab() dl_train = get_dataloader(True) dl_test = get_dataloader(False) for reviews, label in tqdm(dl_train, total=len(dl_train)): for sentence in reviews: ws.fit(sentence) for reviews, label in tqdm(dl_test, total=len(dl_test)): for sentence in reviews: ws.fit(sentence) ws.build_vocab() print(len(ws)) pickle.dump(ws, open("./models/vocab.pkl", "wb"))

模型训练

# -*-coding:utf-8-*-
import pickle import torch import torch.nn as nn import torch.nn.functional as F from torch.optim import Adam from torch.utils.data import DataLoader from tqdm import tqdm from 情感分析.weibo_many_emotion import dataset from 情感分析.中文情感分类.vocab import Vocab train_batch_size = 512 test_batch_size = 128 voc_model = pickle.load(open("./models/vocab.pkl", "rb")) sequence_max_len = 100


def collate_fn(batch): """ 对batch数据进行处理 :param batch: [一个getitem的结果，getitem的结果,getitem的结果] :return: 元组 """ reviews, labels = zip(*batch) reviews = torch.LongTensor([voc_model.transform(i, max_len=sequence_max_len) for i in reviews]) labels = torch.LongTensor(labels) return reviews, labels def get_dataloader(train=True): imdb_dataset = dataset.ImdbDataset(train) batch_size = train_batch_size if train else test_batch_size return DataLoader(imdb_dataset, batch_size=batch_size, shuffle=True, collate_fn=collate_fn) class ImdbModel(nn.Module): def __init__(self): super(ImdbModel, self).__init__() self.embedding = nn.Embedding(num_embeddings=len(voc_model), embedding_dim=200, padding_idx=voc_model.PAD).to() self.lstm = nn.LSTM(input_size=200, hidden_size=64, num_layers=2, batch_first=True, bidirectional=True, dropout=0.1) self.fc1 = nn.Linear(64 * 2, 64) self.fc2 = nn.Linear(64, 4) def forward(self, input): """ :param input:[batch_size,max_len] :return: """ input_embeded = self.embedding(input)  # input embeded :[batch_size,max_len,200]
 output, (h_n, c_n) = self.lstm(input_embeded)  # h_n :[4,batch_size,hidden_size]
        # out :[batch_size,hidden_size*2]
        out = torch.cat([h_n[-1, :, :], h_n[-2, :, :]], dim=-1)  # 拼接正向最后一个输出和反向最后一个输出

        # 进行全连接
        out_fc1 = self.fc1(out) # 进行relu
        out_fc1_relu = F.relu(out_fc1) # 全连接
        out_fc2 = self.fc2(out_fc1_relu)  # out :[batch_size,2]
        return F.log_softmax(out_fc2, dim=-1) def device(): if torch.cuda.is_available(): return torch.device('cuda') else: return torch.device('cpu') def train(imdb_model, epoch): """ :param imdb_model: :param epoch: :return: """ train_dataloader = get_dataloader(train=True) optimizer = Adam(imdb_model.parameters()) for i in range(epoch): bar = tqdm(train_dataloader, total=len(train_dataloader)) for idx, (data, target) in enumerate(bar): optimizer.zero_grad() data = data.to(device()) target = target.to(device()) output = imdb_model(data) loss = F.nll_loss(output, target) loss.backward() optimizer.step() bar.set_description("epcoh:{} idx:{} loss:{:.6f}".format(i, idx, loss.item())) torch.save(imdb_model, 'lstm_model.pkl') def test(imdb_model): """ 验证模型 :param imdb_model: :return: """ test_loss = 0 correct = 0 imdb_model.eval() test_dataloader = get_dataloader(train=False) with torch.no_grad(): for data, target in tqdm(test_dataloader): data = data.to(device()) target = target.to(device()) output = imdb_model(data) test_loss += F.nll_loss(output, target, reduction='sum').item() pred = output.data.max(1, keepdim=True)[1]  # 获取最大值的位置,[batch_size,1]
            correct += pred.eq(target.data.view_as(pred)).sum() test_loss /= len(test_dataloader.dataset) print('\nTest set: Avg. loss: {:.4f}, Accuracy: {}/{} ({:.2f}%)\n'.format( test_loss, correct, len(test_dataloader.dataset), 100. * correct / len(test_dataloader.dataset))) def xlftest(): import numpy as np model = torch.load('lstm_model.pkl') model.to(device()) from 情感分析.weibo_many_emotion.dataset import tokenlize lines=['哈哈哈开心','真是无语，你们怎么搞的','小姐姐，祝你生日快乐','你他妈的有病'] for line in lines: print(line) review = tokenlize(line) # review=tokenlize(line)
        vocab_model = pickle.load(open("./models/vocab.pkl", "rb")) result = vocab_model.transform(review,sequence_max_len) # print(result)
        data = torch.LongTensor(result).to(device()) data=torch.reshape(data,(1,sequence_max_len)).to(device()) # print(data.shape)
        output = model(data) print(output.data) pred = output.data.max(1, keepdim=True)[1]  # 获取最大值的位置,[batch_size,1]
        print(pred.item()) if pred.item() == 0: print("喜悦") elif pred.item() == 1: print("愤怒") elif pred.item() == 2: print("厌恶") elif pred.item() == 3: print("低落") if __name__ == '__main__': # imdb_model = ImdbModel().to(device())
    # train(imdb_model,20)
    # test(imdb_model)
    xlftest()

测试结果：

对四种分类的准确度只有40左右，一部分原因是数据集不规范，还需要进行调参优化