情感分析学习笔记-Task04

83 阅读 0 评论 55 点赞

我是靠谱客的博主苗条汉堡，这篇文章主要介绍情感分析学习笔记-Task04，现在分享给大家，希望可以做个参考。

这次学习的主要是用CNN去处理文本，做情感分析，CNN用在文本上的时候，他的filter是[n x emb_dim]，这里的n，你可以指定为2，3，4，5…，它就类似于n-gram里的n，emb_dim当然就是一个token需要用多少维的向量来表示，其他的地方与前3次的学习内容没有太多变化的地方，主要在模型那里，这里介绍一下原理。

首先，我们的横轴是每个单词（token）embeding的维度，纵轴是文本中的每个单词。如考虑下面2个句子的嵌入句：

然后我们可以使用一个[n x emb_dim]的filter。这将完全覆盖 n个words，因为它们的宽度为emb_dim 尺寸。考虑下面的图像，我们的单词向量用绿色表示。这里我们有4个词和5维嵌入，创建了一个[4x5] “image” 张量。一次覆盖两个词（即bi-grams)）的filter 将是 [2x5] filter，以黄色显示，filter 的每个元素都有一个与之相关的 weight。此filter 的输出（以红色显示）将是一个实数，它是filter覆盖的所有元素的加权和。

然后，filter 向下走一步，以覆盖下一个bi-gram，并计算另一个输出（weighted sum）。

最后，filter 再次向下移动，并计算此 filter 的最终输出。

一般情况下，我们得到的输出是一个向量，其元素数等于图像的高度（或词的长度）减去 filter 的高度加上一。在当前例子中，4-2+1=3。
下面是代码，自己CPU跑起来11分多一点，还可以，适合学习。

复制代码

# 导入所需要的包
import torch
from torchtext.legacy import data
from torchtext.legacy import datasets
import random
import numpy as np

# 数据预处理
SEED = 1234
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
torch.backends.deterministic = True

TEXT = data.Field(tokenize = 'spacy',
                  tokenizer_language = 'en_core_web_sm',
                  batch_first = True)
LABEL = data.LabelField(dtype = torch.float)

# 切分数据集
train_data, test_data = datasets.IMDB.splits(TEXT, LABEL)
train_data, valid_data = train_data.split(random_state = random.seed(SEED))

# 构建vocab，加载预训练词嵌入
MAX_VOCAB_SIZE = 25000
TEXT.build_vocab(train_data,
                 max_size = MAX_VOCAB_SIZE,
                 vectors = "glove.6B.100d",
                 unk_init = torch.Tensor.normal_)
LABEL.build_vocab(train_data)

# 创建迭代器
BATCH_SIZE = 64
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
train_iterator, valid_iterator, test_iterator = data.BucketIterator.splits(
    (train_data, valid_data, test_data),
    batch_size = BATCH_SIZE,
    device = device)

import torch.nn as nn
import torch.nn.functional as F

# 1.我们借助 ‘nn.Conv2d’实现卷积层。‘in_channels’参数是图像中进入卷积层的“通道”数。
# 在实际图像中，通常有3个通道（红色、蓝色和绿色通道各有一个通道），但是当使用文本时，我们只有一个通道，即文本本身。
# ‘out_channels’是 filters 的数量，‘kernel_size’是 filters 的大小。我们的每个“卷积核大小”都将是 [n*emb_dim]，
# 其中n是n-grams的大小。

# 2.之后，我们通过卷积层和池层传递张量，在卷积层之后使用'ReLU'激活函数。
# 池化层的另一个很好的特性是它们可以处理不同长度的句子。而卷积层的输出大小取决于输入的大小，不同的批次包含不同长度的句子。
# 如果没有最大池层，线性层的输入将取决于输入语句的长度，
# 为了避免这种情况，我们将所有句子修剪/填充到相同的长度，但是线性层来说，线性层的输入一直都是filter的总数。

# **注**：如果句子的长度小于实验设置的最大filter，那么必须将句子填充到最大filter的长度。在IMDb数据中不会存在这种情况，所以我们不必担心。

# 3.最后，我们对合并之后的filter输出执行dropout操作，然后将它们通过线性层进行预测。
class CNN(nn.Module):
    def __init__(self, vocab_size, embedding_dim, n_filters, filter_sizes, output_dim,
                 dropout, pad_idx):
        super().__init__()

self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx)

