Palind's Blog
0%

复旦NLP-Beginner任务二

发表于 更新于 Disqus：
本文字数： 10k 阅读时长 ≈ 9 分钟

任务：FudanNLP/nlp-beginner 的任务二。

任务二：基于深度学习的文本分类
熟悉Pytorch，用Pytorch重写《任务一》，实现CNN、RNN的文本分类；

  1. 参考

    https://pytorch.org/

    Convolutional Neural Networks for Sentence Classification https://arxiv.org/abs/1408.5882

    https://machinelearningmastery.com/sequence-classification-lstm-recurrent-neural-networks-python-keras/

  2. word embedding 的方式初始化

  3. 随机embedding的初始化方式

  4. 用glove 预训练的embedding进行初始化 https://nlp.stanford.edu/projects/glove/

  5. 知识点：

    CNN/RNN的特征抽取

    词嵌入

    Dropout

  6. 时间：两周

为了复习巩固一遍 PyTorch 和 pandas，我先写了词袋（BoW），如下：

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
import torch
import pandas as pd
from collections import Counter

# 1. 读取训练集
train = pd.read_csv("train.tsv", sep='\t')

# 2. 分词
train['tokens'] = train['Phrase'].str.lower().str.split()

# 3. 建立词表
all_words = [word for tokens in train['tokens'] for word in tokens]
vocab = {word: i for i, (word, _) in enumerate(Counter(all_words).most_common())}
vocab_size = len(vocab)

# 4. Bag-of-Words 向量化
def bow_vector(tokens):
    vec = torch.zeros(vocab_size)
    for token in tokens:
        if token in vocab:
            vec[vocab[token]] += 1
    return vec


# 5. 把所有句子转成 BoW 向量
# X = torch.stack([bow_vector(tokens) for tokens in train['tokens']])  # shape: [样本数, 词表大小]
# 也可以加个进度条
from tqdm import tqdm
X = torch.stack([bow_vector(tokens) for tokens in tqdm(train['tokens'], desc="把句子转为 BoW 向量")])


# 6. 标签转 Tensor
y = torch.tensor(train['Sentiment'].values)  # shape: [样本数]

print("X 的形状:", X.shape)
print("y 的形状:", y.shape)

# 以下是训练循环

import torch.nn as nn
import torch.optim as optim

model = nn.Linear(vocab_size, 5)

criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.1)

batch_size = 64
num_epochs = 25
X_train = X
y_train = y

for epoch in range(num_epochs):
    for i in range(0, len(X_train), batch_size):
        X_batch = X_train[i:i+batch_size]
        y_batch = y_train[i:i+batch_size]

        # forward
        outputs = model(X_batch)

        # loss
        loss = criterion(outputs, y_batch)

        # backward and update
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

    print(f"Epoch {epoch+1}/{num_epochs}, Loss: {loss.item():.4f}")

test = pd.read_csv("test.tsv", sep='\t')
test['tokens'] = test['Phrase'].str.lower().str.split()

X_test = torch.stack([bow_vector(tokens) for tokens in test['tokens']])

with torch.no_grad():
    outputs = model(X_test)
    predictions = torch.argmax(outputs, dim=1)

predictions = predictions.numpy()

submission = pd.DataFrame({
    "PhraseId": test["PhraseId"],
    "Sentiment": predictions
})

submission.to_csv("submission.csv", index=False)
print("结束")

得分还可以：

1

从任务中，我第一次知道 CNN 还能运用在文本分类上，神奇，就是先词嵌入为向量，再对相邻的进行卷积。现在来简单实现一下。

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
import torch
import torch.nn as nn
import torch.optim as optim
import pandas as pd
from collections import Counter
from torch.nn.utils.rnn import pad_sequence

train = pd.read_csv("train.tsv", sep='\t')
train['tokens'] = train['Phrase'].str.lower().str.split()

all_words = [word for tokens in train['tokens'] for word in tokens]
vocab = {word: i+1 for i, (word, _) in enumerate(Counter(all_words).most_common())}
vocab_size = len(vocab) + 1
# index starts from one
# zero is for padding

