外卖评论情感分类

1
import torch
2
from torch.utils.data import Dataset, DataLoader, random_split
3
import jieba
4
import pickle
5
import os
6
import pandas as pd
7

8

9
class WaimaiDataset(Dataset):
10

11
    def __init__(self, texts, labels, vocab=None, max_len=128, build_vocab=False):
12
        self.texts = texts
13
        self.labels = labels
14
        self.max_len = max_len
15

16
        # build_vocab 为 True 时在训练集上构建词表，否则复用传入的词表
17
        if build_vocab:
18
            self.vocab = self._build_vocab(texts)
19
        else:
20
            self.vocab = vocab
21

22
        self.word2idx = self.vocab if vocab else {}
23

24
    def _build_vocab(self, texts, min_freq=2):
25
        from collections import Counter
26

27
        word_counter = Counter()
28

29
        # jieba 分词并统计词频，空值统一转为空字符串
30
        for text in texts:
31
            text = str(text) if not pd.isna(text) else ""
32
            words = jieba.lcut(text)
33
            word_counter.update(words)
34

35
        # 0 保留给 <PAD>，1 保留给 <UNK>，其余词按最低词频过滤后递增编号
36
        vocab = {'<PAD>': 0, '<UNK>': 1}
37
        idx = 2
38

39
        for word, freq in word_counter.items():
40
            if freq >= min_freq and word not in vocab:
41
                vocab[word] = idx
42
                idx += 1
43

44
        print(f"词表大小: {len(vocab)} (高频词>={min_freq})")
45
        return vocab
46

47
    def _text_to_sequence(self, text):
48
        text = str(text) if not pd.isna(text) else ""
49
        words = jieba.lcut(text)
50

51
        # 词不在词表中时映射为 <UNK>(1)
52
        sequence = [self.word2idx.get(w, 1) for w in words]
53

54
        # 定长填充/截断：不足 max_len 末尾补 0（<PAD>），超出则截断
55
        if len(sequence) < self.max_len:
56
            sequence += [0] * (self.max_len - len(sequence))
57
        else:
58
            sequence = sequence[:self.max_len]
59

60
        return sequence
61

62
    def __len__(self):
63
        return len(self.texts)
64

65
    def __getitem__(self, idx):
66
        text = self.texts[idx]
67
        label = self.labels[idx]
68

69
        sequence = self._text_to_sequence(text)
70
        return torch.tensor(sequence, dtype=torch.long), torch.tensor(label, dtype=torch.long)
71

72

73
class WaimaiDataProcessor:
74

75
    def __init__(self, data_path='./data/waimai_10k.csv', batch_size=64, max_len=128):
76
        self.data_path = data_path
77
        self.batch_size = batch_size
78
        self.max_len = max_len
79

80
    def load_data(self):
81
        if not os.path.exists(self.data_path):
82
            raise FileNotFoundError(
83
                f"\n{'='*60}\n"
84
                f"错误：找不到数据文件！\n"
85
                f"{'='*60}\n"
86
                f"请运行: python download_waimai_10k.py\n"
87
                f"或从以下链接下载：\n"
88
                f"  https://raw.githubusercontent.com/SophonPlus/ChineseNlpCorpus/master/datasets/waimai_10k/waimai_10k.csv\n"
89
                f"{'='*60}"
90
            )
91

92
        df = pd.read_csv(self.data_path)
93
        print(f"加载数据: {len(df)} 条")
94
        print(f"列名: {df.columns.tolist()}")
95
        print(f"标签分布:\n{df['label'].value_counts().sort_index()}")
96

97
        texts = df['review'].tolist()
98
        labels = df['label'].tolist()
99

100
        return texts, labels
101

102
    def get_loaders(self, val_ratio=0.15, test_ratio=0.15):
103
        texts, labels = self.load_data()
104

105
        # 按比例随机划分训练/验证/测试集
106
        total_size = len(texts)
107
        test_size = int(total_size * test_ratio)
108
        val_size = int(total_size * val_ratio)
109
        train_size = total_size - val_size - test_size
110

111
        indices = torch.randperm(total_size).tolist()
112

113
        train_idx = indices[:train_size]
114
        val_idx = indices[train_size:train_size+val_size]
115
        test_idx = indices[train_size+val_size:]
116

117
        train_texts = [texts[i] for i in train_idx]
118
        train_labels = [labels[i] for i in train_idx]
119