# self.conv_0 = nn.Conv2d(in_channels=1,
        #                         out_channels=n_filters,
        #                         kernel_size=(filter_sizes[0], embedding_dim))
        #
        # self.conv_1 = nn.Conv2d(in_channels=1,
        #                         out_channels=n_filters,
        #                         kernel_size=(filter_sizes[1], embedding_dim))
        #
        # self.conv_2 = nn.Conv2d(in_channels=1,
        #                         out_channels=n_filters,
        #                         kernel_size=(filter_sizes[2], embedding_dim))
        # 上述，`CNN` 模型使用了3个不同大小的filters，但我们实际上可以改进我们模型的代码，使其更通用，并且可以使用任意数量的filters。
        self.convs = nn.ModuleList([
            nn.Conv2d(in_channels=1,
                      out_channels=n_filters,
                      kernel_size=(fs, embedding_dim))
            for fs in filter_sizes
        ])

self.fc = nn.Linear(len(filter_sizes) * n_filters, output_dim)

self.dropout = nn.Dropout(dropout)

def forward(self, text):
        # text = [batch size, sent len]

embedded = self.embedding(text)

# embedded = [batch size, sent len, emb dim]

embedded = embedded.unsqueeze(1)

# embedded = [batch size, 1, sent len, emb dim]

# conved_0 = F.relu(self.conv_0(embedded).squeeze(3))
        # conved_1 = F.relu(self.conv_1(embedded).squeeze(3))
        # conved_2 = F.relu(self.conv_2(embedded).squeeze(3))
        # 上述，`CNN` 模型使用了3个不同大小的filters，但我们实际上可以改进我们模型的代码，使其更通用，并且可以使用任意数量的filters。
        conved = [F.relu(conv(embedded)).squeeze(3) for conv in self.convs]

# conved_n = [batch size, n_filters, sent len - filter_sizes[n] + 1]

# pooled_0 = F.max_pool1d(conved_0, conved_0.shape[2]).squeeze(2)
        # pooled_1 = F.max_pool1d(conved_1, conved_1.shape[2]).squeeze(2)
        # pooled_2 = F.max_pool1d(conved_2, conved_2.shape[2]).squeeze(2)
        # 上述，`CNN` 模型使用了3个不同大小的filters，但我们实际上可以改进我们模型的代码，使其更通用，并且可以使用任意数量的filters。
        pooled = [F.max_pool1d(conv, conv.shape[2]).squeeze(2) for conv in conved]

# pooled_n = [batch size, n_filters]

cat = self.dropout(torch.cat(pooled, dim=1))

# cat = [batch size, n_filters * len(filter_sizes)]

return self.fc(cat)

# 创建了`CNN` 类的一个实例。
INPUT_DIM = len(TEXT.vocab)
EMBEDDING_DIM = 100
N_FILTERS = 100
FILTER_SIZES = [3,4,5]
OUTPUT_DIM = 1
DROPOUT = 0.5
PAD_IDX = TEXT.vocab.stoi[TEXT.pad_token]

model = CNN(INPUT_DIM, EMBEDDING_DIM, N_FILTERS, FILTER_SIZES, OUTPUT_DIM, DROPOUT, PAD_IDX)

# 检查我们模型中的参数数量，我们可以看到它与FastText模型大致相同。
def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f'The model has {count_parameters(model):,} trainable parameters')

# 接下来，加载预训练词嵌入
pretrained_embeddings = TEXT.vocab.vectors
model.embedding.weight.data.copy_(pretrained_embeddings)

# 然后，将未知标记和填充标记的初始权重归零。
UNK_IDX = TEXT.vocab.stoi[TEXT.unk_token]
model.embedding.weight.data[UNK_IDX] = torch.zeros(EMBEDDING_DIM)
model.embedding.weight.data[PAD_IDX] = torch.zeros(EMBEDDING_DIM)

# 训练模型

import torch.optim as optim

# 训练和前面task一样，我们初始化优化器、损失函数，并将模型和损失函数放置在CPU上。
optimizer = optim.Adam(model.parameters())
criterion = nn.BCEWithLogitsLoss()
model = model.to(device)
criterion = criterion.to(device)

# 实现了计算精度的函数：
def binary_accuracy(preds, y):
    """
    Returns accuracy per batch, i.e. if you get 8/10 right, this returns 0.8, NOT 8
    """
    #round predictions to the closest integer
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float() #convert into float for division
    acc = correct.sum() / len(correct)
    return acc

# 定义了一个函数来训练我们的模型：
# **注意**：由于再次使用dropout，我们必须记住使用 `model.train()`以确保在训练时能够使用 dropout
def train(model, iterator, optimizer, criterion):
    epoch_loss = 0
    epoch_acc = 0

