from collections import defaultdict, Counter
 
class Vocab:
    def __init__(self, tokens=None):
        self.idx_to_token = list()
        self.token_to_idx = dict()
 
        if tokens is not None:
            if "<unk>" not in tokens:
                tokens = tokens + ["<unk>"]
            for token in tokens:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token) - 1
            self.unk = self.token_to_idx['<unk>']
 
    @classmethod
    def build(cls, text, min_freq=1, reserved_tokens=None):
        token_freqs = defaultdict(int)
        for sentence in text:
            for token in sentence:
                token_freqs[token] += 1
        uniq_tokens = ["<unk>"] + (reserved_tokens if reserved_tokens else [])
        uniq_tokens += [token for token, freq in token_freqs.items() \
                        if freq >= min_freq and token != "<unk>"]
        return cls(uniq_tokens)
 
    def __len__(self):
        #返回詞表大小，即詞表有多少個互不相同的標(biāo)記
        return len(self.idx_to_token)
 
    def __getitem__(self, token):
        #查找對應(yīng)輸入標(biāo)記的索引
        #不存在，返回標(biāo)記<unk>的索引
        return self.token_to_idx.get(token, self.unk)
 
    def convert_tokens_to_ids(self, tokens):
        #查找一系列輸入標(biāo)記對應(yīng)的索引
        return [self[token] for token in tokens]
 
    def convert_ids_to_tokens(self, indices):
        #查找一系列索引值對應(yīng)的標(biāo)記
        return [self.idx_to_token[index] for index in indices]
 
 
def save_vocab(vocab, path):
    with open(path, 'w') as writer:
        writer.write("\n".join(vocab.idx_to_token))
 
 
def read_vocab(path):
    with open(path, 'r') as f:
        tokens = f.read().split('\n')
    return Vocab(tokens)
 

import torch
from torch import nn
#詞表大小為8，Embedding向量維度3
embedding=nn.Embedding(8,3)
#輸入形狀(2,4)的整數(shù)張量，相當(dāng)于2個長度為4的整數(shù)序列，每個整數(shù)范圍是0-7
input=torch.tensor([[0,1,2,1],[4,6,6,7]],dtype=torch.long)
#調(diào)用embedding
output=embedding(input)
print(output)
print(output.shape)

tensor([[[-0.9536,  0.3367, -1.1639],
         [-0.4816, -0.8973, -0.7700],
         [ 1.9007,  1.3130, -1.2717],
         [-0.4816, -0.8973, -0.7700]],
 
        [[ 0.5755,  0.8908, -1.0384],
         [ 0.3724,  0.1216,  2.4466],
         [ 0.3724,  0.1216,  2.4466],
         [ 1.4089,  0.2458,  0.2263]]], grad_fn=<EmbeddingBackward0>)
torch.Size([2, 4, 3])

import torch
from torch import nn
from torch.nn import functional as F
 
class MLP(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_class):
        super(MLP, self).__init__()
        # 詞嵌入層
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        # 線性變換：詞嵌入層->隱含層
        self.linear1 = nn.Linear(embedding_dim, hidden_dim)
        # 使用ReLU激活函數(shù)
        self.activate = F.relu
        # 線性變換：激活層->輸出層
        self.linear2 = nn.Linear(hidden_dim, num_class)
 
    def forward(self, inputs):
        embeddings = self.embedding(inputs)
        # 將序列中多個embedding進(jìn)行聚合（此處是求平均值）
        embedding = embeddings.mean(dim=1)
        hidden = self.activate(self.linear1(embedding))
        outputs = self.linear2(hidden)
        # 獲得每個序列屬于某一類別概率的對數(shù)值
        probs = F.log_softmax(outputs, dim=1)
        return probs
 
mlp = MLP(vocab_size=8, embedding_dim=3, hidden_dim=5, num_class=2)
# 輸入為兩個長度為4的整數(shù)序列
inputs = torch.tensor([[0, 1, 2, 1], [4, 6, 6, 7]], dtype=torch.long)
outputs = mlp(inputs)
print(outputs)

tensor([[-1.1612, -0.3756],
        [-0.8089, -0.5894]], grad_fn=<LogSoftmaxBackward0>)

import torch
from vocab import Vocab
 