120
        # 词表只在训练集上构建，避免验证/测试数据信息泄露
121
        print("\n构建词表...")
122
        temp_dataset = WaimaiDataset(train_texts, train_labels, max_len=self.max_len, build_vocab=True)
123
        vocab = temp_dataset.vocab
124

125
        with open('./data/vocab.pkl', 'wb') as f:
126
            pickle.dump(vocab, f)
127

128
        # 三个数据集必须共用同一个词表，保证索引含义一致
129
        train_dataset = WaimaiDataset(train_texts, train_labels, vocab, self.max_len)
130
        val_dataset = WaimaiDataset([texts[i] for i in val_idx], [labels[i] for i in val_idx], vocab, self.max_len)
131
        test_dataset = WaimaiDataset([texts[i] for i in test_idx], [labels[i] for i in test_idx], vocab, self.max_len)
132

133
        # 训练集打乱顺序，验证/测试集保持顺序
134
        train_loader = DataLoader(train_dataset, batch_size=self.batch_size, shuffle=True, num_workers=0)
135
        val_loader = DataLoader(val_dataset, batch_size=self.batch_size, shuffle=False, num_workers=0)
136
        test_loader = DataLoader(test_dataset, batch_size=self.batch_size, shuffle=False, num_workers=0)
137

138
        print(f"\n数据划分:")
139
        print(f"  训练集: {len(train_dataset)} ({train_size})")
140
        print(f"  验证集: {len(val_dataset)} ({val_size})")
141
        print(f"  测试集: {len(test_dataset)} ({test_size})")
142

143
        return train_loader, val_loader, test_loader, vocab
144

145

146
if __name__ == '__main__':
147
    processor = WaimaiDataProcessor(batch_size=32)
148
    train_loader, val_loader, test_loader, vocab = processor.get_loaders()
149

150
    # 取出一个 batch 检查数据管道
151
    sequences, labels = next(iter(train_loader))
152
    print(f"\n测试batch:")
153
    print(f"  序列形状: {sequences.shape}")
154
    print(f"  标签形状: {labels.shape}")

1
import torch
2
import torch.nn as nn
3
import math
4

5

6
class TextRNN(nn.Module):
7

8
    def __init__(self, vocab_size, embed_dim=128, hidden_dim=256,
9
                 num_classes=2, num_layers=2, dropout=0.5, max_len=128):
10
        super(TextRNN, self).__init__()
11

12
        self.hidden_dim = hidden_dim
13
        self.num_layers = num_layers
14

15
        # padding_idx=0 使 <PAD> 位置的嵌入向量恒为 0
16
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
17

18
        # 双向 LSTM：batch_first 输入 [batch, seq, feature]，双向拼接后输出维度为 hidden*2
19
        self.lstm = nn.LSTM(
20
            input_size=embed_dim,
21
            hidden_size=hidden_dim,
22
            num_layers=num_layers,
23
            batch_first=True,
24
            bidirectional=True,
25
            dropout=dropout if num_layers > 1 else 0
26
        )
27

28
        # 分类头：输入维度为 hidden_dim * 2（双向拼接）
29
        self.dropout = nn.Dropout(dropout)
30
        self.fc = nn.Sequential(
31
            nn.Linear(hidden_dim * 2, hidden_dim),
32
            nn.ReLU(),
33
            nn.Dropout(dropout),
34
            nn.Linear(hidden_dim, num_classes)
35
        )
36

37
    def forward(self, x):
38
        # 词索引序列 -> 嵌入向量 [batch, seq_len, embed_dim]
39
        embedded = self.embedding(x)
40

41
        # LSTM
42
        lstm_out, (hidden, cell) = self.lstm(embedded)  # [batch, seq_len, hidden*2]
43

44
        # 掩码平均池化：屏蔽 <PAD> 位置，均值只按实际长度计算，clamp(min=1) 防止除零
45
        mask = (x != 0).float().unsqueeze(-1)  # [batch, seq_len, 1]
46
        sum_out = torch.sum(lstm_out * mask, dim=1)  # [batch, hidden*2]
47
        avg_out = sum_out / torch.sum(mask, dim=1).clamp(min=1)
48

49
        out = self.dropout(avg_out)
50
        out = self.fc(out)
51
        return out
52

53

54
class PositionalEncoding(nn.Module):
55

56
    def __init__(self, d_model, max_len=5000):
57
        super().__init__()
