文本分類煉丹實錄（上篇）
在自然語言處理領域，文本分類是常見且基礎的任務，并且很多任務如關系抽取等都有文本分類的影子，其中，本文旨在通過文本分類的例子，詳述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模型進行文本分類并進行煉丹，
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/298045.html
標籤：AI