不論是建立語言翻譯的模型或是推測新字的模型，都可被歸類為時序資料分析，個中原因：除了第一個字，一個句子中的每一個字的產生都和前面的文意有所關係。分析語言資料的首要問題是，如何將語言資料數值化？
或許我們可以仿巴赫讚歌的例子建立一個數值和音符的對應表，使用整數數值來表達句子中每一個字義的基本單元(字元)。
目前最流行的方法是一種稱為embedding的方法，它是延申自對應表的方法。簡單地說，embedding將每個具有個別意義的字或詞用一組浮點數向量表示，將單一向量裡的每一個數值，做為特徵值，因此每一個字元是使用一個特徵值向量來表示，這些特徵值代表著字元在語料庫空間中的某一個向度的權重值。
比方而，我們可以使用Tensorflow的 Embedding projector 來建立了一個三維的視覺化空間，來表示某一個字庫，圖中每一個點代表一個字元，加上圓點大小維度，此圖代表著四維的向量空間(Hyperspace)。
我們可以借前文的「音符-整數」對照表來了解embeddings的優點，「音符-整數」對照表僅保存了音符間的順序關係，若應用類似的方法或許我們可以保留字頻之間的關係(一維)，因為embedding是將每一個字元轉換至高向度的空間來表達，所以它除了可以保留字頻外，還可以保留其他字元間的關係，如字義、相似性等等。
StackOverflow的例子
Stack Overflow是一個程式語言的問答論壇，Tensorflow 的語言資料集中收集了一組取自於Stack Overflow的問題集。此資料集將每一個問題儲存於一個文字檔裡面，並將這些檔案細心地分為四個類別：C#，Java，Javascript 和 Python的問題。
在Tensorflow的教學說明文件說明了如何使用Tensorflow來進行此分類作業來幫助了解如何使用Tensorflow的工具進行語言資料分析。以下我們此資料集及PyTorch來說明embedding的概念。
1. 讀入資料
首先，我們印出此資料集訓練組，屬於C#類別的第一個文字檔，來看看資料的原始內容。
import pathlib
import os
dataset_dir = "~/datasets/stack_overflow_16k/train/csharp/"
csharp_dir = pathlib.Path(dataset_dir)
files = os.listdir(csharp_dir)
file = dataset_dir + files[0]
with open(file, 'r', encoding='utf-8') as f:
text = f.read()
# 印出前100個字元(character)
print(text[:100])
# "how to aovid flickering of richtextbox when the text is
# scrolling down fast? i have a richtextbox w"
一如往常，我們先將所有的資料檔讀入記憶體中，因為此資料集已經被分類為訓練組及測試組兩個資料夾，我們將屬於訓練組資料夾中的問題再分類為訓練組及驗證組。
import torch
from torch.utils.data import random_split
FOLDER_NAMES = ['csharp', 'java', 'javascript', 'python']
LABELS = {folder: idx for idx, folder in enumerate(FOLDER_NAMES)}
dataset_dir = "~/datasets/stack_overflow_16k/"
train_dir = pathlib.Path(dataset_dir + "train/")
test_dir = pathlib.Path(dataset_dir + "test/")
def read_data(directory):
files_list, labels_list = [], []
for folder in FOLDER_NAMES:
file_dir = directory / folder
files = os.listdir(file_dir)
for file in files:
file_path = file_dir / file
with open(file_path, 'r', encoding='utf-8') as f:
files_list.append(f.read())
labels_list.append(LABELS[folder])
return labels_list, files_list
train_labels, train_files = read_data(train_dir)
test_labels, test_files = read_data(test_dir)
train_ds = [(label, text) for label, text in zip(train_labels, train_files)]
test_ds = [(label, text) for label, text in zip(test_labels, test_files)]
total_size = len(train_ds)
train_size = int(0.8 * total_size) # 80% for training
valid_size = total_size - train_size # 20% for validation
train_dataset, valid_dataset = random_split(train_ds, [train_size, valid_size])
test_dataset = test_ds
print(train_dataset[0])
# (3,
# '"can\'t understand the result of innerclass i has met a problem when i was learning innerclass of blank...)
2. Tokenization
到此為止，文字資料仍然維持其原始樣貌。我將原始文字轉換為數值的第一步通常需要將其分割為個別意義單元，此步驟稱為tokenize。而這些字義單元稱為token。此步驟通常會將對文字分析沒有影響的字元移除或改變，例如將所有的字母改為小寫，移除表情符號、空白鍵或標點符號等等。Tensorflow及PyTorch套件，或是它們的支援套件都附有標準的tokenization函式，或者我們也可以使用自製函式，例如：
def tokenizer(text):
text = re.sub('<[^>]*>', '', text)
emoticons = re.findall('(?::|;|=)(?:-)?(?:\)|\(|D|P)', text.lower())
text = re.sub('[\W]+', ' ', text.lower()) +\
' '.join(emoticons).replace('-', '')
tokenized = text.split()
return tokenized
# 前例tokenize之後變為，如下所印出的各別字彙的list。
tokenizer(example0)
#['can',
# 't',
# 'understand',
# 'the',
# ...
#]3. Ranking
接下來是將字庫(Corpus)中的token依照特定原則將之排序，最常使用的原則是使用出現次數(字頻)。我們通常會使用整個訓練組做為字庫，為了解釋Ranking的概念，以下僅用單一個句子做為字庫。比方說，我們將此句子：
"cats like playing, and they just can't help to play with other cats. They just meow, meow, meow"
做為字庫，將其tokenize之後，我們會得到14個tokens，它們的字頻對照表。
ID Token Count
0 meow 3
1 cats 2
2 they 2
3 just 2
4 like 1
5 playing 1
6 and 1
7 can 1
8 t 1
9 help 1
10 to 1
11 play 1
12 with 1
13 other 1如此我們就可以得到一個和之前巴赫音樂相仿的對照表。此對照表只有字頻之間的關係，如前文所解釋，RNN讀入時序資料時，會使用內部層的節點來保留時序關係。若我們使用ID欄位做為字元的數值表徵，此例句在讀入模型時，它會呈現如下所示，這些IDs之間的順序關係(例如1->4->5->6 ...)保留著句子結構，此數值概念就是初等統計學中的順序變數(ordinary variable)的概念。
ID Token Count
1 cats 2
4 like 1
5 playing 1
6 and 1
2 they 2
3 just 2
7 can 1
8 t 1
9 help 1
10 to 1
11 play 1
12 with 1
13 other 1
0 meow 3
4. Embedding
Embedding使用連續變數(continuous variable)的概念取代順序變數的概念，相比於僅用字頻排序的靜態對照表來代表字庫中的每個字元，embedding使用多個浮點數來代表每個字元，並將之改為動態對照表。
這裡動態的實現方式是讓RNN模型，根據訓練組的資料去最佳化這些浮點數。以這個單一句子的字庫為例子說明，若我們決定使用三個特徵值，我們可以將此字庫使用如下面右方的embedding矩陣來表示。矩陣中的r0_1，r1_1，r2_1等變數，其實際數值是由資料最佳化來決定。
建立上方特徵值矩陣時，有時我們會在字庫總字元數加上二個額外的字元，其一代表字庫中從未出現的字元，其二代表padding字元(下例中，我們指派數值15作為padding字元)。前者是因為有時測試組及驗證組會包含一些字元未在訓練組之中，而後者則是因為矩陣運算必須維持向量長度的一致。以下我們使用此例，並用PyTorch的函式產生一組初始embedding矩陣。
torch.manual_seed(1)
embedding = nn.Embedding(num_embeddings=14+2, # nwords + 2
embedding_dim=3,
padding_idx=15)
text_encoded_input = torch.LongTensor([[1, 4, 5, 6, 2,
3, 7, 8, 9, 10,
11, 12, 13, 0]])
print(embedding(text_encoded_input))
# tensor([[[-1.6095, -0.1002, -0.6092],
# [-0.2223, 1.6871, 0.2284],
# [ 0.4676, -0.6970, -1.1608],
# [ 0.6995, 0.1991, 0.8657],
# [-0.9798, -1.6091, -0.7121],
# [ 0.3037, -0.7773, -0.2515],
# [ 0.2444, -0.6629, 0.8073],
# [ 1.1017, -0.1759, -2.2456],
# [-1.4465, 0.0612, -0.6177],
# [-0.7981, -0.1316, 1.8793],
# [-0.0721, 0.1578, -0.7735],
# [ 0.1991, 0.0457, 0.1530],
# [-0.4757, -0.1110, 0.2927],
# [-1.5256, -0.7502, -0.6540]]])
我們可以使用torchtext的vocab函式來建立字頻對照表。
from torchtext.vocab import vocab
from collections import Counter, OrderedDict
token_counts = Counter()
for label, text in train_dataset:
tokens = tokenizer(text)
token_counts.update(tokens)
sorted_by_freq_tuples = sorted(token_counts.items(),
key=lambda x: x[1], reverse=True)
ordered_dict = OrderedDict(sorted_by_freq_tuples)
vocab = vocab(ordered_dict)
# 這兒在對照表表中加入0及1，分別代表padding及未知字元。
vocab.insert_token("<pad>", 0)
vocab.insert_token("<unk>", 1)
vocab.set_default_index(1)
nword = len(token_counts)
print('Vocab-size:', nword)
print(sorted_by_freq_tuples[:3])
PyTorch使用Data loader的資料型態，在定義Data loader時，我們可以定義一個中介函數，將原始文字資料轉換為字頻對照表的整數數值。
import torch.nn as nn
device = torch.device("cuda:0")
text_pipeline = lambda x: [vocab[token] for token in tokenizer(x)]
def collate_batch(batch):
label_list, text_list, lengths = [], [], []
for _label, _text in batch:
label_list.append(_label)
processed_text = torch.tensor(text_pipeline(_text),
dtype=torch.int64)
text_list.append(processed_text)
# 記錄每一條問題的字元數
lengths.append(processed_text.size(0))
label_list = torch.tensor(label_list)
lengths = torch.tensor(lengths)
padded_text_list = nn.utils.rnn.pad_sequence(
text_list, batch_first=True)
return padded_text_list.to(device), label_list.to(device), lengths.to(device)
from torch.utils.data import DataLoader
batch_size = 32
train_dl = DataLoader(train_dataset, batch_size=batch_size,
shuffle=True, collate_fn=collate_batch)
valid_dl = DataLoader(valid_dataset, batch_size=batch_size,
shuffle=False, collate_fn=collate_batch)
test_dl = DataLoader(test_dataset, batch_size=batch_size,
shuffle=False, collate_fn=collate_batch)在將資料轉為PyTorch的DataLoader格式，之後的步驟就時之前的文章所展示的過程類似，設置模型，用訓練組找到最佳化的參數值，之後用測試組評估模型對新資料的推測準確度。
小結
雖然使用了兩層LSTM的內在層，驗證組卻只達到約75%的正確率，和Tensorflow所得到的結果有相當的差距，或許改變節點數、embedding的特徵值數目，或是再增加類神經網絡的層數可以增進RNN模型的推測準確率，另外此樣本數僅有8,000筆，總字元數僅達46000多也許是另一個問題所在。
附錄
RNN模型的 Python 程式
import pickle as pkl
import argparse
import numpy as np
import os
import pathlib
import re
from collections import Counter, OrderedDict
import torch
from torch.utils.data import random_split
from torchtext.vocab import vocab
from torch.utils.data import DataLoader
import torch.nn as nn
def tokenizer(text):
text = re.sub("<[^>]*>", "", text)
emoticons = re.findall("(?::|;|=)(?:-)?(?:\)|\(|D|P)", text.lower())
text = re.sub("[\W]+", " ", text.lower()) + " ".join(emoticons).replace("-", "")
tokenized = text.split()
return tokenized
def collate_batch(batch):
label_list, text_list, lengths = [], [], []
for _label, _text in batch:
label_list.append(_label)
processed_text = torch.tensor(text_pipeline(_text), dtype=torch.int64)
text_list.append(processed_text)
lengths.append(processed_text.size(0))
label_list = torch.tensor(label_list)
lengths = torch.tensor(lengths)
padded_text_list = nn.utils.rnn.pad_sequence(text_list, batch_first=True)
return padded_text_list.to(device), label_list.to(device), lengths.to(device)
class RNN(nn.Module):
def __init__(
self, vocab_size, embed_dim, rnn_hidden_size, fc_hidden_size, num_classes
):
super().__init__()
self.embedding = nn.Embedding(vocab_size + 2, embed_dim, padding_idx=0)
self.rnn1 = nn.LSTM(
embed_dim,
rnn_hidden_size,
batch_first=True,
dropout=0.5,
bidirectional=True,
)
self.rnn2 = nn.LSTM(
rnn_hidden_size * 2,
rnn_hidden_size,
batch_first=True,
dropout=0.5,
bidirectional=True,
)
self.fc1 = nn.Linear(
rnn_hidden_size * 2, fc_hidden_size
) # *2 for bidirectional
self.relu = nn.ReLU()
self.fc2 = nn.Linear(fc_hidden_size, num_classes)
self.softmax = nn.Softmax(dim=1)
def forward(self, text, lengths):
out = self.embedding(text)
out = nn.utils.rnn.pack_padded_sequence(
out, lengths.cpu().numpy(), enforce_sorted=False, batch_first=True
)
out, (hidden, cell) = self.rnn1(out)
out, _ = nn.utils.rnn.pad_packed_sequence(out, batch_first=True)
out = nn.utils.rnn.pack_padded_sequence(
out, lengths.cpu().numpy(), enforce_sorted=False, batch_first=True
)
out, (hidden, cell) = self.rnn2(out)
out = torch.cat(
(hidden[-2, :, :], hidden[-1, :, :]), dim=1
) # Concatenate the outputs from both directions
out = self.fc1(out)
out = self.relu(out)
out = self.fc2(out)
out = self.softmax(out)
return out
def train(dataloader):
model.train()
total_acc, total_loss = 0, 0
for text_batch, label_batch, lengths in dataloader:
optimizer.zero_grad()
pred = model(text_batch, lengths)
predicted_label = torch.argmax(pred, dim=1)
loss = loss_fn(pred, label_batch)
loss.backward()
optimizer.step()
acc_vec = predicted_label == label_batch
total_acc += acc_vec.float().sum().item()
total_loss += loss.item() * label_batch.size(0)
return total_acc / len(dataloader.dataset), total_loss / len(dataloader.dataset)
def evaluate(dataloader):
model.eval()
total_acc, total_loss = 0, 0
with torch.no_grad():
for text_batch, label_batch, lengths in dataloader:
pred = model(text_batch, lengths)
predicted_label = torch.argmax(pred, dim=1)
loss = loss_fn(pred, label_batch)
acc_vec = predicted_label == label_batch
total_acc += acc_vec.float().sum().item()
total_loss += loss.item() * label_batch.size(0)
return total_acc / len(dataloader.dataset), total_loss / len(dataloader.dataset)
if __name__ == "__main__":
# python3 fit_stackoverflow.py "tmp_data/stackoverflow_data.pkl" "models/stackoverflow_model.pth" --batch_size 32 --nepoch 30 --nembed 256
parser = argparse.ArgumentParser(description="Process data")
parser.add_argument("data_path", type=str, help="Path to the data file")
parser.add_argument("weight_path", type=str, help="Path to save weights")
parser.add_argument("--batch_size", type=int, default=32, help="Batch size")
parser.add_argument("--nepoch", type=int, default=30, help="Epoch size")
parser.add_argument("--nembed", type=int, default=128, help="Embedding dimension")
args = parser.parse_args()
DATA_INPUT = args.data_path
WEIGHT_OUTPUT = args.weight_path
batch_size = args.batch_size
nepoch = args.nepoch
embed_dim = args.nembed
with open(DATA_INPUT, "rb") as f:
train_dataset, valid_dataset, test_dataset = pkl.load(f)
## Count
token_counts = Counter()
for label, line in train_dataset:
tokens = tokenizer(line)
token_counts.update(tokens)
nword = len(token_counts)
sorted_by_freq_tuples = sorted(
token_counts.items(), key=lambda x: x[1], reverse=True
)
ordered_dict = OrderedDict(sorted_by_freq_tuples)
vocab = vocab(ordered_dict)
vocab.insert_token("<pad>", 0)
vocab.insert_token("<unk>", 1)
vocab.set_default_index(1)
device = torch.device("cuda:0")
text_pipeline = lambda x: [vocab[token] for token in tokenizer(x)]
train_dl = DataLoader(
train_dataset, batch_size=batch_size, shuffle=True, collate_fn=collate_batch
)
valid_dl = DataLoader(
valid_dataset, batch_size=batch_size, shuffle=False, collate_fn=collate_batch
)
test_dl = DataLoader(
test_dataset, batch_size=batch_size, shuffle=False, collate_fn=collate_batch
)
vocab_size = len(vocab)
rnn_hidden_size = 64
fc_hidden_size = 64
nclass = 4
model = RNN(vocab_size, embed_dim, rnn_hidden_size, fc_hidden_size, nclass)
model = model.to(device)
loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
for epoch in range(nepoch):
acc_train, loss_train = train(train_dl)
acc_valid, loss_valid = evaluate(valid_dl)
print(
f"Epoch {epoch} [accuracy loss]: {acc_train:.3f} {loss_train:.2f}, [val_accuracy loss]: {acc_valid:.3f} {loss_valid:,.2f}"
)
torch.save(model.state_dict(), WEIGHT_OUTPUT)