58

59
        # 正弦位置编码：偶数列用 sin、奇数列用 cos，可外推到训练时未见过的长度
60
        pe = torch.zeros(max_len, d_model)
61
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
62
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
63

64
        pe[:, 0::2] = torch.sin(position * div_term)
65
        pe[:, 1::2] = torch.cos(position * div_term)
66

67
        pe = pe.unsqueeze(0)
68
        self.register_buffer('pe', pe)
69

70
    def forward(self, x):
71
        return x + self.pe[:, :x.size(1)]
72

73

74
class TextTransformer(nn.Module):
75

76
    def __init__(self, vocab_size, embed_dim=128, num_heads=4,
77
                 num_layers=2, dim_feedforward=512, num_classes=2,
78
                 dropout=0.3, max_len=128):
79
        super(TextTransformer, self).__init__()
80

81
        self.embed_dim = embed_dim
82

83
        # 词嵌入 + 正弦位置编码
84
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
85
        self.pos_encoder = PositionalEncoding(embed_dim, max_len)
86

87
        # Transformer 编码器：堆叠 num_layers 层，num_heads 需能整除 embed_dim
88
        encoder_layer = nn.TransformerEncoderLayer(
89
            d_model=embed_dim,
90
            nhead=num_heads,
91
            dim_feedforward=dim_feedforward,
92
            dropout=dropout,
93
            batch_first=True
94
        )
95
        self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
96

97
        # 分类头
98
        self.dropout = nn.Dropout(dropout)
99
        self.fc = nn.Sequential(
100
            nn.Linear(embed_dim, dim_feedforward),
101
            nn.ReLU(),
102
            nn.Dropout(dropout),
103
            nn.Linear(dim_feedforward, num_classes)
104
        )
105

106
    def forward(self, x):
107
        # padding mask：True 表示 <PAD> 位置，自注意力会将其忽略
108
        src_key_padding_mask = (x == 0)  # [batch, seq_len]
109

110
        # 嵌入乘 sqrt(d_model) 平衡与位置编码的量级，再叠加位置编码
111
        embedded = self.embedding(x) * math.sqrt(self.embed_dim)
112
        embedded = self.pos_encoder(embedded)
113

114
        encoded = self.transformer_encoder(embedded, src_key_padding_mask=src_key_padding_mask)
115

116
        # 掩码平均池化：与 TextRNN 一致，只对非填充位置取平均
117
        mask = (~src_key_padding_mask).float().unsqueeze(-1)
118
        sum_encoded = torch.sum(encoded * mask, dim=1)
119
        avg_encoded = sum_encoded / torch.sum(mask, dim=1).clamp(min=1)
120

121
        out = self.dropout(avg_encoded)
122
        out = self.fc(out)
123
        return out
124

125

126
if __name__ == '__main__':
127
    vocab_size = 5000
128

129
    rnn = TextRNN(vocab_size)
130
    x = torch.randint(0, vocab_size, (4, 128))
131
    out = rnn(x)
132
    print(f"TextRNN: {out.shape}")
133

134
    trans = TextTransformer(vocab_size)
135
    out = trans(x)
136
    print(f"TextTransformer: {out.shape}")

1
import torch
2
import torch.nn as nn
3
import torch.optim as optim
4
from tqdm import tqdm
5
import json
6
import os
7
import matplotlib.pyplot as plt
8
from sklearn.metrics import confusion_matrix, classification_report
9
import numpy as np
10

11

12
class Trainer:
13

14
    def __init__(self, model, train_loader, val_loader, device,
15
                 lr=0.001, weight_decay=1e-4, model_name='model'):
16
        self.model = model.to(device)
17
        self.train_loader = train_loader
18
        self.val_loader = val_loader
19
        self.device = device
20
        self.model_name = model_name
21

22
        # 交叉熵损失 + Adam 优化器（weight_decay 为 L2 正则化系数）
23
        self.criterion = nn.CrossEntropyLoss()
24
        self.optimizer = optim.Adam(self.model.parameters(), lr=lr, weight_decay=weight_decay)
25

26
        # 验证准确率连续 3 轮不提升时学习率减半（PyTorch 2.7.1 无 verbose 参数）
27
        self.scheduler = optim.lr_scheduler.ReduceLROnPlateau(
28
            self.optimizer, mode='max', factor=0.5, patience=3
29
        )
30

31
        # 训练历史，用于绘制训练曲线