def text_to_sequence(tokens, vocab):
    return [vocab.get(token, 0) for token in tokens]

from tqdm import tqdm
sequences = [torch.tensor(text_to_sequence(tokens, vocab)) for tokens in tqdm(train['tokens'], desc="转换中")]
# what is sequences like?
# [[1, 2, 3, ...], [3, 2, 1, ...], ...]
# num = word, tensor([1, 2, 3]) = sentence

X_padded = pad_sequence(sequences, batch_first=True, padding_value=0)
y = torch.tensor(train['Sentiment'].values)

class TextCNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, num_classes):
        super(TextCNN, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.conv = nn.Conv1d(in_channels=embed_dim, out_channels=100, kernel_size=3)
        self.relu = nn.ReLU()
        self.pool = nn.AdaptiveMaxPool1d(1)
        self.fc = nn.Linear(100, num_classes)

    def forward(self, x):
        x = self.embedding(x)           # [batch, seq_len, embed_dim]
        x = x.permute(0, 2, 1)          # [batch, embed_dim, seq_len]
        x = self.conv(x)                # [batch, 100, new_seq_len = seq_len - kernel_size + 1]
        x = self.relu(x)                # activate
        x = self.pool(x).squeeze(-1)    # [batch, 100]
        x = self.fc(x)                  # [batch, num_classes]
        return x

embed_dim = 50
num_classes = 5
model = TextCNN(vocab_size, embed_dim, num_classes)

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

batch_size = 64
num_epochs = 25

for epoch in range(num_epochs):
    permutation = torch.randperm(X_padded.size(0))
    epoch_loss = 0

    for i in range(0, X_padded.size(0), batch_size):
        indices = permutation[i:i+batch_size]
        batch_x, batch_y = X_padded[indices], y[indices]

        optimizer.zero_grad()
        outputs = model(batch_x)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()

        epoch_loss += loss.item()

    print(f"Epoch {epoch+1}/{num_epochs}, Loss: {epoch_loss:.4f}")

# 读取测试集
test = pd.read_csv("test.tsv", sep='\t')
test['tokens'] = test['Phrase'].str.lower().str.split()

# 转成索引序列
test_sequences = [torch.tensor(text_to_sequence(tokens, vocab)) for tokens in test['tokens']]

# 填充长度
X_test_padded = pad_sequence(test_sequences, batch_first=True, padding_value=0)

model.eval()  # 切换到评估模式
with torch.no_grad():
    outputs = model(X_test_padded)
    predictions = torch.argmax(outputs, dim=1)  # 取最大概率的类别

submission = pd.DataFrame({
    "PhraseId": test["PhraseId"],
    "Sentiment": predictions.numpy()
})

submission.to_csv("submission_cnn.csv", index=False)
print("结束")

结果epoch设小了，没收敛，

1
2
3
Epoch 23/25, Loss: 842.2148
Epoch 24/25, Loss: 814.8164
Epoch 25/25, Loss: 787.9565

先交再说，

2

成绩上升了。

最后看下RNN，

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
import torch
import torch.nn as nn
import torch.optim as optim
import pandas as pd
from collections import Counter
from torch.nn.utils.rnn import pad_sequence, pack_padded_sequence, pad_packed_sequence

# ===============================
# 1️⃣ 读取训练集并分词
# ===============================
train = pd.read_csv("train.tsv", sep='\t')
train['tokens'] = train['Phrase'].str.lower().str.split()

# ===============================
# 2️⃣ 建立词表
# ===============================
all_words = [word for tokens in train['tokens'] for word in tokens]
vocab = {word: i+1 for i, (word, _) in enumerate(Counter(all_words).most_common())}
vocab_size = len(vocab) + 1  # 0 用于 padding

# ===============================
# 3️⃣ 句子转索引 + 记录长度
# ===============================
def text_to_sequence(tokens, vocab):
    return [vocab.get(token, 0) for token in tokens]
# sequences = [torch.tensor(text_to_sequence(tokens, vocab), dtype=torch.long) for tokens in train['tokens']]
# lengths = torch.tensor([len(seq) for seq in sequences])


# ===============================
# 4️⃣ 填充句子
# ===============================
# 上面的处理有问题
sequences = []
lengths = []
y_list = []

for i, tokens in enumerate(train['tokens']):
    seq = torch.tensor(text_to_sequence(tokens, vocab), dtype=torch.long)
    if len(seq) > 0:  # 只保留长度 > 0 的句子
        sequences.append(seq)
        lengths.append(len(seq))
        y_list.append(train['Sentiment'][i])

# 转成 Tensor 并填充
X_padded = pad_sequence(sequences, batch_first=True, padding_value=0)
lengths = torch.tensor(lengths)  # 现在一次性转换为 Tensor
y = torch.tensor(y_list)


# ===============================
# 5️⃣ 定义 LSTM 文本分类模型
# ===============================
class TextLSTM(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_classes):
        super(TextLSTM, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, batch_first=True)
        self.fc = nn.Linear(hidden_dim, num_classes)

    def forward(self, x, lengths):
        # x: [batch, seq_len]
        embed = self.embedding(x)  # [batch, seq_len, embed_dim]

        # pack sequence，告诉 LSTM 哪些是 padding
        packed = pack_padded_sequence(embed, lengths.cpu(), batch_first=True, enforce_sorted=False)
        packed_out, (h_n, c_n) = self.lstm(packed)
        # h_n: [num_layers * num_directions, batch, hidden_dim]

        out = self.fc(h_n[-1])  # 用最后一层的 hidden state 做分类
        return out

# ===============================
# 6️⃣ 初始化模型、损失函数、优化器
# ===============================
embed_dim = 50
hidden_dim = 100
num_classes = 5
model = TextLSTM(vocab_size, embed_dim, hidden_dim, num_classes)

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# ===============================
# 7️⃣ 训练
# ===============================
batch_size = 64
num_epochs = 30

for epoch in range(num_epochs):
    permutation = torch.randperm(X_padded.size(0))
    epoch_loss = 0

    for i in range(0, X_padded.size(0), batch_size):
        indices = permutation[i:i+batch_size]
        batch_x = X_padded[indices]
        batch_y = y[indices]
        batch_lengths = lengths[indices]

        optimizer.zero_grad()
        outputs = model(batch_x, batch_lengths)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()

        epoch_loss += loss.item()

    print(f"Epoch {epoch+1}/{num_epochs}, Loss: {epoch_loss:.4f}")

# ===============================
# 8️⃣ 读取测试集并处理
# ===============================
test = pd.read_csv("test.tsv", sep='\t')
test['tokens'] = test['Phrase'].str.lower().str.split()
test_sequences = [torch.tensor(text_to_sequence(tokens, vocab), dtype=torch.long) for tokens in test['tokens']]
test_lengths = torch.tensor([len(seq) for seq in test_sequences])
X_test_padded = pad_sequence(test_sequences, batch_first=True, padding_value=0)

test_sequences = []
test_lengths = []
test_indices = []  # 记录原始索引，方便最后恢复顺序

for i, tokens in enumerate(test['tokens']):
    seq = torch.tensor(text_to_sequence(tokens, vocab), dtype=torch.long)
    if len(seq) > 0:
        test_sequences.append(seq)
        test_lengths.append(len(seq))
        test_indices.append(i)

X_test_padded = pad_sequence(test_sequences, batch_first=True, padding_value=0)
test_lengths = torch.tensor(test_lengths)


# ===============================
# 9️⃣ 预测
# ===============================
model.eval()
with torch.no_grad():
    outputs = model(X_test_padded, test_lengths)
    predictions = torch.argmax(outputs, dim=1)

all_predictions = torch.zeros(len(test), dtype=torch.long)  # 默认类别 0
for idx, pred in zip(test_indices, predictions):
    all_predictions[idx] = pred

submission = pd.DataFrame({
    "PhraseId": test["PhraseId"],
    "Sentiment": all_predictions.numpy()  # 这里使用 all_predictions
})
submission.to_csv("submission_lstm.csv", index=False)
print("结束")

最终结果是：

3

和 CNN 对比一下，分数反而不高。