文本分类炼丹实录（上篇）

Posted 2021-09-21 小强同学

文本分类炼丹实录（上篇）

在自然语言处理领域，文本分类是常见且基础的任务，并且很多任务如关系抽取等都有文本分类的影子。其中，本文旨在通过文本分类的例子，详述NLP任务常见流程，常见模型结构以及常见调参方法。

数据清洗与分析

正常我们在实际任务中，都是需要对NLP数据进行清洗的，包括标点处理，无意义的词，以及同义词的多种写法等。这里我不多赘述，这种处理与最后的结果关系极大，这类处理也是需要具体问题具体分析。而数据的分析却是有很多类似的地方，我们一般需要查看标签的分布，看看数据是否均衡，看看数据基本情况，如句子的最大长度，平均长度等。下面是我们使用的文本分类的训练集数据。首先可以看到数据有4个字段，包括id，label，标签描述和句子。
标签的分布情况：

同时对于NLP任务，我们还需要关注一下句子长度，长文本与段文本的差别还是有的。很多时候我们并不能直接按最大句子长度处理数据，如果按最长句子进行处理，那么大量的句子需要pad处理，这样的pad引入了更多的参数，会增加过拟合的风险。通常我们可以使用中位数加方差的方式来确定长度。再者，我们还需要统计一下数据中所有词的分布，我们不必使用所有的单词。而高频词中，“的”，“了”等词出现频率是比较高的。正常我们可以取log进行选择所需要的词。过高或过低频率词可以去除，这样的话我们可以构建转属于该数据的停用词表。
我们选取数据中出现频率300-8000的词，其他词构建stopwords词表：

我们将分词之后并且去除停用词后的数据整理成新的数据，也就是在原始的数据后加上几个字段。

Baseline设计

在NLP任务中，TFIDF方法是最常见的baseline，它往往可以取得一般的效果且实现起来比较简单快捷。首先我们导入sklearn等需要使用的工具包，当然常见的NLP工具如gensim中也都有类似的实现函数api可以调用。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
import pandas as pd
from icecream import ic
from sklearn.metrics import accuracy_score
from sklearn.metrics import recall_score
from sklearn.metrics import precision_score
from sklearn.metrics import f1_score

python中常量我们使用大写的变量表示：

TRAIN_CORPUS = 'train_after_analysis.csv'
STOP_WORDS = 'stopwords.txt'
WORDS_COLUMN = 'words_keep'

加载数据确定主要参数：

content = pd.read_csv(TRAIN_CORPUS)
corpus = content[WORDS_COLUMN].values
stop_words_size = 100
WORDS_LONG_TAIL_BEGIN = 10000
WORDS_SIZE = WORDS_LONG_TAIL_BEGIN - stop_words_size
stop_words = open(STOP_WORDS).read().split()[:stop_words_size]

构建tfidf，计算tfidf矩阵作为文本的表示：

tfidf = TfidfVectorizer(max_features=WORDS_SIZE, stop_words=stop_words)
text_vectors = tfidf.fit_transform(corpus)
print(text_vectors.shape)

最后拆分数据集使用随机森林进行分类：

targets = content['label']
x_train, x_test, y_train, y_test = train_test_split(text_vectors, targets, test_size=0.2, random_state=0)
rf = RandomForestClassifier()
rf.fit(x_train, y_train)
accuracy = accuracy_score(rf.predict(x_test), y_test)
ic(accuracy)

最终我们可以看到一个基准结果：

加载预训练静态词向量

Tfidf是一种统计学方法，某种意义上也是以词袋模型来表示文本。word2vec是一种通过大量文本训练的无监督表示文本的方法。也就是通过大量无监督的数据训练文本，使用一个高维向量来表示词。这个向量是固定的，所以也称为静态的词向量。该方法在很多方面超越了tfidf的效果。同时很多机构都训练了词向量，如谷歌，腾讯等。这里我们使用微博预训练好的词向量模型进行文本分类任务。

import bz2
import random
import torch
from tqdm import tqdm
from icecream import ic
WORD_EMBEDDING_FILE = 'sgns.weibo.word.bz2'
token2embedding = {}
def get_embedding(vocabulary: set):
    with bz2.open(WORD_EMBEDDING_FILE) as f:
        token_vectors = f.readlines()
        vob_size, dim = token_vectors[0].split()
        for line in tqdm(token_vectors[1:]):
            tokens = line.split()
            token = tokens[0].decode('utf-8')
            if token in vocabulary:
                token2embedding[token] = list(map(float, tokens[1:]))
                assert len(token2embedding[token]) == int(dim)
    UNK, PAD, BOS, EOS = '<unk> <pad> <bos> <eos>'.split()
    special_token_num = 4
    token2id = {token: _id for _id, token in enumerate(token2embedding.keys(), special_token_num)}
    token2id[PAD] = 0
    token2id[UNK] = 1
    token2id[BOS] = 2
    token2id[EOS] = 3
    id2vec = {token2id[token]: embedding for token, embedding in token2embedding.items()}
    id2vec[0] = [0.] * int(dim)
    id2vec[1] = [0.] * int(dim)
    id2vec[2] = [random.uniform(-1, 1)] * int(dim)
    id2vec[3] = [random.uniform(-1, 1)] * int(dim)

    embedding = [id2vec[_id] for _id in range(len(id2vec))]
    # embedding 0, 1, 2, 3, 4, 5, ... N
    return torch.tensor(embedding, dtype=torch.float), token2id, len(vocabulary) + 4