32
        self.history = {
33
            'train_loss': [], 'train_acc': [],
34
            'val_loss': [], 'val_acc': []
35
        }
36
        self.best_acc = 0.0
37

38
    def train_epoch(self):
39
        # train 模式启用 Dropout 的训练行为
40
        self.model.train()
41
        total_loss = 0.0
42
        correct = 0
43
        total = 0
44

45
        pbar = tqdm(self.train_loader, desc='Training')
46
        for inputs, labels in pbar:
47
            inputs, labels = inputs.to(self.device), labels.to(self.device)
48

49
            # 训练五步：梯度清零 -> 前向传播 -> 计算损失 -> 反向传播 -> 参数更新
50
            self.optimizer.zero_grad()
51
            outputs = self.model(inputs)
52
            loss = self.criterion(outputs, labels)
53
            loss.backward()
54
            self.optimizer.step()
55

56
            total_loss += loss.item()
57
            _, predicted = outputs.max(1)
58
            total += labels.size(0)
59
            correct += predicted.eq(labels).sum().item()
60

61
            pbar.set_postfix({
62
                'loss': f'{loss.item():.4f}',
63
                'acc': f'{100.*correct/total:.2f}%',
64
                'lr': f'{self.optimizer.param_groups[0]["lr"]:.6f}'
65
            })
66

67
        return total_loss / len(self.train_loader), 100. * correct / total
68

69
    def validate(self):
70
        # eval + no_grad：关闭 Dropout、不计算梯度
71
        self.model.eval()
72
        total_loss = 0.0
73
        correct = 0
74
        total = 0
75

76
        with torch.no_grad():
77
            for inputs, labels in tqdm(self.val_loader, desc='Validating'):
78
                inputs, labels = inputs.to(self.device), labels.to(self.device)
79
                outputs = self.model(inputs)
80
                loss = self.criterion(outputs, labels)
81

82
                total_loss += loss.item()
83
                _, predicted = outputs.max(1)
84
                total += labels.size(0)
85
                correct += predicted.eq(labels).sum().item()
86

87
        return total_loss / len(self.val_loader), 100. * correct / total
88

89
    def train(self, epochs=10, save_dir='./checkpoints'):
90
        os.makedirs(save_dir, exist_ok=True)
91
        best_path = os.path.join(save_dir, f'{self.model_name}_best.pth')
92

93
        print(f"\n训练 {self.model_name}")
94
        print(f"设备: {self.device}")
95
        print(f"初始LR: {self.optimizer.param_groups[0]['lr']}")
96

97
        for epoch in range(epochs):
98
            print(f"\nEpoch [{epoch+1}/{epochs}]")
99

100
            train_loss, train_acc = self.train_epoch()
101
            val_loss, val_acc = self.validate()
102

103
            self.history['train_loss'].append(train_loss)
104
            self.history['train_acc'].append(train_acc)
105
            self.history['val_loss'].append(val_loss)
106
            self.history['val_acc'].append(val_acc)
107

108
            # 调度器按验证准确率调整学习率，变化时手动打印
109
            old_lr = self.optimizer.param_groups[0]['lr']
110
            self.scheduler.step(val_acc)
111
            new_lr = self.optimizer.param_groups[0]['lr']
112
            if new_lr != old_lr:
113
                print(f"! LR调整: {old_lr:.6f} -> {new_lr:.6f}")
114

115
            print(f"Train: loss={train_loss:.4f}, acc={train_acc:.2f}%")
116
            print(f"Val:   loss={val_loss:.4f}, acc={val_acc:.2f}%")
117

118
            # 验证准确率创新高时保存当前权重
119
            if val_acc > self.best_acc:
120
                self.best_acc = val_acc
121
                torch.save(self.model.state_dict(), best_path)
122
                print(f"* 保存最佳模型: {val_acc:.2f}%")
123

124
        # 保存训练历史并绘制曲线
125
        with open(os.path.join(save_dir, f'{self.model_name}_history.json'), 'w') as f:
126
            json.dump(self.history, f, indent=2)
127

128
        self.plot_history(save_dir)
129
        return self.history
130

131
    def plot_history(self, save_dir):
132
        fig, axes = plt.subplots(1, 2, figsize=(12, 4))
133

134
        axes[0].plot(self.history['train_loss'], label='Train')
135
        axes[0].plot(self.history['val_loss'], label='Val')
