وبلاگ

وقتی تصمیم گرفتم بیشتر در معماری ترانسفورماتور کاوش کنم، اغلب هنگام خواندن یا تماشای آموزش های آنلاین احساس ناامیدی می کردم، زیرا احساس می کردم همیشه چیزی را از دست می دهند:

• آموزش‌های رسمی Tensorflow یا Pytorch از APIهای خود استفاده می‌کردند، بنابراین در سطح بالایی باقی می‌ماند و من را مجبور می‌کرد تا به پایگاه کد آنها بروم تا ببینم چه چیزی در زیر هود است. زمان زیادی می برد و خواندن 1000 خط کد همیشه آسان نیست.
• سایر آموزش‌های کد سفارشی که پیدا کردم (پیوندهای انتهای مقاله) اغلب موارد استفاده را ساده می‌کنند و به مفاهیمی مانند پوشاندن پردازش دسته‌ای با یک دنباله با طول متغیر نمی‌پردازند.

بنابراین تصمیم گرفتم Transformer خود را بنویسم تا مطمئن شوم که مفاهیم را درک کرده‌ام و می‌توانم از آن با هر مجموعه داده استفاده کنم.

بنابراین، در طول این مقاله ما یک رویکرد روشمند را دنبال می کنیم که در آن ترانسفورماتور را لایه به لایه و بلوک به بلوک اعمال می کنیم.

بدیهی است که بسیاری از پیاده سازی های مختلف و همچنین API های سطح بالا از Pytorch یا Tensorflow در حال حاضر آماده هستند، با – مطمئنم – عملکرد بهتری نسبت به مدلی که ما می خواهیم بسازیم.

“خوب، اما پس چرا از پیاده سازی های TF/Pytorch استفاده نکنیم”؟

هدف این مقاله آموزشی است و من هیچ ادعایی مبنی بر شکست اجرای Pytorch یا Tensorflow ندارم. من معتقدم که تئوری و کد پشت ترانسفورماتورها مشخص نیست، بنابراین امیدوارم که گذراندن این آموزش گام به گام به شما این امکان را بدهد که این مفاهیم را بهتر درک کنید و بعداً هنگام ایجاد کد خود احساس راحتی بیشتری داشته باشید.

دلیل دیگر برای ساخت ترانسفورماتور خود از ابتدا این است که به شما امکان می دهد تا به طور کامل نحوه استفاده از API های فوق را درک کنید. اگر به پیاده سازی Pytorch از forward() در روش کلاس Transformer، کلمات کلیدی مبهم زیادی مانند:

اگر قبلاً با این کلمات کلیدی آشنا هستید، می توانید با خیال راحت از این مقاله صرف نظر کنید.

در غیر این صورت، این مقاله شما را با هر یک از این کلمات کلیدی با مفاهیم اساسی آشنا می کند.

اگر قبلاً نام ChatGPT یا Gemini را شنیده اید، پس قبلاً با یک ترانسفورماتور آشنا شده اید. در واقع “T” ChatGPT مخفف Transformer است.

این معماری برای اولین بار در سال 2017 توسط محققان گوگل در مقاله “توجه تمام چیزی است که نیاز دارید” معرفی شد. این کاملاً انقلابی است، زیرا مدل‌های قبلی برای یادگیری توالی به دنباله (ترجمه ماشینی، گفتار به متن و غیره) به RNN‌ها متکی بودند که از نظر محاسباتی گران بودند به این معنا که باید توالی‌ها را گام به گام پردازش می‌کردند. مرحله، در حالی که ترانسفورمرها فقط باید یک بار به کل دنباله نگاه کنند و پیچیدگی زمانی را از O(n) به O(1) منتقل کنند.

کاربردهای ترانسفورماتورها در زمینه NLP بسیار زیاد است و شامل ترجمه زبان، پاسخگویی به سوالات، خلاصه سازی اسناد، تولید متن و غیره می شود.

ساختار کلی ترانسفورماتور به شرح زیر است:

اولین بلوکی که می‌خواهیم پیاده‌سازی کنیم، در واقع مهمترین بخش Transformer است و توجه چند سر نام دارد. بیایید ببینیم در معماری کلی کجا قرار دارد

توجه مکانیزمی است که واقعاً مختص ترانسفورماتورها نیست و قبلاً در مدل‌های ترتیب به ترتیب RNN استفاده شده است.