其中，我们加载已经训练好了的词向量，同时，在NLP任务中，我们还需要加入4个特殊的token，包括pad，unk,bos和eos。我们给它们4个id。pad是补全，unk是用于我们未见过的单词，bos是句子开始标志，eos是句子结束的标志。

此外，我们先加载预训练的词向量，同时需要做的处理就是将它其中的向量值换成tensor形式。我们使用一个例子来看看加载的情况：

if __name__ == '__main__':
    some_test_words = ['今天', '真是', '一个', '好日子']
    embedding, token2id, _ = get_embedding(set(some_test_words))

加载的词向量显示为：
这样，我们可以把加载词向量的代码整理成一个文件：

import bz2
import random
from tqdm import tqdm
from icecream import ic
import torch
WORD_EMBEDDING_FILE = 'dataset/sgns.weibo.word.bz2'
token2embedding = {}
with bz2.open(WORD_EMBEDDING_FILE) as f:
    token_vectors = f.readlines()
    vob_size, dim = token_vectors[0].split()
print('load embedding file: {} end!'.format(WORD_EMBEDDING_FILE))
def get_embedding(vocabulary: set):
    for line in tqdm(token_vectors[1:]):
        tokens = line.split()
        token = tokens[0].decode('utf-8')
        if token in vocabulary:
            token2embedding[token] = list(map(float, tokens[1:]))
            assert len(token2embedding[token]) == int(dim)
    UNK, PAD, BOS, EOS = '<unk> <pad> <bos> <eos>'.split()
    special_token_num = 4
    token2id = {token: _id for _id, token in enumerate(token2embedding.keys(), special_token_num)}
    token2id[PAD] = 0
    token2id[UNK] = 1
    token2id[BOS] = 2
    token2id[EOS] = 3
    id2vec = {token2id[token]: embedding for token, embedding in token2embedding.items()}
    id2vec[0] = [0.] * int(dim)
    id2vec[1] = [0.] * int(dim)
    id2vec[2] = [random.uniform(-1, 1)] * int(dim)
    id2vec[3] = [random.uniform(-1, 1)] * int(dim)
    embedding = [id2vec[_id] for _id in range(len(id2vec))]
    # embedding 0, 1, 2, 3, 4, 5, ... N
    return torch.tensor(embedding, dtype=torch.float), token2id, len(vocabulary) + 4

数据加载文件

我们将数据集划分，分词等操作写成一个数据加载文件，主要目的是将原始数据集中的文本标签等拿出来，构建分词特征，构建word2id等：

import numpy as np
import pandas as pd
import jieba
from collections import defaultdict
import torch
from operator import add
from functools import reduce
from collections import Counter
from embedding import get_embedding
from torch.utils.data import DataLoader
from icecream import ic

def add_with_print(all_corpus):
    add_with_print.i = 0

    def _wrap(a, b):
        add_with_print.i += 1
        print('{}/{}'.format(add_with_print.i, len(all_corpus)), end=' ')
        return a + b

    return _wrap

def get_all_vocabulary(train_file_path, vocab_size):
    CUT, SENTENCE = 'cut', 'sentence'

    corpus = pd.read_csv(train_file_path)
    corpus[CUT] = corpus[SENTENCE].apply(lambda s: ' '.join(list(jieba.cut(s))))
    sentence_counters = map(Counter, map(lambda s: s.split(), corpus[CUT].values))
    chose_words = reduce(add_with_print(corpus), sentence_counters).most_common(vocab_size)
    return [w for w, _ in chose_words]

def tokenizer(sentence, vocab: dict):
    UNK = 1
    ids = [vocab.get(word, UNK) for word in jieba.cut(sentence)]
    return ids