136
        axes[0].set_title('Loss')
137
        axes[0].legend()
138
        axes[0].grid(True)
139

140
        axes[1].plot(self.history['train_acc'], label='Train')
141
        axes[1].plot(self.history['val_acc'], label='Val')
142
        axes[1].set_title('Accuracy')
143
        axes[1].legend()
144
        axes[1].grid(True)
145

146
        plt.tight_layout()
147
        plt.savefig(os.path.join(save_dir, f'{self.model_name}_curves.png'), dpi=150)
148
        plt.close()
149

150

151
def test(model, test_loader, device, model_path=None, num_classes=2, class_names=None):
152
    # 加载最优权重，map_location 保证跨设备加载
153
    if model_path is not None:
154
        model.load_state_dict(torch.load(model_path, map_location=device))
155
        print(f"加载模型: {model_path}")
156

157
    model = model.to(device)
158
    model.eval()
159

160
    all_preds = []
161
    all_labels = []
162
    correct = 0
163
    total = 0
164

165
    # 推理阶段不计算梯度，收集全部预测用于生成分类报告与混淆矩阵
166
    with torch.no_grad():
167
        for inputs, labels in tqdm(test_loader, desc='Testing'):
168
            inputs, labels = inputs.to(device), labels.to(device)
169
            outputs = model(inputs)
170
            _, predicted = outputs.max(1)
171

172
            total += labels.size(0)
173
            correct += predicted.eq(labels).sum().item()
174
            all_preds.extend(predicted.cpu().numpy())
175
            all_labels.extend(labels.cpu().numpy())
176

177
    accuracy = 100. * correct / total
178
    print(f"\n测试准确率: {accuracy:.2f}%")
179

180
    # 默认类别名：二分类为 负面/正面
181
    if class_names is None:
182
        class_names = ['负面', '正面'] if num_classes == 2 else [f'class_{i}' for i in range(num_classes)]
183

184
    print("\n分类报告:")
185
    print(classification_report(all_labels, all_preds, target_names=class_names, digits=4))
186

187
    # 混淆矩阵热力图，并在格中标注样本数
188
    cm = confusion_matrix(all_labels, all_preds)
189
    plt.figure(figsize=(6, 6))
190
    plt.imshow(cm, interpolation='nearest', cmap=plt.cm.Blues)
191
    plt.title('Confusion Matrix')
192
    plt.colorbar()
193
    tick_marks = np.arange(len(class_names))
194
    plt.xticks(tick_marks, class_names, rotation=45)
195
    plt.yticks(tick_marks, class_names)
196
    plt.ylabel('True Label')
197
    plt.xlabel('Predicted Label')
198

199
    for i in range(len(class_names)):
200
        for j in range(len(class_names)):
201
            plt.text(j, i, cm[i, j], ha="center", va="center",
202
                    color="white" if cm[i, j] > cm.max()/2 else "black")
203

204
    plt.tight_layout()
205
    plt.savefig('confusion_matrix.png', dpi=150)
206
    plt.close()
207

208
    return accuracy
209

210

211
if __name__ == '__main__':
212
    print("=" * 60)
213
    print("测试 train.py 模块")
214
    print("=" * 60)
215

216
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
217
    print(f"设备: {device}")
218

219

220
    # 用随机模拟数据验证 Trainer 与 test()，无需真实数据集
221
    vocab_size, seq_len, num_classes = 100, 20, 2
222
    batch_size, num_samples = 8, 64
223

224
    X = torch.randint(0, vocab_size, (num_samples, seq_len))
225
    y = torch.randint(0, num_classes, (num_samples,))
226

227
    dataset = torch.utils.data.TensorDataset(X, y)
228

229
    # 按 7:1.5:1.5 划分模拟数据集
230
    train_size = int(0.7 * num_samples)
231
    val_size = int(0.15 * num_samples)
232
    test_size = num_samples - train_size - val_size
233

234
    train_ds, val_ds, test_ds = torch.utils.data.random_split(
235
        dataset, [train_size, val_size, test_size]
236
    )
237

238
    train_loader = torch.utils.data.DataLoader(train_ds, batch_size=batch_size, shuffle=True)
239
    val_loader = torch.utils.data.DataLoader(val_ds, batch_size=batch_size, shuffle=False)
240
    test_loader = torch.utils.data.DataLoader(test_ds, batch_size=batch_size, shuffle=False)
241