import torch
import torch.nn as nn
import math
class MultiHeadAttention(nn.Module):
def __init__(self, hidden_dim=256, num_heads=4):
"""
input_dim: Dimensionality of the input.
num_heads: The number of attention heads to split the input into.
"""
super(MultiHeadAttention, self).__init__()
self.hidden_dim = hidden_dim
self.num_heads = num_heads
assert hidden_dim % num_heads == 0, "Hidden dim must be divisible by num heads"
self.Wv = nn.Linear(hidden_dim, hidden_dim, bias=False) # the Value part
self.Wk = nn.Linear(hidden_dim, hidden_dim, bias=False) # the Key part
self.Wq = nn.Linear(hidden_dim, hidden_dim, bias=False) # the Query part
self.Wo = nn.Linear(hidden_dim, hidden_dim, bias=False) # the output layer
def check_sdpa_inputs(self, x):
assert x.size(1) == self.num_heads, f"Expected size of x to be ({-1, self.num_heads, -1, self.hidden_dim // self.num_heads}), got {x.size()}"
assert x.size(3) == self.hidden_dim // self.num_heads
def scaled_dot_product_attention(
self, 
query, 
key, 
value, 
attention_mask=None, 
key_padding_mask=None):
"""
query : tensor of shape (batch_size, num_heads, query_sequence_length, hidden_dim//num_heads)
key : tensor of shape (batch_size, num_heads, key_sequence_length, hidden_dim//num_heads)
value : tensor of shape (batch_size, num_heads, key_sequence_length, hidden_dim//num_heads)
attention_mask : tensor of shape (query_sequence_length, key_sequence_length)
key_padding_mask : tensor of shape (sequence_length, key_sequence_length)
"""
self.check_sdpa_inputs(query)
self.check_sdpa_inputs(key)
self.check_sdpa_inputs(value)
d_k = query.size(-1)
tgt_len, src_len = query.size(-2), key.size(-2)
# logits = (B, H, tgt_len, E) * (B, H, E, src_len) = (B, H, tgt_len, src_len)
logits = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k) 
# Attention mask here
if attention_mask is not None:
if attention_mask.dim() == 2:
assert attention_mask.size() == (tgt_len, src_len)
attention_mask = attention_mask.unsqueeze(0)
logits = logits + attention_mask
else:
raise ValueError(f"Attention mask size {attention_mask.size()}")
# Key mask here
if key_padding_mask is not None:
key_padding_mask = key_padding_mask.unsqueeze(1).unsqueeze(2) # Broadcast over batch size, num heads
logits = logits + key_padding_mask
attention = torch.softmax(logits, dim=-1)
output = torch.matmul(attention, value) # (batch_size, num_heads, sequence_length, hidden_dim)
return output, attention
def split_into_heads(self, x, num_heads):
batch_size, seq_length, hidden_dim = x.size()
x = x.view(batch_size, seq_length, num_heads, hidden_dim // num_heads)
return x.transpose(1, 2) # Final dim will be (batch_size, num_heads, seq_length, , hidden_dim // num_heads)
def combine_heads(self, x):
batch_size, num_heads, seq_length, head_hidden_dim = x.size()
return x.transpose(1, 2).contiguous().view(batch_size, seq_length, num_heads * head_hidden_dim)
def forward(
self, 
q, 
k, 
v, 
attention_mask=None, 
key_padding_mask=None):
"""
q : tensor of shape (batch_size, query_sequence_length, hidden_dim)
k : tensor of shape (batch_size, key_sequence_length, hidden_dim)
v : tensor of shape (batch_size, key_sequence_length, hidden_dim)
attention_mask : tensor of shape (query_sequence_length, key_sequence_length)
key_padding_mask : tensor of shape (sequence_length, key_sequence_length)
"""
q = self.Wq(q)
k = self.Wk(k)
v = self.Wv(v)
q = self.split_into_heads(q, self.num_heads)
k = self.split_into_heads(k, self.num_heads)
v = self.split_into_heads(v, self.num_heads)
# attn_values, attn_weights = self.multihead_attn(q, k, v, attn_mask=attention_mask)
attn_values, attn_weights  = self.scaled_dot_product_attention(
query=q, 
key=k, 
value=v, 
attention_mask=attention_mask,
key_padding_mask=key_padding_mask,
)
grouped = self.combine_heads(attn_values)
output = self.Wo(grouped)
self.attention_weigths = attn_weights
return output

در اینجا لازم است چند مفهوم را توضیح دهیم.

1) پرس و جوها، کلیدها و مقادیر.

این درخواست اطلاعاتی است که می‌خواهید مطابقت دهید،
این کلید و ارزش های اطلاعات ذخیره شده هستند.

مانند استفاده از فرهنگ لغت فکر کنید: وقتی از فرهنگ لغت پایتون استفاده می کنید، اگر پرس و جو شما با کلیدهای فرهنگ لغت مطابقت نداشته باشد، چیزی برگردانده نخواهد شد. اما اگر بخواهیم فرهنگ لغت ما ترکیبی از اطلاعات را که نسبتاً نزدیک است را برگرداند، چه؟ انگار داشتیم:

d = {"panther": 1, "bear": 10, "dog":3}
d["wolf"] = 0.2*d["panther"] + 0.7*d["dog"] + 0.1*d["bear"]

در هسته آن، این تمرکز است: نگاه کردن به بخش های مختلف داده های خود و ترکیب آنها با یکدیگر برای تولید یک ترکیب به عنوان پاسخ به درخواست شما.

