文本情感分类

文本情感分类 1. 案例介绍

为了对前面的word embedding这种常用的文本向量化的方法进行巩固，这里完成一个文本情感分类的案例

现在有一个经典的数据集IMDB数据集，地址：http://ai.stanford.edu/~amaas/data/sentiment/，这是一份包含了5万条流行电影的评论数据，其中训练集25000条，测试集25000条。数据格式如下：

下图左边为名称，其中名称包含两部分，分别是序号和情感评分，（1-4为neg，5-10为pos），右边为评论内容

根据上述的样本，需要使用pytorch完成模型，实现对评论情感进行预测

 

2. 思路分析

首先可以把上述问题定义为分类问题，情感评分分为1-10，10个类别（也可以理解为回归问题，这里当做分类问题考虑）。那么根据之前的经验，大致流程如下：

1. 准备数据集

2. 构建模型

3. 模型训练

4. 模型评估

知道思路之后，那么我们一步步来完成上述步骤

3. 准备数据集

准备数据集和之前的方法一样，实例化dataset，准备dataloader，最终数据可以处理成如下格式：

其中有两点需要注意：

1. 如何完成基础的Dataset的构建和Dataloader的准备

2. 每个batch中文本的长度不一致的问题如何解决

3. 每个batch中的文本如何转化为数字序列

3.1 基础Dataset的准备    【当Dataset中的返回input结果如果是字符串的时候，可以通过修改collate_fn解决异常问题】

dataset.py

import os
import re
from torch.utils.data import Dataset, DataLoader

data_base_path = r'./aclImdb/'


#  1.定义token的方法
def tokenize(test):
    filters = ['!','"','#','$','%','&','\(','\)','\*','\+',',','-','\.','/',':',';','','\?','@'
        ,'\[','\\','\]','^','_','`','\{','\|','\}','~','\t','\n','\x97','\x96','”','“',]
    text = re.sub("", " ", test, flags=re.S)
    text = re.sub("|".join(filters), " ", test, flags=re.S)
    return [i.strip() for i in text.split()]


#  2.准备dataset
class ImdbDataset(Dataset):
    def __init__(self, mode):
        super().__init__()
        if mode == "train":
            text_path = [os.path.join(data_base_path, i) for i in ["train/neg", "train/pos"]]
        else:
            text_path = [os.path.join(data_base_path, i) for i in ["test/neg", "test/pos"]]
        self.total_file_path_list = []
        for i in text_path:
            self.total_file_path_list.extend([os.path.join(i, j) for j in os.listdir(i)])

    def __getitem__(self, item):
        cur_path = self.total_file_path_list[item]
        cur_filename = os.path.basename(cur_path)
        label = int(cur_filename.split("_")[-1].split(".")[0]) - 1  # 处理标题，获取标签label，转化为从[0-9]
        text = tokenize(open(cur_path).read().strip())  # 直接按照空格进行分词
        return label, text

    def __len__(self):
        return len(self.total_file_path_list)


#  3.实例化，准别dataloader
dataset = ImdbDataset(mode="train")
dataloader = DataLoader(dataset=dataset, batch_size=2, shuffle=True)

#  4.观察数输出结果
for idx, (label, text) in enumerate(dataloader):
    print("idx:", idx)
    print("label:", label)
    print("text:", text)
    break

此时运行是报错的，

出现问题的原因在于Dataloader中的参数collate_fn

collate_fn的默认值为torch自定义的default_collate,collate_fn的作用就是对每个batch进行处理，而默认的default_collate处理出错。

解决问题的思路：

手段1：考虑先把数据转化为数字序列，观察其结果是否符合要求，之前使用DataLoader并未出现类似错误

手段2：考虑自定义一个collate_fn，观察结果

这里使用方式2，自定义一个collate_fn,然后观察结果：

def collate_fn(batch):
    #  batch是一个列表，其中是一个一个的元组，每个元组是dataset中_getitem__的结果
    batch = list(zip(*batch))
    labels = torch.tensor(batch[0], dtype=torch.int32)
    texts = batch[1]
    del batch
    return labels, texts

修改后的总代码：

dataset.py

import os
import re
import torch
from torch.utils.data import Dataset, DataLoader

data_base_path = r'./aclImdb/'


#  1.定义token的方法
def tokenize(test):
    filters = ['!','"','#','$','%','&','\(','\)','\*','\+',',','-','\.','/',':',';','','\?','@'
        ,'\[','\\','\]','^','_','`','\{','\|','\}','~','\t','\n','\x97','\x96','”','“',]
    text = re.sub("", " ", test, flags=re.S)
    text = re.sub("|".join(filters), " ", test, flags=re.S)
    return [i.strip() for i in text.split()]