242
    print(f"\n模拟数据：")
243
    print(f"  总样本: {num_samples}")
244
    print(f"  训练集: {train_size}, 验证集: {val_size}, 测试集: {test_size}")
245
    print(f"  序列长度: {seq_len}, 词表大小: {vocab_size}, 类别数: {num_classes}")
246

247

248
    # 结构极简的 DummyModel，用于快速验证训练闭环
249
    class DummyModel(nn.Module):
250
        def __init__(self, vocab_size, num_classes):
251
            super().__init__()
252
            self.embedding = nn.Embedding(vocab_size, 16, padding_idx=0)
253
            self.fc = nn.Linear(16, num_classes)
254

255
        def forward(self, x):
256
            embedded = self.embedding(x).mean(dim=1)
257
            return self.fc(embedded)
258

259
    model = DummyModel(vocab_size, num_classes)
260
    print(f"\n模拟模型参数量: {sum(p.numel() for p in model.parameters())}")
261

262

263
    trainer = Trainer(model, train_loader, val_loader, device,
264
                      lr=0.01, model_name='TestModel')
265
    print("\n开始训练 2 个 epoch...")
266
    history = trainer.train(epochs=2, save_dir='./test_checkpoints')
267

268
    # 用刚保存的权重验证 test() 函数
269
    print("\n" + "=" * 60)
270
    print("测试 test() 函数")
271
    print("=" * 60)
272
    test_acc = test(model, test_loader, device,
273
                    model_path='./test_checkpoints/TestModel_best.pth',
274
                    num_classes=num_classes,
275
                    class_names=['负面', '正面'])
276
    print(f"\n测试完成，最终准确率: {test_acc:.2f}%")
277

278
    # 清理测试产生的临时文件
279
    import shutil
280
    if os.path.exists('./test_checkpoints'):
281
        shutil.rmtree('./test_checkpoints')
282
        print("\n测试检查点已清理: ./test_checkpoints")
283
    if os.path.exists('confusion_matrix.png'):
284
        os.remove('confusion_matrix.png')
285
        print("测试混淆矩阵图已清理: confusion_matrix.png")
286

287
    print("\n" + "=" * 60)
288
    print("所有测试通过！")
289
    print("=" * 60)

1
import torch
2
import os
3
import sys
4
from text_data import WaimaiDataProcessor
5
from text_models import TextRNN, TextTransformer
6
from train import Trainer, test
7

8

9
def main():
10
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
11
    print(f"{'='*60}")
12
    print("实验2：外卖评论情感分类（TextRNN vs TextTransformer）")
13
    print(f"{'='*60}")
14
    print(f"设备: {device}")
15
    if not torch.cuda.is_available():
16
        print("警告: 使用CPU训练（较慢）")
17

18
    print(f"\n{'='*60}")
19
    print("步骤1: 数据工程")
20
    print(f"{'='*60}")
21

22

23
    try:
24
        processor = WaimaiDataProcessor(
25
            data_path='./data/waimai_10k.csv',
26
            batch_size=64,
27
            max_len=128
28
        )
29
        train_loader, val_loader, test_loader, vocab = processor.get_loaders(
30
            val_ratio=0.15,
31
            test_ratio=0.15
32
        )
33
        vocab_size = len(vocab)
34
        print(f"\n词表大小: {vocab_size}")
35

36
    except FileNotFoundError as e:
37
        print(e)
38
        return
39

40
    # 词表随权重一并保存，供可视化脚本加载复用
41
    import pickle
42
    os.makedirs('./checkpoints', exist_ok=True)
43
    with open('./checkpoints/vocab.pkl', 'wb') as f:
44
        pickle.dump(vocab, f)
45
    print(f"词表已保存: {len(vocab)}")
46

47

48
    print(f"\n{'='*60}")
49
    print("步骤2: TextRNN 训练与测试")
50
    print(f"{'='*60}")
51

52
    rnn_model = TextRNN(
53
        vocab_size=vocab_size,
54
        embed_dim=128,
55
        hidden_dim=256,
56
        num_classes=2,
57
        num_layers=2,
58
        dropout=0.5
59
    )
60
    rnn_params = sum(p.numel() for p in rnn_model.parameters()) / 1e6
61
    print(f"参数量: {rnn_params:.2f}M")
62

63
    trainer_rnn = Trainer(rnn_model, train_loader, val_loader, device,
64
                         lr=0.001, model_name='TextRNN')
