<jupyter_start><jupyter_text>数据读取与处理<jupyter_code>text_field = torchtext.data.Field(
    lower=True, 
    include_lengths=True,
    batch_first=True
)
label_field = torchtext.data.Field(
    sequential=False,
    use_vocab=False,
    preprocessing=lambda x: 0 if x=='neg' else 1,
    dtype=torch.float
)

# 生成IMDB数据集
train_data, test_data = torchtext.datasets.IMDB.splits(
    text_field=text_field, 
    label_field=label_field,
    root='/media/bnu/data/nlp-practice/sentiment-analysis/',
)
# 划分验证集
train_data, valid_data = train_data.split(split_ratio=0.8)
print('Text Sample:')
print(' '.join(train_data.examples[0].text))
print('-' * 60)
print('Label Sample:')
print(train_data.examples[0].label)
print('-' * 60)
print()
print('Train Examples:', len(train_data))
print('Valid Examples:', len(valid_data))
print('Test Examples:', len(test_data))
# 使用预训练词向量构建单词表
text_field.build_vocab(
    train_data, 
    max_size=25000, 
    vectors='glove.6B.100d',
    vectors_cache='/media/bnu/data/nlp-practice/word-vector',
    unk_init=torch.Tensor.zero_
)
print('Text Vocab Size:', len(text_field.vocab))
print('-' * 60)
print('Text Str -> Index:')
print(list(text_field.vocab.stoi.items())[:10])
# 创建Iterators
train_iter, valid_iter, test_iter = torchtext.data.BucketIterator.splits(
    (train_data, valid_data, test_data),
    batch_sizes=(32, 32, 32),
    device=device,
    shuffle=True
)
print('Inputs Data Shape:', next(iter(train_iter)).text[0].shape)
print('Inputs Length Shape:', next(iter(train_iter)).text[1].shape)
print('Target Data Shape:', next(iter(train_iter)).label.shape)<jupyter_output>Inputs Data Shape: torch.Size([32, 484])
Inputs Length Shape: torch.Size([32])
Target Data Shape: torch.Size([32])<jupyter_text>定义模型<jupyter_code>class WordAVGModel(nn.Module):
    
    def __init__(self, n_words, n_embed):
        super(WordAVGModel, self).__init__()
        self.embed = nn.Embedding(n_words, n_embed)
        self.fc = nn.Linear(n_embed, 1)
    
    def forward(self, inputs):
        # inputs shape: (batch_size, max_len)
        
        # x_embed shape: (batch_size, max_len, embed_size)
        x_embed = self.embed(inputs)
        # x_mean shape: (batch_size, embed_size)
        x_mean = torch.mean(x_embed, dim=1)
        
        # outputs shape: (batch_size)
        return self.fc(x_mean).squeeze()
    
    def init_embed(self, pretrained_embeddings):
        self.embed.weight.data.copy_(pretrained_embeddings)
        

inputs = next(iter(train_iter)).text[0]
model = WordAVGModel(len(text_field.vocab), 100)
model.init_embed(text_field.vocab.vectors)
model.to(device)
outputs = model(inputs)
print(outputs.shape)
class BiLSTMModel(nn.Module):
    
    def __init__(self, n_words, n_embed, n_hidden, n_layers):
        super(BiLSTMModel, self).__init__()
        self.embed = nn.Embedding(n_words, n_embed)
        self.rnn = nn.LSTM(n_embed, n_hidden, 
                           num_layers=n_layers, 
                           bidirectional=True, 
                           dropout=0.2,
                           batch_first=True)
        self.fc = nn.Linear(n_hidden * 2, 1)
        self.dropout = nn.Dropout(0.2)
        
    def forward(self, inputs):
        # inputs shape: (batch_size, max_len)
        
        # x_embed shape: (batch_size, max_len, embed_size)
        x_embed = self.dropout(self.embed(inputs))
        
        # x_rnn shape: (batch_size, max_len, 2 * hidden_size)
        # h_n shape: (num_layer * num_dir, batch_size, hidden_size)
        x_rnn, (h_n, c_n) = self.rnn(x_embed)
        
        x_hidden = self.dropout(torch.cat((h_n[-1], h_n[-2]), dim=1))
        
        # outputs shape: (batch_size)
        return self.fc(x_hidden).squeeze()
    
    def init_embed(self, pretrained_embeddings):
        self.embed.weight.data.copy_(pretrained_embeddings)
        
        