#  2.准备dataset
class ImdbDataset(Dataset):
    def __init__(self, mode):
        super().__init__()
        if mode == "train":
            text_path = [os.path.join(data_base_path, i) for i in ["train/neg", "train/pos"]]
        else:
            text_path = [os.path.join(data_base_path, i) for i in ["test/neg", "test/pos"]]
        self.total_file_path_list = []
        for i in text_path:
            self.total_file_path_list.extend([os.path.join(i, j) for j in os.listdir(i)])

    def __getitem__(self, item):
        cur_path = self.total_file_path_list[item]
        cur_filename = os.path.basename(cur_path)
        label = int(cur_filename.split("_")[-1].split(".")[0]) - 1  # 处理标题，获取标签label，转化为从[0-9]
        text = tokenize(open(cur_path).read().strip())  # 直接按照空格进行分词
        return label, text

    def __len__(self):
        return len(self.total_file_path_list)


def collate_fn(batch):
    #  batch是一个列表，其中是一个一个的元组，每个元组是dataset中_getitem__的结果
    batch = list(zip(*batch))
    labels = torch.tensor(batch[0], dtype=torch.int32)
    texts = batch[1]
    del batch
    return labels, texts


#  3.实例化，准别dataloader
dataset = ImdbDataset(mode="train")
dataloader = DataLoader(dataset=dataset, batch_size=2, shuffle=True, collate_fn=collate_fn)

#  4.观察数输出结果
for idx, (label, text) in enumerate(dataloader):
    print("idx:", idx)
    print("label:", label)
    print("text:", text)
    break

运行结果：

可以把上述代码的实例化部分封装成函数，示例代码如下，后面的代码是直接调用封装好的函数

import os
import re
import torch
from torch.utils.data import Dataset, DataLoader

data_base_path = r'../data/aclImdb/'


#  1.定义token的方法
def tokenize(test):
    filters = ['!', '"', '#', '$', '%', '&', '\(', '\)', '\*', '\+', ',', '-', '\.', '/', ':', ';', '', '\?',
               '@'
        , '\[', '\\', '\]', '^', '_', '`', '\{', '\|', '\}', '~', '\t', '\n', '\x97', '\x96', '”', '“', ]
    text = re.sub("", " ", test, flags=re.S)
    text = re.sub("|".join(filters), " ", test, flags=re.S)
    return [i.strip() for i in text.split()]


#  2.准备dataset
class ImdbDataset(Dataset):
    def __init__(self, mode):
        super().__init__()
        if mode == "train":
            text_path = [os.path.join(data_base_path, i) for i in ["train/neg", "train/pos"]]
        else:
            text_path = [os.path.join(data_base_path, i) for i in ["test/neg", "test/pos"]]
        self.total_file_path_list = []
        for i in text_path:
            self.total_file_path_list.extend([os.path.join(i, j) for j in os.listdir(i)])

    def __getitem__(self, item):
        cur_path = self.total_file_path_list[item]
        cur_filename = os.path.basename(cur_path)
        label = int(cur_filename.split("_")[-1].split(".")[0]) - 1  # 处理标题，获取标签label，转化为从[0-9]
        text = tokenize(open(cur_path).read().strip())  # 直接按照空格进行分词
        return label, text

    def __len__(self):
        return len(self.total_file_path_list)


def collate_fn(batch):
    #  batch是一个列表，其中是一个一个的元组，每个元组是dataset中_getitem__的结果
    batch = list(zip(*batch))
    labels = torch.tensor(batch[0], dtype=torch.int32)
    texts = batch[1]
    del batch
    return labels, texts


def get_dataloader():
    imdb_dataset = ImdbDataset('train')
    data_loader = DataLoader(imdb_dataset, shuffle=True, batch_size=2, collate_fn=collate_fn)
    return data_loader


#  3.实例化，准别dataloader
dataset = ImdbDataset(mode="train")
dataloader = DataLoader(dataset=dataset, batch_size=2, shuffle=True, collate_fn=collate_fn)

#  4.观察数输出结果
# for idx, (label, text) in enumerate(dataloader):
for idx, (label, text) in enumerate(get_dataloader()):
    print("idx:", idx)
    print("label:", label)
    print("text:", text)
    break

3.2 文本序列化

再介绍word embedding的时候，不会直接把文本转化为向量，而是先转化为数字，再把数字转化为向量，那么这个过程该如何实现呢？

这里我们可以考虑把文本中的每个词语和其对应的数字，使用字典保存，同时实现方法把句子通过字典映射为包含数字的列表。

实现文本序列化之前，考虑以下几点:

1. 如何使用字典把词语和数字进行对应

2. 不同的词语出现的次数不尽相同，是否需要对高频或者低频词语进行过滤，以及总的词语数量是否需要进行限制

3. 得到词典之后，如何把句子转化为数字序列，如何把数字序列转化为句子

4. 不同句子长度不相同，每个batch的句子如何构造成相同的长度（可以对短句子进行填充，填充特殊字符）

5. 对于新出现的词语在词典中没有出现怎么办（可以使用特殊字符代理）