model.train()

for batch in iterator:
        optimizer.zero_grad()

predictions = model(batch.text).squeeze(1)

loss = criterion(predictions, batch.label)

acc = binary_accuracy(predictions, batch.label)

loss.backward()

optimizer.step()

epoch_loss += loss.item()
        epoch_acc += acc.item()

return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 定义了一个函数来测试我们的模型：
# **注意**：同样，由于使用的是dropout，我们必须记住使用`model.eval（）`来确保在评估时能够关闭 dropout。
def evaluate(model, iterator, criterion):
    epoch_loss = 0
    epoch_acc = 0

model.eval()

with torch.no_grad():
        for batch in iterator:
            predictions = model(batch.text).squeeze(1)

loss = criterion(predictions, batch.label)

acc = binary_accuracy(predictions, batch.label)

epoch_loss += loss.item()
            epoch_acc += acc.item()

return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 通过函数得到一个epoch需要多长时间：
import time
def epoch_time(start_time, end_time):
    elapsed_time = end_time - start_time
    elapsed_mins = int(elapsed_time / 60)
    elapsed_secs = int(elapsed_time - (elapsed_mins * 60))
    return elapsed_mins, elapsed_secs

# 最后，训练我们的模型：
N_EPOCHS = 1
best_valid_loss = float('inf')
for epoch in range(N_EPOCHS):

start_time = time.time()

train_loss, train_acc = train(model, train_iterator, optimizer, criterion)
    valid_loss, valid_acc = evaluate(model, valid_iterator, criterion)

end_time = time.time()

epoch_mins, epoch_secs = epoch_time(start_time, end_time)

if valid_loss < best_valid_loss:
        best_valid_loss = valid_loss
        torch.save(model.state_dict(), 'tut4-model.pt')

print(f'Epoch: {epoch + 1:02} | Epoch Time: {epoch_mins}m {epoch_secs}s')
    print(f'tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc * 100:.2f}%')
    print(f't Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc * 100:.2f}%')

# 保存模型，在测试集上看我们的模型损失和精度，我们得到的测试结果与前2个模型结果差不多！
model.load_state_dict(torch.load('tut4-model.pt'))
test_loss, test_acc = evaluate(model, test_iterator, criterion)
print(f'Test Loss: {test_loss:.3f} | Test Acc: {test_acc*100:.2f}%')

# 模型的验证
import spacy
nlp = spacy.load('en_core_web_sm')

def predict_sentiment(model, sentence, min_len = 5):
    model.eval()
    tokenized = [tok.text for tok in nlp.tokenizer(sentence)]
    if len(tokenized) < min_len:
        tokenized += ['<pad>'] * (min_len - len(tokenized))
    indexed = [TEXT.vocab.stoi[t] for t in tokenized]
    tensor = torch.LongTensor(indexed).to(device)
    tensor = tensor.unsqueeze(0)
    prediction = torch.sigmoid(model(tensor))
    return prediction.item()

# 负面评论的例子
predict_sentiment(model, "This film is terrible")
# 正面评论的例子
predict_sentiment(model, "This film is great")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
# 导入所需要的包
import torch
from torchtext.legacy import data
from torchtext.legacy import datasets
import random
import numpy as np

# 数据预处理
SEED = 1234
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
torch.backends.deterministic = True

TEXT = data.Field(tokenize = 'spacy',
                  tokenizer_language = 'en_core_web_sm',
                  batch_first = True)
LABEL = data.LabelField(dtype = torch.float)

# 切分数据集
train_data, test_data = datasets.IMDB.splits(TEXT, LABEL)
train_data, valid_data = train_data.split(random_state = random.seed(SEED))

# 构建vocab，加载预训练词嵌入
MAX_VOCAB_SIZE = 25000
TEXT.build_vocab(train_data,
                 max_size = MAX_VOCAB_SIZE,
                 vectors = "glove.6B.100d",
                 unk_init = torch.Tensor.normal_)
LABEL.build_vocab(train_data)

# 创建迭代器
BATCH_SIZE = 64
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
train_iterator, valid_iterator, test_iterator = data.BucketIterator.splits(
    (train_data, valid_data, test_data),
    batch_size = BATCH_SIZE,
    device = device)

import torch.nn as nn
import torch.nn.functional as F

# 1.我们借助 ‘nn.Conv2d’实现卷积层。‘in_channels’参数是图像中进入卷积层的“通道”数。
# 在实际图像中，通常有3个通道（红色、蓝色和绿色通道各有一个通道），但是当使用文本时，我们只有一个通道，即文本本身。
# ‘out_channels’是 filters 的数量，‘kernel_size’是 filters 的大小。我们的每个“卷积核大小”都将是 [n*emb_dim]，
# 其中n是n-grams的大小。