def load_sentence_polarity():
    from nltk.corpus import sentence_polarity
 
    #使用全部橘子集合（已經(jīng)過標(biāo)記解析）創(chuàng)建詞表
    vocab = Vocab.build(sentence_polarity.sents())
 
    #褒貶各4000句子作為訓(xùn)練數(shù)據(jù)，使用創(chuàng)建的詞表將標(biāo)記映射為相應(yīng)的索引
    #褒義標(biāo)簽為0，貶義標(biāo)簽為1
    #每個樣例是一個有索引值列表和標(biāo)簽組成的元祖
    train_data = [(vocab.convert_tokens_to_ids(sentence), 0)
                  for sentence in sentence_polarity.sents(categories='pos')[:4000]] \
        + [(vocab.convert_tokens_to_ids(sentence), 1)
            for sentence in sentence_polarity.sents(categories='neg')[:4000]]
 
    #其余的作為測試數(shù)據(jù)
    test_data = [(vocab.convert_tokens_to_ids(sentence), 0)
                 for sentence in sentence_polarity.sents(categories='pos')[4000:]] \
        + [(vocab.convert_tokens_to_ids(sentence), 1)
            for sentence in sentence_polarity.sents(categories='neg')[4000:]]
 
    return train_data, test_data, vocab
 
def length_to_mask(lengths):
    max_len = torch.max(lengths)
    mask = torch.arange(max_len).expand(lengths.shape[0], max_len) < lengths.unsqueeze(1)
    return mask
 
def load_treebank():
    from nltk.corpus import treebank
    sents, postags = zip(*(zip(*sent) for sent in treebank.tagged_sents()))
 
    vocab = Vocab.build(sents, reserved_tokens=["<pad>"])
 
    tag_vocab = Vocab.build(postags)
 
    train_data = [(vocab.convert_tokens_to_ids(sentence), tag_vocab.convert_tokens_to_ids(tags)) for sentence, tags in zip(sents[:3000], postags[:3000])]
    test_data = [(vocab.convert_tokens_to_ids(sentence), tag_vocab.convert_tokens_to_ids(tags)) for sentence, tags in zip(sents[3000:], postags[3000:])]
 
    return train_data, test_data, vocab, tag_vocab
 

import torch
from torch import nn, optim
from torch.nn import functional as F
from torch.utils.data import Dataset, DataLoader
from collections import defaultdict
from vocab import Vocab
from utils import load_sentence_polarity
 
class BowDataset(Dataset):
    def __init__(self, data):
        #data為原始數(shù)據(jù)
        self.data = data
    def __len__(self):
        #數(shù)據(jù)集中樣例的數(shù)目
        return len(self.data)
    def __getitem__(self, i):
        #返回下標(biāo)i的數(shù)據(jù)
        return self.data[i]
 
#用于對樣本進(jìn)行整理
def collate_fn(examples):
    #從獨立樣本集合中構(gòu)建各批次的輸入輸出 
    #其中，BowDataset類定義了一個樣本的數(shù)據(jù)結(jié)構(gòu)，即輸入標(biāo)簽和輪出標(biāo)悠的元組
    # 因此，將輸入inputs定義為一個張量的列表，其中每個張量為原始句子中標(biāo)記序列 
    inputs = [torch.tensor(ex[0]) for ex in examples]
    #輸出的目標(biāo)targets 為該批次中全部樣例輸出結(jié)果 （0或1） 構(gòu)成的張量
    targets = torch.tensor([ex[1] for ex in examples], dtype=torch.long)
    # 獲取一個批次中每個樣例的序列長度 
    offsets = [0] + [i.shape[0] for i in inputs]
    #根據(jù)序列的長度，轉(zhuǎn)換為每個序列起始位置的偏移量 
    offsets = torch.tensor(offsets[:-1]).cumsum(dim=0)
    #將inputs列表中的張量拼接成一個大的張量 
    inputs = torch.cat(inputs)
    return inputs, offsets, targets

import torch
from torch import nn, optim
from torch.nn import functional as F
from torch.utils.data import Dataset, DataLoader
from collections import defaultdict
from vocab import Vocab
from utils import load_sentence_polarity
 
 
# tqdm是一個Python模塊，能以進(jìn)度條的方式顯示迭代的進(jìn)度
from tqdm.auto import tqdm
 
# 超參數(shù)設(shè)置
embedding_dim = 128
hidden_dim = 256
num_class = 2
batch_size = 32
num_epoch = 5
 
# 加載數(shù)據(jù)
train_data, test_data, vocab = load_sentence_polarity()
train_dataset = BowDataset(train_data)
test_dataset = BowDataset(test_data)
train_data_loader = DataLoader(train_dataset, batch_size=batch_size, collate_fn=collate_fn, shuffle=True)
test_data_loader = DataLoader(test_dataset, batch_size=1, collate_fn=collate_fn, shuffle=False)
 
