昨天你學過 LSTM但你肯定還搞不清楚它到底在做什麼,而今天我會帶你從零手刻一個雙層 LSTM,並套用在經典的 IMDB 影評情緒分類任務中。這篇重點不在於訓練出一個超強、超快的模型,而是幫你搞懂 LSTM 的真正運作邏輯與自然語言處理的最基礎,從Embedding、到手動實作 forward 傳播流程,通通自己動手來一遍。
而今天重點會在 Padding 怎麼處理才不會影響模型學習?為何要自己 Embedding 是在幹嘛用的?BiLSTM 又是怎麼同時捕捉前後文語意的?透過完整拆解與 PyTorch 實作,我們要的不只是會用,而是真正理解每一層背後的運算邏輯與設計。
IMDB 資料集是情感分析任務中最經典的語料之一,通常包含兩欄:一欄是使用者撰寫的電影評論(review
),另一欄是該評論的情感標籤(sentiment
),標示為 positive
或 negative
。
我們之所以能從中分析情緒,是因為使用者在撰寫評論時往往會自然表達感受,例如「好看」、「無聊」等詞語隱含了正面或負面的情緒傾向。這些語言特徵能被電腦模型學習,透過大量標註資料建立語意與情感之間的對應關係,進而判斷新評論的情緒屬性。
檔案連結:點我
因此第一步就是對資料進行前處理,其中我們需要將這些文字標籤轉換成數值型態,例如把 positive
映射成 1,negative
映射成 0,這樣做的目的是為了讓模型能夠計算損失函數,而在這裡由於我把檔案轉換成csv文件,因此我們會使用 pandas
來讀取其檔案,並用values
轉換成numpy格式。
import pandas as pd
df = pd.read_csv('imdb_data.csv')
reviews = df['review'].values
sentiments = df['sentiment'].values
sentiments = (sentiments == 'positive').astype(int) # positive→1, negative→0
print(f'review: {reviews[0][:30]}...\nsentiment label:{sentiments[0]}')
在資料前處理方面,雖然建議先移除評論中的 HTML 標籤或特殊換行符號,以提升模型的穩定性與表現,但由於這在本任務中並非重點,因此此處先略過相關處理。
接下來我們要讓文字變成電腦看得懂的格式,也就是把它轉成 ID 序列。這裡我們直接用 Hugging Face 提供的 AutoTokenizer
,選的是 bert-base-uncased
這個模型所搭配的 WordPiece 斷詞器(有時候大家口語上會叫它 BPE,但嚴格來說 BERT 用的是 WordPiece),這個 tokenizer 會幫你做好以下事情:
單字(Word)
拆成子詞(subword)
;from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
input_datas = tokenizer(
reviews[:2].tolist(),
max_length=10,
truncation=True,
padding="longest",
return_tensors='pt'
)
print('Tokenizer 輸出:')
print(input_datas)
print('還原文字:')
print(tokenizer.decode(input_datas['input_ids'][0]))
print(tokenizer.decode(input_datas['input_ids'][1]))
當我們把句子丟進 tokenizer
後,它會輸出像這樣的結果:
Tokenizer輸出:
{'input_ids': tensor([[ 101, 22953, 2213, 4381, 2152, 2003, 1037, 9476, 4038, 102],
[ 101, 11573, 2791, 1006, 2030, 2160, 24913, 2004, 2577, 102]]),
'token_type_ids': tensor([[0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]),
'attention_mask': tensor([[1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]])}
這裡的 input_ids
就是把原始文字轉成了一串 ID,也就是模型可以理解的形式。像 [CLS]
是 101、[SEP]
是 102每個單字或子詞也會對應到唯一的編號。token_type_ids
則是用來區分兩個句子的標記(像做句子配對任務時會用到),而 attention_mask
則是用來告訴模型哪些位置是實際的字哪些只是 padding。而當我們用 decode()
還原這些 ID,可以看到結果像這樣:
還原文字:
[CLS] bromwell high is a cartoon comedy [SEP]
[CLS] homelessness ( or houselessness as george [SEP]
通常我們會加上像 truncation=True
、padding="longest"
這類參數,是為了讓每筆輸入長度一致、不會爆記憶體,又能有效利用資源。而 decode(...)
中如果不想看到[CLS]、[SEP]被還原出來還可以加上 skip_special_tokens=True
,還可以讓還原結果更乾淨,看起來就像單純的原始句子。
這是我們第一次真的動手做 DataLoader,所以邊做邊解釋一下。PyTorch 裡的 Dataset
是用來定義一筆資料長什麼樣子,而 DataLoader
則是負責怎麼把多筆資料湊成一個 batch。
在用 DataLoader 的時候它會先透過 __getitem__
回傳原始的文字跟標籤,再透過 __len__
來知道整筆資料有多長,判斷什麼時候跑完。通常除了 Dataset
跟 DataLoader
,我們還會搭配 collate_fn
一起用。
這個 collate_fn
的功能就是在每次組 batch 的時候,可以動態地處理資料,像是做 padding 或是隨機資料增強之類的操作都會放在這裡。所以我們這邊也是把真正的斷詞放在 collate_fn
裡面做,這樣可以一次處理整個 batch 的文本。
from sklearn.model_selection import train_test_split
from torch.utils.data import Dataset, DataLoader
import torch
class IMDB(Dataset):
def __init__(self, x, y, tokenizer):
self.x = x
self.y = y
self.tokenizer = tokenizer
def __getitem__(self, index):
return self.x[index], self.y[index]
def __len__(self):
return len(self.x)
至於 collate_fn
的實際做法,其實就是根據 __getitem__
回傳的 (text, label)
把資料拿出來做進一步處理。在這裡我們會先把文字轉成張量,然後把 [CLS]
跟 [SEP]
這兩個 token 拿掉(這兩個是什麼意思我們後面會講),只留下中間的 token。最後我們會回傳一個字典,裡面包含 input_ids
跟 labels
。
為什麼要用字典格式呢?因為我們在 Day 10 做的訓練器設計是動態的,也就是說它會根據這些 key 自動抓對應的輸入來餵模型。所以這邊這兩個欄位,剛好就是模型訓練時要用的兩個輸入。
def collate_fn(self, batch):
batch_x, batch_y = zip(*batch)
ids = self.tokenizer(
batch_x,
max_length=128,
truncation=True,
padding="longest",
return_tensors='pt'
).input_ids
# 移除 [CLS] 與 [SEP](通常在頭尾)
input_ids = ids[:, 1:-1]
labels = torch.tensor(batch_y, dtype=torch.long)
return { 'input_ids': input_ids, 'labels': labels }
在切分資料跟建立 DataLoader 的時候,為什麼在 Windows 上 num_workers
一定要設成 0 呢?
因為 Windows 跑多線程(multi-processing)
預設是用 spawn 的方式開子程序,簡單來說就是它會重新執行一次主程式。如果你的程式沒有包在 if __name__ == "__main__":
裡就很容易整個炸掉,出現奇怪的錯誤。
x_train, x_valid, y_train, y_valid = train_test_split(
reviews, sentiments, train_size=0.8, random_state=46, shuffle=True
)
trainset = IMDB(x_train, y_train, tokenizer)
validset = IMDB(x_valid, y_valid, tokenizer)
valid_loader = DataLoader(
validset, batch_size=32, shuffle=True, num_workers=0,
pin_memory=True, collate_fn=validset.collate_fn
)
train_loader = DataLoader(
trainset, batch_size=32, shuffle=True, num_workers=0,
pin_memory=True, collate_fn=trainset.collate_fn
)
當然如果你之後是在 Linux 或 WSL 環境下跑的話,就可以放心把 num_workers
調高,加速資料載入速度。
在自然語言處理(NLP)中,Embedding 是一個非常關鍵的模型層。它的本質其實就是一張「vocab_size × emb_dim
」的表格,也可以想成是一個詞彙查詢表。每個 token(詞或字)對應到表格中的一列向量,而這個向量的長度就是 emb_dim
,通常是模型設定的參數。
這些向量在一開始是隨機初始化的,也就是說每個 token 剛開始都只是亂數對應到某個位置。但在模型訓練的過程中,這些向量會不斷被調整。最終模型會學到讓語意相近的 token 對應到相近的向量,也就是說,它能幫助模型理解詞與詞之間的語意關係。
你可以把 Embedding 想成一個查表的機制只要輸入 token 的 ID,就能快速查到對應的語意向量,而這張表不只是存資料,更會隨著模型學習自動調整。在實作的時候有個小細節要特別注意:為了讓每個 batch 裡的句子長度一樣,我們會在比較短的句子後面補上一些 padding token。但這些 padding 其實只是拿來對齊格式,它們本身沒有任何語意。
所以當我們把這些 token 丟進 embedding 裡時,不能讓它們產生真正的特徵值。也就是說,padding 對應的那一列向量應該是全 0,而且在訓練過程中也不能被更新。
import torch
import torch.nn as nn
import torch.nn.functional as F
class MyEmbedding(nn.Module):
def __init__(self, num_embeddings, embedding_dim, padding_idx=None):
super().__init__()
self.weight = nn.Parameter(torch.empty(num_embeddings, embedding_dim))
self.padding_idx = padding_idx
nn.init.normal_(self.weight, mean=0.0, std=embedding_dim ** -0.5)
if padding_idx is not None:
with torch.no_grad():
self.weight[padding_idx].zero_()
# 確保反傳不更新 padding 列
self.weight.register_hook(self._zero_pad_grad)
為了做到這件事,我們會幫模型加一個小小的 hook,裡面用 _zero_pad_grad
這個做法,在反向傳播的時候手動把 padding_idx
對應那一列的梯度清成 0。像是 self.weight[padding_idx].zero_()
這行,就是確保這個 padding 向量一開始就是 0,而且以後也不會被動到。這樣模型在訓練時就不會誤以為 padding 有什麼實際意義,能讓學到的語意表示更乾淨。
def _zero_pad_grad(self, grad):
if self.padding_idx is None:
return grad
grad = grad.clone()
grad[self.padding_idx].zero_()
return grad
def forward(self, input_ids): # [B,T] -> [B,T,E]
out = self.weight[input_ids]
if self.padding_idx is not None:
mask = (input_ids == self.padding_idx).unsqueeze(-1)
out = out.masked_fill(mask, 0.0)
return out
那有人可能會問,這跟直接用 nn.Embedding 有什麼不一樣? 其實我們這邊手動實作,是為了讓大家更清楚地看到怎麼處理 padding_idx,還有搭配 mask 的邏輯。畢竟這些細節在實際應用中很重要,理解原理才能靈活調整。但如果是實務上其實直接用 PyTorch 內建的 nn.Embedding(num_embeddings, embedding_dim, padding_idx=...) 就可以了,不用自己手動寫。
padding token 沒有實際語意,那麼對於像 LSTM 這種一個時間步一個時間步處理的模型,我們該怎麼讓它**跳過 padding 呢?**這邊我們就用昨天的 LSTM進行改寫讓大家看得更清楚。
class LinearTanhLSTMCore(nn.Module):
def __init__(self, input_size, hidden_size):
super().__init__()
H = hidden_size
self.x_i = nn.Linear(input_size, H, bias=True)
self.x_f = nn.Linear(input_size, H, bias=True)
self.x_g = nn.Linear(input_size, H, bias=True)
self.x_o = nn.Linear(input_size, H, bias=True)
self.h_i = nn.Linear(H, H, bias=True)
self.h_f = nn.Linear(H, H, bias=True)
self.h_g = nn.Linear(H, H, bias=True)
self.h_o = nn.Linear(H, H, bias=True)
self.hidden_size = H
for m in self.modules():
if isinstance(m, nn.Linear):
nn.init.xavier_uniform_(m.weight)
nn.init.zeros_(m.bias)
因此我們要看forward時每個時間步我們都會看一下當下這個 token 是不是 padding,這是靠 mask
來判斷的。有效的 token 對應到 mask
裡是 1,padding 的地方則是 0。
def forward(self, x, mask=None, h0=None, c0=None):
B, T, _ = x.shape
H = self.hidden_size
device = x.device
h = torch.zeros(B, H, device=device) if h0 is None else h0
c = torch.zeros(B, H, device=device) if c0 is None else c0
if mask is None:
mask = torch.ones(B, T, dtype=torch.bool, device=device)
for t in range(T):
xt = x[:, t, :]
valid = mask[:, t].unsqueeze(1).to(x.dtype) # [B,1] 1.0 有效, 0.0 padding
i = torch.sigmoid(self.x_i(xt) + self.h_i(h))
f = torch.sigmoid(self.x_f(xt) + self.h_f(h))
g = torch.tanh( self.x_g(xt) + self.h_g(h))
o = torch.sigmoid(self.x_o(xt) + self.h_o(h))
c_new = f * c + i * g
h_new = o * torch.tanh(c_new)
# 只在有效 token 上更新狀態
h = valid * h_new + (1.0 - valid) * h
c = valid * c_new + (1.0 - valid) * c
return h, (h, c)
所以我們在更新 LSTM 的狀態(hidden state 跟 cell state)時,就能根據這個 mask
決定要不要更新。如果是 padding,那我們就保留原本的狀態,不讓它參與學習。這樣模型就能專注在真正有內容的部分,不會被 padding 影響。簡單來說就是在每個時間步都問一句:「這個 token 有沒有意義?」如果沒有,就當作沒看到,LSTM 的狀態維持原樣不變。
同樣的這樣的做法雖然不是最快的,只是為了展示基礎原理,在實務上如果要快可以搭配Pytorch的 PackedSequence 或其他更進階的做法來優化。
當我們在處理自然語言時,一個詞的意思常常不只是看它本身,而是要搭配前後文一起理解。但如果模型只能從頭讀到尾,就可能錯過那種「看到後面才恍然大悟」的情況。這時候,雙向 LSTM(BiLSTM)就派上用場了。它的做法是一邊用正向 LSTM 從前面讀過去,一邊用反向 LSTM 從後面讀回來,然後把這兩邊的資訊合起來,讓模型能同時抓住上下文的意思。
這時候如果我們只用單向的 LSTM,也就是模型只能從左讀到右(正向),那它就只能看到目前 token 的過去,沒辦法預測或理解後面可能發生的事。
而在實作方式上我們可以用剛剛建立的兩個 LinearTanhLSTMCore
各自跑完,把兩個 LSTM 的最後 hidden state 接起來最後丟進一個全連接層,做二分類任務
class BiLSTMClassifier(nn.Module):
def __init__(self, vocab_size, emb_dim, hidden_size, num_classes=2, padding_idx=0, dropout=0.2):
super().__init__()
self.embed = MyEmbedding(vocab_size, emb_dim, padding_idx=padding_idx)
self.fwd = LinearTanhLSTMCore(emb_dim, hidden_size)
self.bwd = LinearTanhLSTMCore(emb_dim, hidden_size)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size * 2, num_classes)
nn.init.xavier_uniform_(self.fc.weight)
nn.init.zeros_(self.fc.bias)
self.padding_idx = padding_idx
self.criterion = nn.CrossEntropyLoss() # 二分類用 CE 配 2-logits
def forward(self, input_ids, labels=None): # input_ids: [B,T], labels: [B]
mask = input_ids != self.padding_idx # [B,T] bool
x = self.embed(input_ids) # [B,T,E]
h_fwd, _ = self.fwd(x, mask=mask) # [B,H]
x_rev = torch.flip(x, dims=[1])
mask_rev = torch.flip(mask, dims=[1])
h_bwd, _ = self.bwd(x_rev, mask=mask_rev) # [B,H]
h_cat = torch.cat([h_fwd, h_bwd], dim=1) # [B,2H]
logits = self.fc(self.dropout(h_cat)) # [B,2]
loss = None
if labels is not None:
loss = self.criterion(logits, labels) # labels: int64 in {0,1}
return loss, logits
model = BiLSTMClassifier(
vocab_size=len(tokenizer),
emb_dim=256,
hidden_size=128,
num_classes=2,
padding_idx=tokenizer.pad_token_id or 0,
dropout=0.2
)
這裡我們把 loss 放在輸出的第 0 個位置,預測結果放在第 2 個位置,這樣設計是為了配合我們之前設定好的訓練器。另外,輸入的 input_ids
跟 labels
的名稱也不能亂改,因為這些名稱是根據 Dataloader 產生的索引來對應資料的。如果換了名字,整個流程就對不上了。
到了這一步我們就是把資料跟模型交給你現成提供的 Trainer 來訓練就好。至於優化器的部分,直接用大家最常用的 Adam 起手式就可以了簡單又夠用。
from trainer import Trainer
import torch.optim as optim
optimizer = optim.Adam(model.parameters(), lr=1e-3)
trainer = Trainer(
epochs=100,
train_loader=train_loader,
valid_loader=valid_loader,
model=model,
optimizer=optimizer,
)
trainer.train(show_loss=false)
輸出結果:
Train Epoch 0: 100%|██████████| 1250/1250 [17:25<00:00, 1.20it/s, loss=0.487]
Valid Epoch 0: 100%|██████████| 313/313 [01:19<00:00, 3.96it/s, loss=0.249]
Saving Model With Loss 0.44163
Train Loss: 0.50866 | Valid Loss: 0.44163 | Best Loss: 0.44163
Train Epoch 1: 100%|██████████| 1250/1250 [17:21<00:00, 1.20it/s, loss=0.412]
Valid Epoch 1: 100%|██████████| 313/313 [01:21<00:00, 3.83it/s, loss=0.495]
Saving Model With Loss 0.38657
Train Loss: 0.31322 | Valid Loss: 0.38657 | Best Loss: 0.38657
Train Epoch 2: 100%|██████████| 1250/1250 [17:16<00:00, 1.21it/s, loss=0.113]
Valid Epoch 2: 100%|██████████| 313/313 [01:20<00:00, 3.90it/s, loss=0.144]
Train Loss: 0.19454 | Valid Loss: 0.45054 | Best Loss: 0.38657
在訓練的過程中,有幾個實用的小建議可以參考。首先要注意觀察 train
和 valid loss
是否出現分岔的情況,如果可能代表模型過擬合,這時可以在模型加入 dropout
或設定 優化器的weight_decay
來做正規化。或是我們用trainer
的grad_clip
防止梯度爆炸的問題。
現在你應該已經更了解LSTM這個模型到底在做什麼了。今天我們也順便介紹了Embedding的架構,這其實就是自然語言處理裡的核心基礎之一。不過在AI模型的應用裡,除了我們常見的分類模型之外,還有一種很重要的生成式模型,明天我會帶你一步步理解一個簡單的文字生成框架,讓你知道怎麼讓模型「寫出文字」。