65
    trainer_rnn.train(epochs=10)
66

67
    # 训练结束后从 checkpoint 加载最优权重，在测试集评估
68
    test_acc_rnn = test(rnn_model, test_loader, device,
69
                       model_path='./checkpoints/TextRNN_best.pth',
70
                       num_classes=2,
71
                       class_names=['负面', '正面'])
72

73
    print(f"\n{'='*60}")
74
    print("步骤3: TextTransformer 训练与测试")
75
    print(f"{'='*60}")
76

77
    trans_model = TextTransformer(
78
        vocab_size=vocab_size,
79
        embed_dim=128,
80
        num_heads=4,
81
        num_layers=2,
82
        dim_feedforward=512,
83
        num_classes=2,
84
        dropout=0.3
85
    )
86
    trans_params = sum(p.numel() for p in trans_model.parameters()) / 1e6
87
    print(f"参数量: {trans_params:.2f}M")
88

89
    trainer_trans = Trainer(trans_model, train_loader, val_loader, device,
90
                           lr=0.001, model_name='TextTransformer')
91
    trainer_trans.train(epochs=10)
92

93
    test_acc_trans = test(trans_model, test_loader, device,
94
                         model_path='./checkpoints/TextTransformer_best.pth',
95
                         num_classes=2,
96
                         class_names=['负面', '正面'])
97

98
    print(f"\n{'='*60}")
99
    print("步骤4: 实验结果对比")
100
    print(f"{'='*60}")
101
    print(f"{'模型':<20} {'参数量':<10} {'测试准确率':<10}")
102
    print("-" * 60)
103
    print(f"{'TextRNN':<20} {rnn_params:<10.2f} {test_acc_rnn:<10.2f}%")
104
    print(f"{'TextTransformer':<20} {trans_params:<10.2f} {test_acc_trans:<10.2f}%")
105
    print(f"{'='*60}")
106

107
    # 对比结果写入文本文件
108
    with open('experiment2_results.txt', 'w') as f:
109
        f.write("实验2：外卖评论情感分类对比结果\n")
110
        f.write(f"{'='*60}\n")
111
        f.write(f"数据集: waimai_10k (11,987条)\n")
112
        f.write(f"TextRNN: 参数量={rnn_params:.2f}M, 准确率={test_acc_rnn:.2f}%\n")
113
        f.write(f"TextTransformer: 参数量={trans_params:.2f}M, 准确率={test_acc_trans:.2f}%\n")
114

115
    print("\n结果已保存: experiment2_results.txt")
116

117

118
if __name__ == '__main__':
119
    main()

1
import torch
2
import random
3
import os
4
import pickle
5
import pandas as pd
6
from text_data import WaimaiDataset
7
from text_models import TextRNN, TextTransformer
8

9

10
def visualize_text_comparison(num_samples=10, save_path='visualization_text.txt'):
11
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
12
    print(f"使用设备: {device}")
13

14
    # 检查权重与词表文件是否齐全，缺失则提示先运行 text_main.py
15
    if not os.path.exists('./checkpoints/TextRNN_best.pth'):
16
        print("错误：找不到 TextRNN_best.pth")
17
        return False
18

19
    if not os.path.exists('./checkpoints/TextTransformer_best.pth'):
20
        print("错误：找不到 TextTransformer_best.pth")
21
        return False
22

23
    # 加载与训练时一致的词表，保证索引含义相同
24
    vocab_path = './checkpoints/vocab.pkl'
25
    if not os.path.exists(vocab_path):
26
        print(f"错误：找不到词表文件 {vocab_path}")
27
        print("请先运行 text_main.py 完成训练，会生成词表文件")
28
        return False
29

30
    with open(vocab_path, 'rb') as f:
31
        vocab = pickle.load(f)
32

33
    vocab_size = len(vocab)
34
    print(f"加载词表: {vocab_size} 个词")
35

36
    print("加载模型...")
37

38
    rnn = TextRNN(vocab_size, embed_dim=128, hidden_dim=256, num_classes=2).to(device)
39
    rnn.load_state_dict(torch.load('./checkpoints/TextRNN_best.pth', map_location=device))
40
    rnn.eval()
41

42
    transformer = TextTransformer(vocab_size, embed_dim=128, num_heads=4,
43
                                 num_layers=2, dim_feedforward=512, num_classes=2).to(device)