بخش مربوط به کد جایی است که ما وزن توجه بین پرس و جو و کلیدها را محاسبه می کنیم

logits = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k) # we compute the weights of attention

و موردی که در آن وزن های نرمال شده را به مقادیر اعمال می کنیم:

attention = torch.softmax(logits, dim=-1)
output = torch.matmul(attention, value) # (batch_size, num_heads, sequence_length, hidden_dim)

2) پوشش توجه و بالشتک

وقتی به بخش‌هایی از یک ورودی متوالی توجه می‌کنیم، نمی‌خواهیم اطلاعات بی‌فایده یا ممنوعه را وارد کنیم.

اطلاعات بی فایده، به عنوان مثال، padding است: نمادهای padding که برای تراز کردن همه دنباله‌ها در یک دسته با اندازه توالی یکسانی استفاده می‌شوند باید توسط مدل ما نادیده گرفته شوند. در بخش پایانی به این موضوع باز خواهیم گشت

اطلاعات ممنوعه کمی پیچیده تر است. هنگامی که مدل آموزش داده می شود، یاد می گیرد که توالی ورودی را رمزگذاری کند و اهداف را با ورودی ها تراز کند. با این حال، از آنجایی که فرآیند استنتاج شامل نگاه کردن به نشانه‌های منتشر شده قبلی برای پیش‌بینی مورد بعدی است (به تولید متن در ChatGPT فکر کنید)، باید قوانین مشابهی را در طول آموزش اعمال کنیم.

به همین دلیل است که الف را اجرا می کنیم ماسک علّی اطمینان حاصل شود که اهداف، در هر مرحله زمانی، فقط می توانند اطلاعات گذشته را ببینند. در اینجا بخش مربوطه است که در آن ماسک اعمال می شود (محاسبه ماسک در پایان پوشش داده شده است)

if attention_mask is not None:
if attention_mask.dim() == 2:
assert attention_mask.size() == (tgt_len, src_len)
attention_mask = attention_mask.unsqueeze(0)
logits = logits + attention_mask

مربوط به قسمت زیر از ترانسفورماتور است:

هنگام دریافت و پردازش ورودی، ترانسفورماتور هیچ حس نظمی ندارد زیرا به دنباله به عنوان یک کل نگاه می کند، برخلاف آنچه که RNN ها انجام می دهند. بنابراین، باید یک سری زمانی اضافه کنیم تا ترانسفورماتور بتواند وابستگی ها را یاد بگیرد.

جزئیات خاص نحوه عملکرد رمزگذاری موقعیتی فراتر از حوصله این مقاله است، اما برای اطلاع از این موضوع به راحتی مقاله اصلی را بخوانید.

# Taken from https://pytorch.org/tutorials/beginner/transformer_tutorial.html#define-the-model
class PositionalEncoding(nn.Module):
def __init__(self, d_model, dropout=0.1, max_len=5000):
super(PositionalEncoding, self).__init__()
self.dropout = nn.Dropout(p=dropout)
pe = torch.zeros(max_len, d_model)
position = torch.arange(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)
self.register_buffer('pe', pe)
def forward(self, x):
"""
Arguments:
x: Tensor, shape ``[batch_size, seq_len, embedding_dim]``
"""
x = x + self.pe[:, :x.size(1), :]
return x

ما به داشتن یک رمزگذار کامل نزدیک می شویم! رمزگذار سمت چپ ترانسفورماتور است

ما بخش کوچکی را به کد خود اضافه می کنیم که قسمت عبور به جلو است:

class PositionWiseFeedForward(nn.Module):
def __init__(self, d_model: int, d_ff: int):
super(PositionWiseFeedForward, self).__init__()
self.fc1 = nn.Linear(d_model, d_ff)
self.fc2 = nn.Linear(d_ff, d_model)
self.relu = nn.ReLU()
def forward(self, x):
return self.fc2(self.relu(self.fc1(x)))

با کنار هم قرار دادن قطعات، یک ماژول Encoder دریافت می کنیم!

class EncoderBlock(nn.Module):
def __init__(self, n_dim: int, dropout: float, n_heads: int):
super(EncoderBlock, self).__init__()
self.mha = MultiHeadAttention(hidden_dim=n_dim, num_heads=n_heads)
self.norm1 = nn.LayerNorm(n_dim)
self.ff = PositionWiseFeedForward(n_dim, n_dim)
self.norm2 = nn.LayerNorm(n_dim)
self.dropout = nn.Dropout(dropout)
def forward(self, x, src_padding_mask=None):
assert x.ndim==3, "Expected input to be 3-dim, got {}".format(x.ndim)
att_output = self.mha(x, x, x, key_padding_mask=src_padding_mask)
x = x + self.dropout(self.norm1(att_output))
ff_output = self.ff(x)
output = x + self.norm2(ff_output)
return output