inputs = next(iter(train_iter)).text[0]
model = BiLSTMModel(len(text_field.vocab), 100, 100, 2)
model.init_embed(text_field.vocab.vectors)
model.to(device)
outputs = model(inputs)
print(outputs.shape)
class CNNModel(nn.Module):
    
    def __init__(self, n_words, n_embed, n_filters, kernel_sizes):
        super(CNNModel, self).__init__()
        self.embed = nn.Embedding(n_words, n_embed)
        self.convs = nn.ModuleList([
            nn.Conv1d(n_embed, n_filters, kernel_size=ks)
            for ks in kernel_sizes
        ])
        self.fc = nn.Linear(len(kernel_sizes) * n_filters, 1)
        self.dropout = nn.Dropout(0.5)
        
    def forward(self, inputs):
        # inputs shape: (batch_size, max_len)
        
        # x_embed shape: (batch_size, max_len, embed_size)
        x_embed = self.embed(inputs)
        # x_embed shape: (batch_size, embed_size, max_len)
        x_embed = x_embed.transpose(1, 2)

        
        # x_convs[0] shape: (batch_size, num_filter, max_len - kernel_size + 1)
        x_convs = [F.relu(conv(x_embed)) for conv in self.convs]
        # x_pools[0] shape: (batch_size, num_filter)
        x_pools = [F.max_pool1d(x_conv, x_conv.shape[2]).squeeze() 
                   for x_conv in x_convs]
        
        # x_cat shape: (batch_size, num_kernel * num_filter)
        x_cat = self.dropout(torch.cat(x_pools, dim=1))
        
        # outputs shape: (batch_size)
        return self.fc(x_cat).squeeze()
        
        
    def init_embed(self, pretrained_embeddings):
        self.embed.weight.data.copy_(pretrained_embeddings)
        
        
inputs = next(iter(train_iter)).text[0]
model = CNNModel(len(text_field.vocab), 100, 50, [2, 3, 4, 5])
model.init_embed(text_field.vocab.vectors)
model.to(device)
outputs = model(inputs)
print(outputs.shape)<jupyter_output>torch.Size([32])<jupyter_text>模型训练<jupyter_code>class IDMBLearner:
    
    def __init__(self, model):
        self.model = model
        self.model.to(device)
        self.criterion = nn.BCEWithLogitsLoss()
        self.optimizer = torch.optim.Adam(self.model.parameters())
        self.history = defaultdict(list)
        
    def calc_correct_count(self, outputs, targets):
        preds = torch.round(torch.sigmoid(outputs))
        return (preds == targets).long().sum().item()
        
        
    def fit(self, n_epochs):
        for epoch in range(n_epochs):
            self.model.train()
            total_loss = 0.0
            total_samples, total_correct = 0, 0
            
            pbar = tqdm.notebook.tqdm(train_iter)
            pbar.set_description(f'Epoch {epoch+1} --> Train')
            
            for batch in pbar:
                inputs, targets = batch.text[0].to(device), batch.label.to(device)
                outputs = self.model(inputs)
                
                loss = self.criterion(outputs, targets)
                self.optimizer.zero_grad()
                loss.backward()
                self.optimizer.step()
                
                total_samples += len(targets)
                total_loss += loss.item() * len(targets)
                total_correct += self.calc_correct_count(outputs, targets)
                
                pbar.set_postfix(loss=total_loss/total_samples, acc=total_correct/total_samples)
            
            self.history['train_loss'].append(total_loss/total_samples)
            self.history['train_acc'].append(total_correct/total_samples)
                
            self.model.eval()
            total_loss = 0.0
            total_samples, total_correct = 0, 0
            
            pbar = tqdm.notebook.tqdm(valid_iter)
            pbar.set_description(f'Epoch {epoch+1} --> Valid')
            
            with torch.no_grad():
                for batch in pbar:
                    inputs, targets = batch.text[0].to(device), batch.label.to(device)
                    outputs = self.model(inputs)
                    
                    loss = self.criterion(outputs, targets)
                    
                    total_samples += len(targets)
                    total_loss += loss.item() * len(targets)
                    total_correct += self.calc_correct_count(outputs, targets)
                
                    pbar.set_postfix(loss=total_loss/total_samples, acc=total_correct/total_samples)
                
            self.history['valid_loss'].append(total_loss/total_samples)
            self.history['valid_acc'].append(total_correct/total_samples)
model = WordAVGModel(len(text_field.vocab), 100)
model.init_embed(text_field.vocab.vectors)
model.to(device)
learner = IDMBLearner(model)
learner.fit(10)
model = BiLSTMModel(len(text_field.vocab), 100, 256, 2)
model.init_embed(text_field.vocab.vectors)
model.to(device)
learner = IDMBLearner(model)
learner.fit(10)
model = CNNModel(len(text_field.vocab), 100, 50, [2, 3, 4, 5])
model.init_embed(text_field.vocab.vectors)
model.to(device)
learner = IDMBLearner(model)
learner.fit(10)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12298136645962733, 'fraction_numerical': 0.014116318464144552, 'mean_word_length': 2.558055443953395, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '380378', 'license': 'MIT', 'n_tokens_mistral': 3140, 'n_tokens_neox': 3022, 'n_words': 482, 'path': 'notebooks/exercises/PyTorch_TextClassification.ipynb', 'repo_name': 'RegiusQuant/nlp-practice', 'size': '8855'}