使用pytorch和torchtext进行文本分类的实例

from nltk.stem import SnowballStemmer
stemmer = SnowballStemmer("english") # 选择语言
from nltk.stem import WordNetLemmatizer 
wnl = WordNetLemmatizer()

stemmer.stem('knives')
# knive
wnl.lemmatize('knives')
# knife

tokenize = lambda x: x.split()

import spacy

spacy_en = spacy.load('en')
def tokenizer(text):
 return [toke.text for toke in spacy_en.tokenizer(text)]

def DataLoader():
 tokenize = lambda x: x.split()
 # 用户评论，include_lengths设为True是为了方便之后使用torch的pack_padded_sequence
 REVIEW = data.Field(sequential=True,tokenize=tokenize, include_lengths=True)
 # 情感极性
 POLARITY = data.LabelField(sequential=False, use_vocab=False, dtype = torch.long)
 # 假如train.csv文件并不是只有两列，比如1、3列是review和polarity，2列是我们不需要的数据，
 # 那么就要添加一个全是None的元组， fields列表存储的Field的顺序必须和csv文件中每一列的顺序对应，
 # 否则review可能就加载到polarity Field里去了
 fields = [('review', REVIEW), (None, None), ('polarity', POLARITY)]
 
 # 加载train,valid,test数据
 train_data, valid_data, test_data = data.TabularDataset.splits(
         path = 'amazon',
         train = 'train.csv',
      validation = 'valid.csv',
      test = 'test.csv',
         format = 'csv',
         fields = fields,
         skip_header = False # 是否跳过文件的第一行
 )
 return REVIEW, POLARITY, train_data

from torchtext.vocab import Vectors, Glove
import torch

REVIEW, POLARITY, train_data = DataLoader()
# vectors = Vectors(name='glove.6B.300d.txt', cache='.vector_cache')
REVIEW.build_vocab(train_data, # 建词表是用训练集建，不要用验证集和测试集
     max_size=400000, # 单词表容量
     vectors='glove.6B.300d', # 还有'glove.840B.300d'已经很多可以选
     unk_init=torch.Tensor.normal_ # 初始化train_data中不存在预训练词向量词表中的单词
)

# print(REVIEW.vocab.freqs.most_common(20)) 数据集里最常出现的20个单词
# print(REVIEW.vocab.itos[:10])  列表 index to word
# print(REVIEW.vocab.stoi)    字典 word to index

pretrained_embeddings = REVIEW.vocab.vectors
model.embedding.weight.data.copy_(pretrained_embeddings)
UNK_IDX = REVIEW.vocab.stoi[REVIEW.unk_token]
PAD_IDX = REVIEW.vocab.stoi[REVIEW.pad_token]
# 因为预训练的权重的unk和pad的词向量不是在我们的数据集语料上训练得到的，所以最好置零
model.embedding.weight.data[UNK_IDX] = torch.zeros(EMBEDDING_DIM)
model.embedding.weight.data[PAD_IDX] = torch.zeros(EMBEDDING_DIM)

train_iterator, valid_iterator, test_iterator = data.BucketIterator.splits(
            (train_data, valid_data, test_data),
            batch_size=32,
            sort_within_batch=True,
            sort_key = lambda x:len(x.review),
            device=torch.device('cpu'))

for batch in train_iterator:
 # 因为REVIEW Field的inclue_lengths为True，所以还会包含一个句子长度的Tensor
 review, review_len = batch.review 
 # review.size = (seq_length, batch_size) , review_len.size = (batch_size, )
 polarity = batch.polarity
 # polarity.size = (batch_size, )
 predictions = model(review, review_lengths)
 loss = criterion(predictions, polarity) # criterion = nn.CrossEntropyLoss()

import torch.nn as nn
import torch.nn.functional as F
from torch.nn.utils.rnn import pack_padded_sequence
import torch


class LSTM(nn.Module):

 def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim,
     n_layers, bidirectional, dropout, pad_idx):
  super(LSTM, self).__init__()
  self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx)
  self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers=n_layers,
       bidirectional=bidirectional, dropout=dropout)
  self.Ws = nn.Parameter(torch.Tensor(hidden_dim, output_dim))
  self.bs = nn.Parameter(torch.zeros((output_dim, )))
  nn.init.uniform_(self.Ws, -0.1, 0.1)
  nn.init.uniform_(self.bs, -0.1, 0.1)
  self.dropout = nn.Dropout(p=0.5)

 def forward(self, x, x_len):
  x = self.embedding(x)
  x = pack_padded_sequence(x, x_len)
  H, (h_n, c_n) = self.lstm(x)
  h_n = self.dropout(h_n)
  h_n = torch.squeeze(h_n)
  res = torch.matmul(h_n, self.Ws) + self.bs
  y = F.softmax(res, dim=1)
  # y.size(batch_size, output_dim)
  return y

def train(model, iterator, optimizer, criterion):
 epoch_loss = 0
 num_sample = 0
 correct = 0

 model.train()
 for batch in iterator:
  optimizer.zero_grad()
  review, review_lengths = batch.review
  polarity = batch.polarity
  predictions = model(review, review_lengths)
  correct += torch.sum(torch.argmax(preds, dim=1) == polarity)
  loss = criterion(predictions, polarity)
  loss.backward()
  epoch_loss += loss.item()
  num_sample += len(batch)
  optimizer.step()

 return epoch_loss / num_sample, correct.float() / num_sample

if __name__ == '__main__':
 for epoch in range(N_EPOCHS):
 train_loss, acc = train(model, train_iter, optimizer, criterion)
 print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {acc* 100:.2f}%')

train_data = data.TabularDataset.splits(
         path = '',
         train = 'train.csv',
         format = 'csv',
         fields = fields,
         skip_header = False # 是否跳过文件的第一行
)

train_data = data.TabularDataset.splits(
         path = '',
         train = 'train.csv',
         format = 'csv',
         fields = fields,
         skip_header = False # 是否跳过文件的第一行
)[0] # 注意这里的切片，选择元组的第一个也是唯一一个元素赋给train_data

train_iter = data.BucketIterator(
   train_data,
   batch_size=32,
   sort_key=lambda x:len(x.review),
   sort_within_batch=True,
   shuffle=True # 训练集需要shuffle，但因为验证测试集不需要
    # 可以生成验证和测试集的迭代器直接用data.iterator.Iterator类就足够了
)