همانطور که در نمودار نشان داده شده است، رمزگذار در واقع حاوی N بلوک یا لایه رمزگذار و همچنین یک لایه جاسازی برای ورودی های ما است. بنابراین بیایید با افزودن بلوک‌های Embedding، Positional Encoder و Encoder یک Encoder ایجاد کنیم:

class Encoder(nn.Module):
def __init__(
self, 
vocab_size: int, 
n_dim: int, 
dropout: float, 
n_encoder_blocks: int,
n_heads: int):
super(Encoder, self).__init__()
self.n_dim = n_dim
self.embedding = nn.Embedding(
num_embeddings=vocab_size, 
embedding_dim=n_dim
)
self.positional_encoding = PositionalEncoding(
d_model=n_dim, 
dropout=dropout
)    
self.encoder_blocks = nn.ModuleList([
EncoderBlock(n_dim, dropout, n_heads) for _ in range(n_encoder_blocks)
])
def forward(self, x, padding_mask=None):
x = self.embedding(x) * math.sqrt(self.n_dim)
x = self.positional_encoding(x)
for block in self.encoder_blocks:
x = block(x=x, src_padding_mask=padding_mask)
return x

قسمت رسیور قسمت سمت چپ است و نیاز به کمی کار بیشتر دارد.

یه چیزی هست به اسم توجه چند وجهی پوشانده شده است. آنچه را که قبلا گفتیم به خاطر بسپار ماسک علّی ? خوب این چیزی است که اینجا اتفاق می افتد. ما از پارامتر Attention_mask ماژول توجه چند سر خود برای نشان دادن این استفاده خواهیم کرد (جزئیات بیشتر در مورد نحوه محاسبه ماسک در پایان):

# Stuff before
self.self_attention = MultiHeadAttention(hidden_dim=n_dim, num_heads=n_heads)
masked_att_output = self.self_attention(
q=tgt, 
k=tgt, 
v=tgt, 
attention_mask=tgt_mask, <-- HERE IS THE CAUSAL MASK
key_padding_mask=tgt_padding_mask)
# Stuff after

ملاحظه دوم نام دارد توجه متقاطع. از پرس و جوی رمزگشا برای مطابقت با کلید رمزگذار و مقادیر استفاده می کند! احتیاط: طول آنها در طول تمرین می تواند متفاوت باشد، بنابراین به طور کلی تمرین خوبی است که به وضوح شکل های مورد انتظار ورودی ها را به شرح زیر تعریف کنید:

def scaled_dot_product_attention(
self, 
query, 
key, 
value, 
attention_mask=None, 
key_padding_mask=None):
"""
query : tensor of shape (batch_size, num_heads, query_sequence_length, hidden_dim//num_heads)
key : tensor of shape (batch_size, num_heads, key_sequence_length, hidden_dim//num_heads)
value : tensor of shape (batch_size, num_heads, key_sequence_length, hidden_dim//num_heads)
attention_mask : tensor of shape (query_sequence_length, key_sequence_length)
key_padding_mask : tensor of shape (sequence_length, key_sequence_length)
"""

و در اینجا قسمتی که از خروجی رمزگذار استفاده می کنیم نامیده می شود حافظهبا ورودی رمزگشای ما:

# Stuff before
self.cross_attention = MultiHeadAttention(hidden_dim=n_dim, num_heads=n_heads)
cross_att_output = self.cross_attention(
q=x1, 
k=memory, 
v=memory, 
attention_mask=None,  <-- NO CAUSAL MASK HERE
key_padding_mask=memory_padding_mask) <-- WE NEED TO USE THE PADDING OF THE SOURCE
# Stuff after

با کنار هم قرار دادن قطعات، در نهایت برای رمزگشا به این می رسیم:

class DecoderBlock(nn.Module):
def __init__(self, n_dim: int, dropout: float, n_heads: int):
super(DecoderBlock, self).__init__()
# The first Multi-Head Attention has a mask to avoid looking at the future
self.self_attention = MultiHeadAttention(hidden_dim=n_dim, num_heads=n_heads)
self.norm1 = nn.LayerNorm(n_dim)
# The second Multi-Head Attention will take inputs from the encoder as key/value inputs
self.cross_attention = MultiHeadAttention(hidden_dim=n_dim, num_heads=n_heads)
self.norm2 = nn.LayerNorm(n_dim)
self.ff = PositionWiseFeedForward(n_dim, n_dim)
self.norm3 = nn.LayerNorm(n_dim)
# self.dropout = nn.Dropout(dropout)
def forward(self, tgt, memory, tgt_mask=None, tgt_padding_mask=None, memory_padding_mask=None):
masked_att_output = self.self_attention(
q=tgt, k=tgt, v=tgt, attention_mask=tgt_mask, key_padding_mask=tgt_padding_mask)
x1 = tgt + self.norm1(masked_att_output)
cross_att_output = self.cross_attention(
q=x1, k=memory, v=memory, attention_mask=None, key_padding_mask=memory_padding_mask)
x2 = x1 + self.norm2(cross_att_output)
ff_output = self.ff(x2)
output = x2 + self.norm3(ff_output)
return output
class Decoder(nn.Module):
def __init__(
self, 
vocab_size: int, 
n_dim: int, 
dropout: float, 
max_seq_len: int,
n_decoder_blocks: int,
n_heads: int):
super(Decoder, self).__init__()
self.embedding = nn.Embedding(
num_embeddings=vocab_size, 
embedding_dim=n_dim
)
self.positional_encoding = PositionalEncoding(
d_model=n_dim, 
dropout=dropout
)
self.decoder_blocks = nn.ModuleList([
DecoderBlock(n_dim, dropout, n_heads) for _ in range(n_decoder_blocks)
])
def forward(self, tgt, memory, tgt_mask=None, tgt_padding_mask=None, memory_padding_mask=None):
x = self.embedding(tgt)
x = self.positional_encoding(x)
for block in self.decoder_blocks:
x = block(x, memory, tgt_mask=tgt_mask, tgt_padding_mask=tgt_padding_mask, memory_padding_mask=memory_padding_mask)
return x

بخش توجه چند فصل قبل را به خاطر بسپارید که در آن به خاموش کردن بخش‌های خاصی از ورودی‌ها در هنگام توجه اشاره کردیم.

در طول آموزش، دسته‌ای از ورودی‌ها و اهداف را در نظر می‌گیریم که هر نمونه از آن‌ها می‌تواند دارای طول متغیر باشد. مثال زیر را در نظر بگیرید که در آن ما 4 کلمه را گروه بندی می کنیم: موز، هندوانه، گلابی، بلوبری. برای پردازش آنها به عنوان یک گروه، باید همه کلمات را بر اساس طول طولانی ترین کلمه (هندوانه) تراز کنیم. بنابراین ما یک کاراکتر اضافی، PAD، به هر کلمه اضافه می کنیم تا طول همه آنها به اندازه یک هندوانه باشد.

در تصویر زیر، جدول بالا نشان دهنده داده های خام و جدول پایین نشان دهنده نسخه کدگذاری شده است:

در مورد ما، ما می خواهیم شاخص های padding را از وزن توجه محاسبه شده حذف کنیم. بنابراین، ما می توانیم یک ماسک را برای داده های منبع و هدف به صورت زیر محاسبه کنیم:

padding_mask = (x == PAD_IDX)

حالا ماسک های علت را چطور؟ خوب، اگر بخواهیم در هر مرحله زمانی، مدل فقط بتواند از مراحل گذشته بازدید کند، این بدان معناست که برای هر مرحله زمانی T، مدل فقط می تواند هر مرحله t را برای t در 1…T بازدید کند. این یک حلقه برای دو برابر است، بنابراین می توانیم از یک ماتریس برای محاسبه آن استفاده کنیم:

def generate_square_subsequent_mask(size: int):
"""Generate a triangular (size, size) mask. From PyTorch docs."""
mask = (1 - torch.triu(torch.ones(size, size), diagonal=1)).bool()
mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))
return mask

بیایید اکنون ترانسفورماتور خود را با کنار هم قرار دادن قطعات بسازیم!

در مورد استفاده ما، از یک مجموعه داده بسیار ساده برای نشان دادن نحوه یادگیری ترانسفورماتورها استفاده خواهیم کرد.

اما چرا از ترانسفورماتور برای معکوس کردن کلمات استفاده کنیم؟ من قبلاً می دانم که چگونه این کار را در پایتون با word انجام دهم[::-1] !»

هدف در اینجا این است که ببینیم آیا مکانیسم توجه ترانسفورماتور کار می کند یا خیر. آنچه ما انتظار داریم این است که وقتی یک دنباله ورودی داده می شود، شاهد تغییر وزن های توجه از راست به چپ باشیم. اگر چنین است، به این معنی است که Transformer ما یک دستور زبان بسیار ساده را یاد گرفته است که فقط از راست به چپ می‌خواند، و می‌تواند هنگام ترجمه زبان واقعی به گرامرهای پیچیده‌تر تعمیم دهد.

بیایید ابتدا با کلاس Transformer سفارشی خود شروع کنیم:

import torch
import torch.nn as nn
import math
from .encoder import Encoder
from .decoder import Decoder
class Transformer(nn.Module):
def __init__(self, **kwargs):
super(Transformer, self).__init__()
for k, v in kwargs.items():
print(f" * {k}={v}")
self.vocab_size = kwargs.get('vocab_size')
self.model_dim = kwargs.get('model_dim')
self.dropout = kwargs.get('dropout')
self.n_encoder_layers = kwargs.get('n_encoder_layers')
self.n_decoder_layers = kwargs.get('n_decoder_layers')
self.n_heads = kwargs.get('n_heads')
self.batch_size = kwargs.get('batch_size')
self.PAD_IDX = kwargs.get('pad_idx', 0)
self.encoder = Encoder(
self.vocab_size, self.model_dim, self.dropout, self.n_encoder_layers, self.n_heads)
self.decoder = Decoder(
self.vocab_size, self.model_dim, self.dropout, self.n_decoder_layers, self.n_heads)
self.fc = nn.Linear(self.model_dim, self.vocab_size)
@staticmethod    
def generate_square_subsequent_mask(size: int):
"""Generate a triangular (size, size) mask. From PyTorch docs."""
mask = (1 - torch.triu(torch.ones(size, size), diagonal=1)).bool()
mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))
return mask
def encode(
self, 
x: torch.Tensor, 
) -> torch.Tensor:
"""
Input
x: (B, S) with elements in (0, C) where C is num_classes
Output
(B, S, E) embedding
"""
mask = (x == self.PAD_IDX).float()
encoder_padding_mask = mask.masked_fill(mask == 1, float('-inf'))
# (B, S, E)
encoder_output = self.encoder(
x, 
padding_mask=encoder_padding_mask
)  
return encoder_output, encoder_padding_mask
def decode(
self, 
tgt: torch.Tensor, 
memory: torch.Tensor, 
memory_padding_mask=None
) -> torch.Tensor:
"""
B = Batch size
S = Source sequence length
L = Target sequence length
E = Model dimension
Input
encoded_x: (B, S, E)
y: (B, L) with elements in (0, C) where C is num_classes
Output
(B, L, C) logits
"""
mask = (tgt == self.PAD_IDX).float()
tgt_padding_mask = mask.masked_fill(mask == 1, float('-inf'))
decoder_output = self.decoder(
tgt=tgt, 
memory=memory, 
tgt_mask=self.generate_square_subsequent_mask(tgt.size(1)), 
tgt_padding_mask=tgt_padding_mask, 
memory_padding_mask=memory_padding_mask,
)  
output = self.fc(decoder_output)  # shape (B, L, C)
return output
def forward(
self, 
x: torch.Tensor, 
y: torch.Tensor, 
) -> torch.Tensor:
"""
Input
x: (B, Sx) with elements in (0, C) where C is num_classes
y: (B, Sy) with elements in (0, C) where C is num_classes
Output
(B, L, C) logits
"""
# Encoder output shape (B, S, E)
encoder_output, encoder_padding_mask = self.encode(x)  
# Decoder output shape (B, L, C)
decoder_output = self.decode(
tgt=y, 
memory=encoder_output, 
memory_padding_mask=encoder_padding_mask
)  
return decoder_output

استنباط با رمزگشایی حریصانه

باید روشی اضافه کنیم که مانند روش شناخته شده عمل کند model.predict در scikit.learn. هدف این است که از مدل بخواهیم به صورت پویا پیش بینی هایی را با توجه به یک ورودی ارائه دهد. در طول استنتاج، هیچ هدفی وجود ندارد: مدل با انتشار یک نشانه، توجه به خروجی شروع می‌کند و از پیش‌بینی خود برای ادامه انتشار نشانه‌ها استفاده می‌کند. به همین دلیل است که این مدل ها اغلب مدل های خودرگرسیون نامیده می شوند زیرا از پیش بینی های گذشته برای پیش بینی آینده استفاده می کنند.

مشکل رمزگشایی حریصانه این است که در هر مرحله به توکن با بیشترین احتمال نگاه می کند. در صورتی که اولین توکن ها کاملا اشتباه باشند، این می تواند به پیش بینی های بسیار بدی منجر شود. روش‌های رمزگشایی دیگری مانند جستجوی پرتو وجود دارد که به فهرست کوتاهی از دنباله‌های نامزد نگاه می‌کند (به جای argmax، توکن‌های top-k را در هر مرحله ذخیره کنید) و دنباله را با بیشترین احتمال کل برمی‌گرداند.

در حال حاضر، بیایید رمزگشایی حریصانه را پیاده سازی کنیم و آن را به مدل ترانسفورماتور خود اضافه کنیم:

def predict(
self,
x: torch.Tensor,
sos_idx: int=1,
eos_idx: int=2,
max_length: int=None
) -> torch.Tensor:
"""
Method to use at inference time. Predict y from x one token at a time. This method is greedy
decoding. Beam search can be used instead for a potential accuracy boost.
Input
x: str
Output
(B, L, C) logits
"""
# Pad the tokens with beginning and end of sentence tokens
x = torch.cat([
torch.tensor([sos_idx]), 
x, 
torch.tensor([eos_idx])]
).unsqueeze(0)
encoder_output, mask = self.transformer.encode(x) # (B, S, E)
if not max_length:
max_length = x.size(1)
outputs = torch.ones((x.size()[0], max_length)).type_as(x).long() * sos_idx
for step in range(1, max_length):
y = outputs[:, :step]
probs = self.transformer.decode(y, encoder_output)
output = torch.argmax(probs, dim=-1)
# Uncomment if you want to see step by step predicitons
# print(f"Knowing {y} we output {output[:, -1]}")
if output[:, -1].detach().numpy() in (eos_idx, sos_idx):
break
outputs[:, step] = output[:, -1]
return outputs