# 加載模型
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = MLP(len(vocab), embedding_dim, hidden_dim, num_class)
model.to(device) # 將模型加載到CPU或GPU設(shè)備
 
#訓(xùn)練過程
nll_loss = nn.NLLLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001) # 使用Adam優(yōu)化器
 
model.train()
for epoch in range(num_epoch):
    total_loss = 0
    for batch in tqdm(train_data_loader, desc=f"Training Epoch {epoch}"):
        inputs, offsets, targets = [x.to(device) for x in batch]
        log_probs = model(inputs, offsets)
        loss = nll_loss(log_probs, targets)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Loss: {total_loss:.2f}")
 
# 測試過程
acc = 0
for batch in tqdm(test_data_loader, desc=f"Testing"):
    inputs, offsets, targets = [x.to(device) for x in batch]
    with torch.no_grad():
        output = model(inputs, offsets)
        acc += (output.argmax(dim=1) == targets).sum().item()
 
# 輸出在測試集上的準(zhǔn)確率
print(f"Acc: {acc / len(test_data_loader):.2f}")

import torch
from torch import nn, optim
from torch.nn import functional as F
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pad_sequence
from collections import defaultdict
from vocab import Vocab
from utils import load_sentence_polarity
 
class CnnDataset(Dataset):
    def __init__(self, data):
        self.data = data
    def __len__(self):
        return len(self.data)
    def __getitem__(self, i):
        return self.data[i]
 
#另外，數(shù)據(jù)整理函數(shù)也需要進(jìn)行一些修改。  
# pad_seguence 兩數(shù)實現(xiàn)補齊 （Padding）功能，使得一個批次中 全部序列長度相同（同最大長度序列），不足的默認(rèn)使用0補齊。 
def collate_fn(examples):
    inputs = [torch.tensor(ex[0]) for ex in examples]
    targets = torch.tensor([ex[1] for ex in examples], dtype=torch.long)
    # 對batch內(nèi)的樣本進(jìn)行padding，使其具有相同長度
    inputs = pad_sequence(inputs, batch_first=True)
    return inputs, targets
 
#模型不同，需要從nn.Module類派生一個 CNN 子類
class CNN(nn.Module):
    def __init__(self, vocab_size, embedding_dim, filter_size, num_filter, num_class):
        super(CNN, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        #padding=1表示在卷積操作之前，將序列的前后各補充1個輸入
        self.conv1d = nn.Conv1d(embedding_dim, num_filter, filter_size, padding=1)
        self.activate = F.relu
        self.linear = nn.Linear(num_filter, num_class)
    def forward(self, inputs):
        embedding = self.embedding(inputs)
        convolution = self.activate(self.conv1d(embedding.permute(0, 2, 1)))
        pooling = F.max_pool1d(convolution, kernel_size=convolution.shape[2])
        outputs = self.linear(pooling.squeeze(dim=2))
        log_probs = F.log_softmax(outputs, dim=1)
        return log_probs
 
#tqdm是一個Pyth模塊，能以進(jìn)度條的方式顯示迭代的進(jìn)度
from tqdm.auto import tqdm
 
#超參數(shù)設(shè)置
embedding_dim = 128
hidden_dim = 256
num_class = 2
batch_size = 32
num_epoch = 5
#在調(diào)用卷積神經(jīng)網(wǎng)絡(luò)時，還需要設(shè)置兩個額外的超參數(shù)，分別為filter_size =3（卷積核的大?。┖蚽um_tilter =100（卷積核的個數(shù)）。
filter_size = 3
num_filter = 100
 
#加載數(shù)據(jù)
train_data, test_data, vocab = load_sentence_polarity()
train_dataset = CnnDataset(train_data)
test_dataset = CnnDataset(test_data)
train_data_loader = DataLoader(train_dataset, batch_size=batch_size, collate_fn=collate_fn, shuffle=True)
test_data_loader = DataLoader(test_dataset, batch_size=1, collate_fn=collate_fn, shuffle=False)
 
#加載模型
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = CNN(len(vocab), embedding_dim, filter_size, num_filter, num_class)
model.to(device) #將模型加載到CPU或GPU設(shè)備
 
#訓(xùn)練過程
nll_loss = nn.NLLLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001) #使用Adam優(yōu)化器
 