思路分析：

1. 对所有句子进行分词

2. 词语存入字典，根据次数对词语进行过滤，并统计次数

3. 实现文本转数字序列的方法

4. 实现数字序列转文本方法

wordSequence.py

import numpy as np


#  构建字典，实现方法把句子转换成为数字序列和其翻转
class Word2Sequence(object):
    UNK_TAG = "UNK"  # 表示特殊的字符
    PAD_TAG = "PAD"  # 表示对句子的填充

    UNK = 0
    PAD = 1

    def __init__(self):
        # 初始词典
        self.dict = {
            self.UNK_TAG: self.UNK,
            self.PAD_TAG: self.PAD  # 初始键值对
        }
        self.fited = False

    #  把单词转换为索引
    def to_index(self, word):
        """word -> index"""
        assert self.fited == True, "必须先进行fit操作"
        return self.dict.get(word, self.UNK)

    #  把索引转换为词语/字
    def to_word(self, index):
        """index -> word"""
        assert self.fited, "必须先进行fit操作"
        if index in self.inversed_dict:
            return self.inversed_dict[index]
        return self.UNK_TAG

    def __len__(self):
        return self(self.dict)

    def fit(self, sentences, min_count=1, max_count=None, max_feature=None):  # max_count最大不限制
        """
        把所有的句子放到词典中去，把单个句子放到dict中去
        :param sentence: [[word1,word2,word3],[word1,word3,wordn..],...]
        :param min_count:最小出现的次数
        :param max_count:最大出现的次数
        :param max_feature:总词语的最大数量  一个保留多少个单词
        :return:
        """
        count = {}
        for sentence in sentences:
            for a in sentence:
                if a not in count:
                    count[a] = 0
                count[a] += 1
        # 比最小的数量大和比最大的数量小的需要
        if min_count is not None:  # 删除count中词频小于min word
            count = {k: v for k, v in count.items() if v >= min_count}
        if max_count is not None:  # 删除词语大于max 的值
            count = {k: v for k, v in count.items() if v  max_feature:
                count = count[-int(max_feature):]
            for w, _ in count:
                self.dict[w] = len(self.dict)
        else:
            for w in sorted(count.keys()):
                self.dict[w] = len(self.dict)

        self.fited = True
        #  准备一个index->word的字典
        self.inversed_dict = dict(zip(self.dict.values(), self.dict.keys()))

    #  把新来的句子转换成序列
    def transform(self, sentence, max_len=None):
        """
        实现把句子转化为数组（向量）
        :param sentence:
        :param max_len:
        :return:
        """
        assert self.fited, "必须先进行fit操作"
        if max_len is not None:
            r = [self.PAD] * max_len
        else:
            r = [self.PAD] * len(sentence)
        if max_len is not None and len(sentence) > max_len:
            sentence = sentence[:max_len]
        for index, word in enumerate(sentence):
            r[index] = self.to_index(word)
        return np.array(r, dtype=np.int64)

    def inverse_transform(self, indices):
        """
        实现从数组 转化为文字
        :param indices: [1,2,3....]
        :return:[word1,word2.....]
        """
        sentence = []
        for i in indices:
            word = self.to_word(i)
            sentence.append(word)
        return sentence


if __name__ == '__main__':
    w2s = Word2Sequence()
    w2s.fit([
        ["你", "好", "么"],
        ["你", "好", "哦"]])
    print(w2s.dict)
    print(w2s.fited)
    print(w2s.transform(["你", "好", "嘛"]))
    print(w2s.transform(["你好嘛"], max_len=10))

运行结果：

完成了wordsequence之后，接下来就是保存现有样本中的数据字典，方便后续的使用。

实现对IMDB数据的处理和保存

#1. 对IMDB的数据记性fit操作
def fit_save_word_sequence():
    from wordSequence import Word2Sequence

    ws = Word2Sequence()
    train_path = [os.path.join(data_base_path,i)  for i in ["train/neg","train/pos"]]
    total_file_path_list = []
    for i in train_path:
        total_file_path_list.extend([os.path.join(i, j) for j in os.listdir(i)])
    for cur_path in tqdm(total_file_path_list,ascii=True,desc="fitting"):
        ws.fit(tokenize(open(cur_path).read().strip()))
    ws.build_vocab()
    # 对wordSequesnce进行保存
    pickle.dump(ws,open("./model/ws.pkl","wb"))

#2. 在dataset中使用wordsequence
ws = pickle.load(open("./model/ws.pkl","rb"))

def collate_fn(batch):
    MAX_LEN = 500 
    #MAX_LEN = max([len(i) for i in texts]) #取当前batch的最大值作为batch的最大长度

    batch = list(zip(*batch))
    labes = torch.tensor(batch[0],dtype=torch.int)

    texts = batch[1]
    #获取每个文本的长度
    lengths = [len(i) if len(i)