داده های اسباب بازی ایجاد کنید

ما یک مجموعه داده کوچک را تعریف می کنیم که کلمات را معکوس می کند، به این معنی که “helloworld” “dlrowolleh” را برمی گرداند:

import numpy as np
import torch
from torch.utils.data import Dataset
np.random.seed(0)
def generate_random_string():
len = np.random.randint(10, 20)
return "".join([chr(x) for x in np.random.randint(97, 97+26, len)])
class ReverseDataset(Dataset):
def __init__(self, n_samples, pad_idx, sos_idx, eos_idx):
super(ReverseDataset, self).__init__()
self.pad_idx = pad_idx
self.sos_idx = sos_idx
self.eos_idx = eos_idx
self.values = [generate_random_string() for _ in range(n_samples)]
self.labels = [x[::-1] for x in self.values]
def __len__(self):
return len(self.values)  # number of samples in the dataset
def __getitem__(self, index):
return self.text_transform(self.values[index].rstrip("\n")), \
self.text_transform(self.labels[index].rstrip("\n"))
def text_transform(self, x):
return torch.tensor([self.sos_idx] + [ord(z)-97+3 for z in x] + [self.eos_idx]

اکنون مراحل آموزش و ارزیابی را تعریف می کنیم:

PAD_IDX = 0
SOS_IDX = 1
EOS_IDX = 2
def train(model, optimizer, loader, loss_fn, epoch):
model.train()
losses = 0
acc = 0
history_loss = []
history_acc = [] 
with tqdm(loader, position=0, leave=True) as tepoch:
for x, y in tepoch:
tepoch.set_description(f"Epoch {epoch}")
optimizer.zero_grad()
logits = model(x, y[:, :-1])
loss = loss_fn(logits.contiguous().view(-1, model.vocab_size), y[:, 1:].contiguous().view(-1))
loss.backward()
optimizer.step()
losses += loss.item()
preds = logits.argmax(dim=-1)
masked_pred = preds * (y[:, 1:]!=PAD_IDX)
accuracy = (masked_pred == y[:, 1:]).float().mean()
acc += accuracy.item()
history_loss.append(loss.item())
history_acc.append(accuracy.item())
tepoch.set_postfix(loss=loss.item(), accuracy=100. * accuracy.item())
return losses / len(list(loader)), acc / len(list(loader)), history_loss, history_acc
def evaluate(model, loader, loss_fn):
model.eval()
losses = 0
acc = 0
history_loss = []
history_acc = [] 
for x, y in tqdm(loader, position=0, leave=True):
logits = model(x, y[:, :-1])
loss = loss_fn(logits.contiguous().view(-1, model.vocab_size), y[:, 1:].contiguous().view(-1))
losses += loss.item()
preds = logits.argmax(dim=-1)
masked_pred = preds * (y[:, 1:]!=PAD_IDX)
accuracy = (masked_pred == y[:, 1:]).float().mean()
acc += accuracy.item()
history_loss.append(loss.item())
history_acc.append(accuracy.item())
return losses / len(list(loader)), acc / len(list(loader)), history_loss, history_acc

و الگو را برای چندین دوره آموزش دهید:

def collate_fn(batch):
""" 
This function pads inputs with PAD_IDX to have batches of equal length
"""
src_batch, tgt_batch = [], []
for src_sample, tgt_sample in batch:
src_batch.append(src_sample)
tgt_batch.append(tgt_sample)
src_batch = pad_sequence(src_batch, padding_value=PAD_IDX, batch_first=True)
tgt_batch = pad_sequence(tgt_batch, padding_value=PAD_IDX, batch_first=True)
return src_batch, tgt_batch
# Model hyperparameters
args = {
'vocab_size': 128,
'model_dim': 128,
'dropout': 0.1,
'n_encoder_layers': 1,
'n_decoder_layers': 1,
'n_heads': 4
}
# Define model here
model = Transformer(**args)
# Instantiate datasets
train_iter = ReverseDataset(50000, pad_idx=PAD_IDX, sos_idx=SOS_IDX, eos_idx=EOS_IDX)
eval_iter = ReverseDataset(10000, pad_idx=PAD_IDX, sos_idx=SOS_IDX, eos_idx=EOS_IDX)
dataloader_train = DataLoader(train_iter, batch_size=256, collate_fn=collate_fn)
dataloader_val = DataLoader(eval_iter, batch_size=256, collate_fn=collate_fn)
# During debugging, we ensure sources and targets are indeed reversed
# s, t = next(iter(dataloader_train))
# print(s[:4, ...])
# print(t[:4, ...])
# print(s.size())
# Initialize model parameters
for p in model.parameters():
if p.dim() > 1:
nn.init.xavier_uniform_(p)
# Define loss function : we ignore logits which are padding tokens
loss_fn = torch.nn.CrossEntropyLoss(ignore_index=PAD_IDX)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001, betas=(0.9, 0.98), eps=1e-9)
# Save history to dictionnary
history = {
'train_loss': [],
'eval_loss': [],
'train_acc': [],
'eval_acc': []
}
# Main loop
for epoch in range(1, 4):
start_time = time.time()
train_loss, train_acc, hist_loss, hist_acc = train(model, optimizer, dataloader_train, loss_fn, epoch)
history['train_loss'] += hist_loss
history['train_acc'] += hist_acc
end_time = time.time()
val_loss, val_acc, hist_loss, hist_acc = evaluate(model, dataloader_val, loss_fn)
history['eval_loss'] += hist_loss
history['eval_acc'] += hist_acc
print((f"Epoch: {epoch}, Train loss: {train_loss:.3f}, Train acc: {train_acc:.3f}, Val loss: {val_loss:.3f}, Val acc: {val_acc:.3f} "f"Epoch time = {(end_time - start_time):.3f}s"))