# 2.之后，我们通过卷积层和池层传递张量，在卷积层之后使用'ReLU'激活函数。
# 池化层的另一个很好的特性是它们可以处理不同长度的句子。而卷积层的输出大小取决于输入的大小，不同的批次包含不同长度的句子。
# 如果没有最大池层，线性层的输入将取决于输入语句的长度，
# 为了避免这种情况，我们将所有句子修剪/填充到相同的长度，但是线性层来说，线性层的输入一直都是filter的总数。

# **注**：如果句子的长度小于实验设置的最大filter，那么必须将句子填充到最大filter的长度。在IMDb数据中不会存在这种情况，所以我们不必担心。

# 3.最后，我们对合并之后的filter输出执行dropout操作，然后将它们通过线性层进行预测。
class CNN(nn.Module):
    def __init__(self, vocab_size, embedding_dim, n_filters, filter_sizes, output_dim,
                 dropout, pad_idx):
        super().__init__()

        self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx)

        # self.conv_0 = nn.Conv2d(in_channels=1,
        #                         out_channels=n_filters,
        #                         kernel_size=(filter_sizes[0], embedding_dim))
        #
        # self.conv_1 = nn.Conv2d(in_channels=1,
        #                         out_channels=n_filters,
        #                         kernel_size=(filter_sizes[1], embedding_dim))
        #
        # self.conv_2 = nn.Conv2d(in_channels=1,
        #                         out_channels=n_filters,
        #                         kernel_size=(filter_sizes[2], embedding_dim))
        # 上述，`CNN` 模型使用了3个不同大小的filters，但我们实际上可以改进我们模型的代码，使其更通用，并且可以使用任意数量的filters。
        self.convs = nn.ModuleList([
            nn.Conv2d(in_channels=1,
                      out_channels=n_filters,
                      kernel_size=(fs, embedding_dim))
            for fs in filter_sizes
        ])

        self.fc = nn.Linear(len(filter_sizes) * n_filters, output_dim)

        self.dropout = nn.Dropout(dropout)

    def forward(self, text):
        # text = [batch size, sent len]

        embedded = self.embedding(text)

        # embedded = [batch size, sent len, emb dim]

        embedded = embedded.unsqueeze(1)

        # embedded = [batch size, 1, sent len, emb dim]

        # conved_0 = F.relu(self.conv_0(embedded).squeeze(3))
        # conved_1 = F.relu(self.conv_1(embedded).squeeze(3))
        # conved_2 = F.relu(self.conv_2(embedded).squeeze(3))
        # 上述，`CNN` 模型使用了3个不同大小的filters，但我们实际上可以改进我们模型的代码，使其更通用，并且可以使用任意数量的filters。
        conved = [F.relu(conv(embedded)).squeeze(3) for conv in self.convs]

        # conved_n = [batch size, n_filters, sent len - filter_sizes[n] + 1]

        # pooled_0 = F.max_pool1d(conved_0, conved_0.shape[2]).squeeze(2)
        # pooled_1 = F.max_pool1d(conved_1, conved_1.shape[2]).squeeze(2)
        # pooled_2 = F.max_pool1d(conved_2, conved_2.shape[2]).squeeze(2)
        # 上述，`CNN` 模型使用了3个不同大小的filters，但我们实际上可以改进我们模型的代码，使其更通用，并且可以使用任意数量的filters。
        pooled = [F.max_pool1d(conv, conv.shape[2]).squeeze(2) for conv in conved]

        # pooled_n = [batch size, n_filters]

        cat = self.dropout(torch.cat(pooled, dim=1))

        # cat = [batch size, n_filters * len(filter_sizes)]

        return self.fc(cat)

# 创建了`CNN` 类的一个实例。
INPUT_DIM = len(TEXT.vocab)
EMBEDDING_DIM = 100
N_FILTERS = 100
FILTER_SIZES = [3,4,5]
OUTPUT_DIM = 1
DROPOUT = 0.5
PAD_IDX = TEXT.vocab.stoi[TEXT.pad_token]

model = CNN(INPUT_DIM, EMBEDDING_DIM, N_FILTERS, FILTER_SIZES, OUTPUT_DIM, DROPOUT, PAD_IDX)

# 检查我们模型中的参数数量，我们可以看到它与FastText模型大致相同。
def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f'The model has {count_parameters(model):,} trainable parameters')