model.train()
for epoch in range(num_epoch):
    total_loss = 0
    for batch in tqdm(train_data_loader, desc=f"Training Epoch {epoch}"):
        inputs, targets = [x.to(device) for x in batch]
        log_probs = model(inputs)
        loss = nll_loss(log_probs, targets)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Loss: {total_loss:.2f}")
 
#測試過程
acc = 0
for batch in tqdm(test_data_loader, desc=f"Testing"):
    inputs, targets = [x.to(device) for x in batch]
    with torch.no_grad():
        output = model(inputs)
        acc += (output.argmax(dim=1) == targets).sum().item()
 
#輸出在測試集上的準(zhǔn)確率
print(f"Acc: {acc / len(test_data_loader):.2f}")

import torch
from torch import nn, optim
from torch.nn import functional as F
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pad_sequence, pack_padded_sequence
from collections import defaultdict
from vocab import Vocab
from utils import load_sentence_polarity
 
#tqdm是一個Python模塊，能以進(jìn)度條的方式顯式迭代的進(jìn)度
from tqdm.auto import tqdm
 
class LstmDataset(Dataset):
    def __init__(self, data):
        self.data = data
    def __len__(self):
        return len(self.data)
    def __getitem__(self, i):
        return self.data[i]
 
#整理函數(shù)
def collate_fn(examples):
    #獲得每個序列的長度
    lengths = torch.tensor([len(ex[0]) for ex in examples])
    inputs = [torch.tensor(ex[0]) for ex in examples]
    targets = torch.tensor([ex[1] for ex in examples], dtype=torch.long)
    # 對batch內(nèi)的樣本進(jìn)行padding，使其具有相同長度
    inputs = pad_sequence(inputs, batch_first=True)
    return inputs, lengths, targets
 
#需要從nn.Module派生一個LSTM子類
class LSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_class):
        super(LSTM, self).__init__()
        self.embeddings = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True)
        self.output = nn.Linear(hidden_dim, num_class)
 
    def forward(self, inputs, lengths):
        embeddings = self.embeddings(inputs)
        #使用 pack_padded_sequence將變長序列打包
        x_pack = pack_padded_sequence(embeddings, lengths, batch_first=True, enforce_sorted=False)
        hidden, (hn, cn) = self.lstm(x_pack)
        outputs = self.output(hn[-1])
        log_probs = F.log_softmax(outputs, dim=-1)
        return log_probs
 
embedding_dim = 128
hidden_dim = 256
num_class = 2
batch_size = 32
num_epoch = 5
 
#加載數(shù)據(jù)
train_data, test_data, vocab = load_sentence_polarity()
train_dataset = LstmDataset(train_data)
test_dataset = LstmDataset(test_data)
train_data_loader = DataLoader(train_dataset, batch_size=batch_size, collate_fn=collate_fn, shuffle=True)
test_data_loader = DataLoader(test_dataset, batch_size=1, collate_fn=collate_fn, shuffle=False)
 
#加載模型
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = LSTM(len(vocab), embedding_dim, hidden_dim, num_class)
model.to(device) #將模型加載到GPU中（如果已經(jīng)正確安裝）
 
#訓(xùn)練過程
nll_loss = nn.NLLLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001) #使用Adam優(yōu)化器
 
model.train()
for epoch in range(num_epoch):
    total_loss = 0
    for batch in tqdm(train_data_loader, desc=f"Training Epoch {epoch}"):
        inputs, lengths, targets = [x.to(device) for x in batch]
        log_probs = model(inputs, lengths)
        loss = nll_loss(log_probs, targets)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Loss: {total_loss:.2f}")
 
#測試過程
acc = 0
for batch in tqdm(test_data_loader, desc=f"Testing"):
    inputs, lengths, targets = [x.to(device) for x in batch]
    with torch.no_grad():
        output = model(inputs, lengths)
        acc += (output.argmax(dim=1) == targets).sum().item()
 
#輸出在測試集上的準(zhǔn)確率
print(f"Acc: {acc / len(test_data_loader):.2f}")

# Defined in Section 4.6.8
 
import math
import torch
from torch import nn, optim
from torch.nn import functional as F
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pad_sequence, pack_padded_sequence
from collections import defaultdict
from vocab import Vocab
from utils import load_sentence_polarity, length_to_mask
 
# tqdm是一個Pyth模塊，能以進(jìn)度條的方式顯式迭代的進(jìn)度
from tqdm.auto import tqdm
 