توجه را تجسم کنید

ما یک تابع کوچک برای دسترسی به وزن سرهای توجه تعریف می کنیم:

fig = plt.figure(figsize=(10., 10.))
images = model.decoder.decoder_blocks[0].cross_attention.attention_weigths[0,...].detach().numpy()
grid = ImageGrid(fig, 111,  # similar to subplot(111)
nrows_ncols=(2, 2),  # creates 2x2 grid of axes
axes_pad=0.1,  # pad between axes in inch.
)
for ax, im in zip(grid, images):
# Iterating over the grid returns the Axes.
ax.imshow(im)

هنگام خواندن وزنه ها از بالا می توانیم یک الگوی زیبا از راست به چپ ببینیم. قسمت های عمودی در پایین محور y قطعاً ممکن است وزن های پوشانده شده را به دلیل پوشش ماسک نشان دهند

تست مدل ما!

برای آزمایش مدل خود با داده های جدید، کمی تعریف می کنیم Translator کلاس برای کمک به ما در رمزگشایی:

class Translator(nn.Module):
def __init__(self, transformer):
super(Translator, self).__init__()
self.transformer = transformer
@staticmethod
def str_to_tokens(s):
return [ord(z)-97+3 for z in s]
@staticmethod
def tokens_to_str(tokens):
return "".join([chr(x+94) for x in tokens])
def __call__(self, sentence, max_length=None, pad=False):
x = torch.tensor(self.str_to_tokens(sentence))
outputs = self.transformer.predict(sentence)
return self.tokens_to_str(outputs[0])

شما باید بتوانید موارد زیر را ببینید:

و اگر سر توجه را چاپ کنیم موارد زیر را رعایت می کنیم:

fig = plt.figure()
images = model.decoder.decoder_blocks[0].cross_attention.attention_weigths[0,...].detach().numpy().mean(axis=0)
fig, ax = plt.subplots(1,1, figsize=(10., 10.))
# Iterating over the grid returs the Axes.
ax.set_yticks(range(len(out)))
ax.set_xticks(range(len(sentence)))
ax.xaxis.set_label_position('top') 
ax.set_xticklabels(iter(sentence))
ax.set_yticklabels([f"step {i}" for i in range(len(out))])
ax.imshow(images)

وقتی جمله “reversethis” خود را برعکس می کنیم، به وضوح می بینیم که الگو از راست به چپ حرکت می کند! (مرحله 0 در واقع شروع نشانگر جمله را می گیرد).

تمام است، اکنون می توانید Transformer را بنویسید و از آن با مجموعه داده های بزرگتر برای انجام ترجمه ماشینی ایجاد BERT خود استفاده کنید.

من می‌خواستم این آموزش اخطارهای نوشتن Transformer را به شما نشان دهد: padding و masking احتمالاً قسمت‌هایی هستند که بیشترین توجه را می‌طلبند (جناسی در نظر گرفته شده)، زیرا تعیین می‌کنند که مدل در طول استنتاج چقدر خوب عمل می‌کند.

در مقالات بعدی، نحوه ایجاد مدل BERT خود و نحوه استفاده از Equinox، یک کتابخانه با کارایی بالا در بالای JAX را بررسی خواهیم کرد.

روی خط بمانید!

(+) “ترانسفورماتور مشروح”
(+) “ترانسفورماتورها از ابتدا”
(+) “ترجمه ماشین عصبی با ترانسفورماتور و کراس”
(+) “ترانسفورماتور مصور”
(+) آموزش مطالعات پیشرفته دانشگاه آمستردام
(+) آموزش Pytorch برای Transformers