# 接下来，加载预训练词嵌入
pretrained_embeddings = TEXT.vocab.vectors
model.embedding.weight.data.copy_(pretrained_embeddings)

# 然后，将未知标记和填充标记的初始权重归零。
UNK_IDX = TEXT.vocab.stoi[TEXT.unk_token]
model.embedding.weight.data[UNK_IDX] = torch.zeros(EMBEDDING_DIM)
model.embedding.weight.data[PAD_IDX] = torch.zeros(EMBEDDING_DIM)

# 训练模型

import torch.optim as optim

# 训练和前面task一样，我们初始化优化器、损失函数，并将模型和损失函数放置在CPU上。
optimizer = optim.Adam(model.parameters())
criterion = nn.BCEWithLogitsLoss()
model = model.to(device)
criterion = criterion.to(device)

# 实现了计算精度的函数：
def binary_accuracy(preds, y):
    """
    Returns accuracy per batch, i.e. if you get 8/10 right, this returns 0.8, NOT 8
    """
    #round predictions to the closest integer
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float() #convert into float for division
    acc = correct.sum() / len(correct)
    return acc

# 定义了一个函数来训练我们的模型：
# **注意**：由于再次使用dropout，我们必须记住使用 `model.train()`以确保在训练时能够使用 dropout
def train(model, iterator, optimizer, criterion):
    epoch_loss = 0
    epoch_acc = 0

    model.train()

    for batch in iterator:
        optimizer.zero_grad()

        predictions = model(batch.text).squeeze(1)

        loss = criterion(predictions, batch.label)

        acc = binary_accuracy(predictions, batch.label)

        loss.backward()

        optimizer.step()

        epoch_loss += loss.item()
        epoch_acc += acc.item()

    return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 定义了一个函数来测试我们的模型：
# **注意**：同样，由于使用的是dropout，我们必须记住使用`model.eval（）`来确保在评估时能够关闭 dropout。
def evaluate(model, iterator, criterion):
    epoch_loss = 0
    epoch_acc = 0

    model.eval()

    with torch.no_grad():
        for batch in iterator:
            predictions = model(batch.text).squeeze(1)

            loss = criterion(predictions, batch.label)

            acc = binary_accuracy(predictions, batch.label)

            epoch_loss += loss.item()
            epoch_acc += acc.item()

    return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 通过函数得到一个epoch需要多长时间：
import time
def epoch_time(start_time, end_time):
    elapsed_time = end_time - start_time
    elapsed_mins = int(elapsed_time / 60)
    elapsed_secs = int(elapsed_time - (elapsed_mins * 60))
    return elapsed_mins, elapsed_secs

# 最后，训练我们的模型：
N_EPOCHS = 1
best_valid_loss = float('inf')
for epoch in range(N_EPOCHS):

    start_time = time.time()

    train_loss, train_acc = train(model, train_iterator, optimizer, criterion)
    valid_loss, valid_acc = evaluate(model, valid_iterator, criterion)

    end_time = time.time()

    epoch_mins, epoch_secs = epoch_time(start_time, end_time)

    if valid_loss < best_valid_loss:
        best_valid_loss = valid_loss
        torch.save(model.state_dict(), 'tut4-model.pt')

    print(f'Epoch: {epoch + 1:02} | Epoch Time: {epoch_mins}m {epoch_secs}s')
    print(f'tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc * 100:.2f}%')
    print(f't Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc * 100:.2f}%')

# 保存模型，在测试集上看我们的模型损失和精度，我们得到的测试结果与前2个模型结果差不多！
model.load_state_dict(torch.load('tut4-model.pt'))
test_loss, test_acc = evaluate(model, test_iterator, criterion)
print(f'Test Loss: {test_loss:.3f} | Test Acc: {test_acc*100:.2f}%')


# 模型的验证
import spacy
nlp = spacy.load('en_core_web_sm')

def predict_sentiment(model, sentence, min_len = 5):
    model.eval()
    tokenized = [tok.text for tok in nlp.tokenizer(sentence)]
    if len(tokenized) < min_len:
        tokenized += ['<pad>'] * (min_len - len(tokenized))
    indexed = [TEXT.vocab.stoi[t] for t in tokenized]
    tensor = torch.LongTensor(indexed).to(device)
    tensor = tensor.unsqueeze(0)
    prediction = torch.sigmoid(model(tensor))
    return prediction.item()

# 负面评论的例子
predict_sentiment(model, "This film is terrible")
# 正面评论的例子
predict_sentiment(model, "This film is great")