class TransformerDataset(Dataset):
    def __init__(self, data):
        self.data = data
    def __len__(self):
        return len(self.data)
    def __getitem__(self, i):
        return self.data[i]
 
def collate_fn(examples):
    lengths = torch.tensor([len(ex[0]) for ex in examples])
    inputs = [torch.tensor(ex[0]) for ex in examples]
    targets = torch.tensor([ex[1] for ex in examples], dtype=torch.long)
    # 對batch內(nèi)的樣本進(jìn)行padding，使其具有相同長度
    inputs = pad_sequence(inputs, batch_first=True)
    return inputs, lengths, targets
 
class PositionalEncoding(nn.Module):
    def __init__(self, d_model, dropout=0.1, max_len=512):
        super(PositionalEncoding, self).__init__()
 
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0).transpose(0, 1)
        self.register_buffer('pe', pe)
 
    def forward(self, x):
        x = x + self.pe[:x.size(0), :]
        return x
 
class Transformer(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_class,
                 dim_feedforward=512, num_head=2, num_layers=2, dropout=0.1, max_len=128, activation: str = "relu"):
        super(Transformer, self).__init__()
        # 詞嵌入層
        self.embedding_dim = embedding_dim
        self.embeddings = nn.Embedding(vocab_size, embedding_dim)
        self.position_embedding = PositionalEncoding(embedding_dim, dropout, max_len)
        # 編碼層：使用Transformer
        encoder_layer = nn.TransformerEncoderLayer(hidden_dim, num_head, dim_feedforward, dropout, activation)
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
        # 輸出層
        self.output = nn.Linear(hidden_dim, num_class)
 
 
    def forward(self, inputs, lengths):
        inputs = torch.transpose(inputs, 0, 1)
        #與LSTM處理情況相同，輸入的數(shù)據(jù)的第一維是批次，需要轉(zhuǎn)換為TransformerEncoder
        #所需要的第1維是長度，第二維是批次的形狀
        hidden_states = self.embeddings(inputs)
        hidden_states = self.position_embedding(hidden_states)
        #length_to_mask作用是根據(jù)批次中每個序列的長度生成mask矩陣，以便處理長度不一致的序列，忽略掉比較短的序列的無效部分
        #src_key_padding_mask的參數(shù)正好與length_to_mask的結(jié)果相反（無自注意力的部分為true）
        attention_mask = length_to_mask(lengths.to('cpu')) == False
        #根據(jù)批次中的每個序列長度生成mask矩陣
        hidden_states = self.transformer(hidden_states, src_key_padding_mask=attention_mask.to('cuda'))
        hidden_states = hidden_states[0, :, :]
        #取第一個標(biāo)記的輸出結(jié)果作為分類層的輸入
        output = self.output(hidden_states)
        log_probs = F.log_softmax(output, dim=1)
        return log_probs
 
embedding_dim = 128
hidden_dim = 128
num_class = 2
batch_size = 32
num_epoch = 5
 
# 加載數(shù)據(jù)
train_data, test_data, vocab = load_sentence_polarity()
train_dataset = TransformerDataset(train_data)
test_dataset = TransformerDataset(test_data)
train_data_loader = DataLoader(train_dataset, batch_size=batch_size, collate_fn=collate_fn, shuffle=True)
test_data_loader = DataLoader(test_dataset, batch_size=1, collate_fn=collate_fn, shuffle=False)
 
# 加載模型
device = torch.device('cuda')
model = Transformer(len(vocab), embedding_dim, hidden_dim, num_class)
model.to(device) # 將模型加載到GPU中（如果已經(jīng)正確安裝）
 
# 訓(xùn)練過程
nll_loss = nn.NLLLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001) # 使用Adam優(yōu)化器
 
model.train()
for epoch in range(num_epoch):
    total_loss = 0
    for batch in tqdm(train_data_loader, desc=f"Training Epoch {epoch}"):
        inputs, lengths, targets = [x.to(device) for x in batch]
        log_probs = model(inputs, lengths)
        loss = nll_loss(log_probs, targets)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Loss: {total_loss:.2f}")
 
# 測試過程
acc = 0
for batch in tqdm(test_data_loader, desc=f"Testing"):
    inputs, lengths, targets = [x.to(device) for x in batch]
    with torch.no_grad():
        output = model(inputs, lengths)
        acc += (output.argmax(dim=1) == targets).sum().item()
 
# 輸出在測試集上的準(zhǔn)確率
print(f"Acc: {acc / len(test_data_loader):.2f}")