44
    transformer.load_state_dict(torch.load('./checkpoints/TextTransformer_best.pth', map_location=device))
45
    transformer.eval()
46

47
    # 从原始 CSV 中随机抽取指定数量的评论
48
    print("选择测试样本...")
49
    df = pd.read_csv('./data/waimai_10k.csv')
50
    test_indices = random.sample(range(len(df)), num_samples)
51
    sample_df = df.iloc[test_indices].reset_index(drop=True)
52

53
    label_names = {0: '负面', 1: '正面'}
54
    results = []
55

56
    print("进行预测...")
57
    for i, row in sample_df.iterrows():
58
        text = str(row['review'])
59
        true_label = row['label']
60

61
        # 复用 WaimaiDataset 的序列化逻辑，保证与训练时预处理一致
62
        dataset = WaimaiDataset([text], [true_label], vocab=vocab, max_len=128)
63
        seq, _ = dataset[0]
64
        seq = seq.unsqueeze(0).to(device)
65

66
        # Softmax 将 logits 转为概率分布，取最大项作为预测类别与置信度
67
        with torch.no_grad():
68
            output_rnn = rnn(seq)
69
            probs_rnn = torch.softmax(output_rnn, dim=1)
70
            pred_rnn = torch.argmax(probs_rnn, dim=1).item()
71
            conf_rnn = torch.max(probs_rnn).item()
72

73
        with torch.no_grad():
74
            output_trans = transformer(seq)
75
            probs_trans = torch.softmax(output_trans, dim=1)
76
            pred_trans = torch.argmax(probs_trans, dim=1).item()
77
            conf_trans = torch.max(probs_trans).item()
78

79
        results.append({
80
            'text': text[:80] + '...' if len(text) > 80 else text,
81
            'true_label': true_label,
82
            'true_name': label_names[true_label],
83
            'rnn_pred': pred_rnn,
84
            'rnn_name': label_names[pred_rnn],
85
            'rnn_conf': conf_rnn,
86
            'rnn_correct': pred_rnn == true_label,
87
            'trans_pred': pred_trans,
88
            'trans_name': label_names[pred_trans],
89
            'trans_conf': conf_trans,
90
            'trans_correct': pred_trans == true_label
91
        })
92

93
    # 生成逐样本对比报告
94
    lines = []
95
    lines.append("=" * 80)
96
    lines.append("实验2：外卖评论情感分类 - 模型对比可视化")
97
    lines.append("对比模型：TextRNN vs TextTransformer")
98
    lines.append("=" * 80)
99
    lines.append("")
100

101
    for i, r in enumerate(results, 1):
102
        lines.append(f"【样本 {i}】")
103
        lines.append(f"文本: {r['text']}")
104
        lines.append(f"真实标签: {r['true_name']} (label={r['true_label']})")
105
        lines.append("")
106
        lines.append(f"  TextRNN预测:        {r['rnn_name']:4s} (置信度: {r['rnn_conf']:.4f}) {'✓' if r['rnn_correct'] else '✗'}")
107
        lines.append(f"  TextTransformer预测: {r['trans_name']:4s} (置信度: {r['trans_conf']:.4f}) {'✓' if r['trans_correct'] else '✗'}")
108
        lines.append("-" * 80)
109
        lines.append("")
110

111
    # 统计两个模型在这批样本上的正确率
112
    correct_rnn = sum(1 for r in results if r['rnn_correct'])
113
    correct_trans = sum(1 for r in results if r['trans_correct'])
114

115
    lines.append("=" * 80)
116
    lines.append("统计汇总")
117
    lines.append("=" * 80)
118
    lines.append(f"测试样本数: {num_samples}")
119
    lines.append(f"TextRNN正确率:        {correct_rnn}/{num_samples} ({100*correct_rnn/num_samples:.1f}%)")
120
    lines.append(f"TextTransformer正确率: {correct_trans}/{num_samples} ({100*correct_trans/num_samples:.1f}%)")
121
    lines.append("=" * 80)
122

123
    content = '\n'.join(lines)
124
    with open(save_path, 'w', encoding='utf-8') as f:
125
        f.write(content)
126

127
    print(f"\n✓ 可视化结果已保存: {save_path}")
128
    print(f"\n预览（前3个样本）:")
129
    print('\n'.join(lines[:25]))
130

131
    return True
132

133

134
if __name__ == '__main__':
135
    visualize_text_comparison(num_samples=10)