def get_train_data(train_file, vocab2ids):
    val_ratio = 0.2
    content = pd.read_csv(train_file)
    num_val = int(len(content) * val_ratio)
    LABEL, SENTENCE = 'label', 'sentence'
    labels = content[LABEL].values
    content['input_ids'] = content[SENTENCE].apply(lambda s: ' '.join([str(id_) for id_ in tokenizer(s, vocab2ids)]))
    sentence_ids = np.array([[int(id_) for id_ in v.split()] for v in content['input_ids'].values])
    ids = np.random.choice(range(len(content)), size=len(content))
    # shuffle ids
    train_ids = ids[num_val:]
    val_ids = ids[:num_val]
    X_train, y_train = sentence_ids[train_ids], labels[train_ids]
    X_val, y_val = sentence_ids[val_ids], labels[val_ids]
    label2id = {label: i for i, label in enumerate(np.unique(y_train))}
    id2label = {i: label for label, i in label2id.items()}
    y_train = torch.tensor([label2id[y] for y in y_train], dtype=torch.long)
    y_val = torch.tensor([label2id[y] for y in y_val], dtype=torch.long)
    return X_train, y_train, X_val, y_val, label2id, id2label

def build_dataloader(X_train, y_train, X_val, y_val, batch_size):
    train_dataloader = DataLoader([(x, y) for x, y in zip(X_train, y_train)], batch_size=batch_size, num_workers=4, shuffle=True)
    val_dataloader = DataLoader([(x, y) for x, y in zip(X_val, y_val)], batch_size=batch_size, num_workers=4, shuffle=True)
    return train_dataloader, val_dataloader
if __name__ == '__main__':
    # vocab_size = 10000
    # vocabulary = get_all_vocabulary(train_file_path='dataset/train.csv', vocab_size=vocab_size)
    # assert isinstance(vocabulary, list)
    # assert isinstance(vocabulary[0], str)
    # assert len(vocabulary) <= vocab_size
    #
    f = open('dataset/vocabulary.txt', 'r')
    vocabulary = f.readlines()
    vocabulary = [v.strip() for v in vocabulary]

    embedding, token2id, vocab_size = get_embedding(set(vocabulary))

    X_train, y_train, X_val, y_val, label2id, id2label = get_train_data('train.csv', vocab2ids=token2id)

    print(X_train, y_train, X_val, y_val, label2id, id2label)

    train_loader, val_loader = build_dataloader(X_train, y_train, X_val, y_val, batch_size=128)

    for i, (x, y) in enumerate(train_loader):
        ic(x)
        ic(y)
        if i > 10: break

textcnn进行文本分类

加载词向量后，我们通常使用一些深度神经网络进行特征提取。TextCNN是最常见的文本分类模型之一：

class TextCNN(nn.Module):
    def __init__(self, word_embedding, each_filter_num, filter_heights, drop_out, num_classes):
        super(TextCNN, self).__init__()
        self.embedding = nn.Embedding.from_pretrained(word_embedding, freeze=True)
        self.convs = nn.ModuleList([
            nn.Conv2d(in_channels=1, out_channels=each_filter_num,
                      kernel_size=(h, word_embedding.shape[0]))
            for h in filter_heights
        ])
        self.dropout = nn.Dropout(drop_out)
        self.fc = nn.Linear(each_filter_num * len(filter_heights), num_classes)
    def conv_and_pool(self, x, conv):
        x = F.relu(conv(x)).squeeze(3)
        x = F.max_pool1d(x, x.size(2)).squeeze(2)
        return x
    def forward(self, input_ids=None):
        word_embeddings = self.embedding(input_ids)
        sentence_embedding = word_embeddings.unsqueeze(1)
        out = torch.cat([self.conv_and_pool(sentence_embedding, conv) for conv in self.convs], 1)
        out = self.dropout(out)
        out = self.fc(out)
        outputs = (out, )
        return outputs

首先我们从预训练文件中加载词向量，nn.Embedding.from_pretrained()函数中第一个参数就是传入的词向量矩阵，freeze是在训练过程中是否冻结这层。nn.ModuleList是可以生成一个模型的列表，简单来说这个列表里是多个卷积之后的结果。nn.Conv2d()中输入的通道数为1，因为不是RGB这种形式，输出通道即多个卷积核计算后的结果。后面我们再加上dropout层以及一个线性层，经过线性层后分为n个类别。我们句子的嵌入维度，我们需要进行处理后输入模型：sentence_embedding = word_embeddings.unsqueeze(1)。每一个卷积核都有一个输出，多个卷积核形成一个矩阵经过线性层分为多个类别。
测试模型输出结果：

if __name__ == '__main__':
    some_text_sentence = '今天股市大跌'
    words = list(jieba.cut(some_text_sentence))
    embedding, token2id, _ = get_embedding(set(words))
    text_cnn_model = TextCNN(embedding, each_filter_num=128, filter_heights=[2, 3, 5], drop_out=0.3,num_classes=15)
    ids =[token2id[w] for w in words]
    some_text_sentence = '测试一个新句子'
    words = list(jieba.cut(some_text_sentence))
    embedding, token2id, _ = get_embedding(set(words))

本文我们使用基于统计学方法与词向量方法进行文本分类，其中数据处理，加载词向量等部分都是NLP任务中常见操作，完成了baseline基本模块构建。后一篇我们将使用Bert模型进行文本分类并进行炼丹。