import tiktoken import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader

class GPTDatasetV1(Dataset): def init(self, txt, tokenizer, max_length, stride): self.input_ids = [] self.target_ids = []

Tokenize the entire text

token_ids = tokenizer.encode(txt, allowed_special={“<|endoftext|>”})

Use a sliding window to chunk the book into overlapping sequences of max_length

for i in range(0, len(token_ids) - max_length, stride): input_chunk = token_ids[i:i + max_length] target_chunk = token_ids[i + 1: i + max_length + 1] self.input_ids.append(torch.tensor(input_chunk)) self.target_ids.append(torch.tensor(target_chunk))

def len(self): return len(self.input_ids)

def getitem(self, idx): return self.input_ids[idx], self.target_ids[idx]

def create_dataloader_v1(txt, batch_size=4, max_length=256, stride=128, shuffle=True, drop_last=True, num_workers=0):

Initialize the tokenizer

tokenizer = tiktoken.get_encoding(“gpt2”)

Create dataset

dataset = GPTDatasetV1(txt, tokenizer, max_length, stride)

Create dataloader

dataloader = DataLoader( dataset, batch_size=batch_size, shuffle=shuffle, drop_last=drop_last, num_workers=num_workers)

return dataloader

class MultiHeadAttention(nn.Module): def init(self, d_in, d_out, context_length, dropout, num_heads, qkv_bias=False): super().init() assert d_out % num_heads == 0, “d_out must be divisible by n_heads”

self.d_out = d_out self.num_heads = num_heads self.head_dim = d_out // num_heads # Reduce the projection dim to match desired output dim

self.W_query = nn.Linear(d_in, d_out, bias=qkv_bias) self.W_key = nn.Linear(d_in, d_out, bias=qkv_bias) self.W_value = nn.Linear(d_in, d_out, bias=qkv_bias) self.out_proj = nn.Linear(d_out, d_out) # Linear layer to combine head outputs self.dropout = nn.Dropout(dropout) self.register_buffer(‘mask’, torch.triu(torch.ones(context_length, context_length), diagonal=1))

def forward(self, x): b, num_tokens, d_in = x.shape

keys = self.W_key(x) # Shape: (b, num_tokens, d_out) queries = self.W_query(x) values = self.W_value(x)

We implicitly split the matrix by adding a num_heads dimension

Unroll last dim: (b, num_tokens, d_out) -> (b, num_tokens, num_heads, head_dim)

keys = keys.view(b, num_tokens, self.num_heads, self.head_dim) values = values.view(b, num_tokens, self.num_heads, self.head_dim) queries = queries.view(b, num_tokens, self.num_heads, self.head_dim)

Transpose: (b, num_tokens, num_heads, head_dim) -> (b, num_heads, num_tokens, head_dim)

keys = keys.transpose(1, 2) queries = queries.transpose(1, 2) values = values.transpose(1, 2)

Compute scaled dot-product attention (aka self-attention) with a causal mask

attn_scores = queries @ keys.transpose(2, 3) # Dot product for each head

Original mask truncated to the number of tokens and converted to boolean

mask_bool = self.mask.bool()[:num_tokens, :num_tokens]

Use the mask to fill attention scores

attn_scores.masked_fill_(mask_bool, -torch.inf)

attn_weights = torch.softmax(attn_scores / keys.shape[-1]**0.5, dim=-1) attn_weights = self.dropout(attn_weights)

Shape: (b, num_tokens, num_heads, head_dim)

context_vec = (attn_weights @ values).transpose(1, 2)

Combine heads, where self.d_out = self.num_heads * self.head_dim

context_vec = context_vec.reshape(b, num_tokens, self.d_out) context_vec = self.out_proj(context_vec) # optional projection

return context_vec

class LayerNorm(nn.Module): def init(self, emb_dim): super().init() self.eps = 1e-5 self.scale = nn.Parameter(torch.ones(emb_dim)) self.shift = nn.Parameter(torch.zeros(emb_dim))

def forward(self, x): mean = x.mean(dim=-1, keepdim=True) var = x.var(dim=-1, keepdim=True, unbiased=False) norm_x = (x - mean) / torch.sqrt(var + self.eps) return self.scale * norm_x + self.shift

class GELU(nn.Module): def init(self): super().init()

def forward(self, x): return 0.5 * x * (1 + torch.tanh( torch.sqrt(torch.tensor(2.0 / torch.pi)) * (x + 0.044715 * torch.pow(x, 3)) ))

class FeedForward(nn.Module): def init(self, cfg): super().init() self.layers = nn.Sequential( nn.Linear(cfg[“emb_dim”], 4 * cfg[“emb_dim”]), GELU(), nn.Linear(4 * cfg[“emb_dim”], cfg[“emb_dim”]), )

def forward(self, x): return self.layers(x)

class TransformerBlock(nn.Module): def init(self, cfg): super().init() self.att = MultiHeadAttention( d_in=cfg[“emb_dim”], d_out=cfg[“emb_dim”], context_length=cfg[“context_length”], num_heads=cfg[“n_heads”], dropout=cfg[“drop_rate”], qkv_bias=cfg[“qkv_bias”]) self.ff = FeedForward(cfg) self.norm1 = LayerNorm(cfg[“emb_dim”]) self.norm2 = LayerNorm(cfg[“emb_dim”]) self.drop_shortcut = nn.Dropout(cfg[“drop_rate”])

def forward(self, x):

Shortcut connection for attention block

shortcut = x x = self.norm1(x) x = self.att(x) # Shape [batch_size, num_tokens, emb_size] x = self.drop_shortcut(x) x = x + shortcut # Add the original input back

Shortcut connection for feed-forward block

shortcut = x x = self.norm2(x) x = self.ff(x) x = self.drop_shortcut(x) x = x + shortcut # Add the original input back

return x

class GPTModel(nn.Module): def init(self, cfg): super().init() self.tok_emb = nn.Embedding(cfg[“vocab_size”], cfg[“emb_dim”]) self.pos_emb = nn.Embedding(cfg[“context_length”], cfg[“emb_dim”]) self.drop_emb = nn.Dropout(cfg[“drop_rate”])

self.trf_blocks = nn.Sequential( *[TransformerBlock(cfg) for _ in range(cfg[“n_layers”])])

self.final_norm = LayerNorm(cfg[“emb_dim”]) self.out_head = nn.Linear(cfg[“emb_dim”], cfg[“vocab_size”], bias=False)

def forward(self, in_idx): batch_size, seq_len = in_idx.shape tok_embeds = self.tok_emb(in_idx) pos_embeds = self.pos_emb(torch.arange(seq_len, device=in_idx.device)) x = tok_embeds + pos_embeds # Shape [batch_size, num_tokens, emb_size] x = self.drop_emb(x) x = self.trf_blocks(x) x = self.final_norm(x) logits = self.out_head(x) return logits

</details>
```python
# Download contents to train the data with
import os
import urllib.request

file_path = "the-verdict.txt"
url = "https://raw.githubusercontent.com/rasbt/LLMs-from-scratch/main/ch02/01_main-chapter-code/the-verdict.txt"

if not os.path.exists(file_path):
with urllib.request.urlopen(url) as response:
text_data = response.read().decode('utf-8')
with open(file_path, "w", encoding="utf-8") as file:
file.write(text_data)
else:
with open(file_path, "r", encoding="utf-8") as file:
text_data = file.read()

total_characters = len(text_data)
tokenizer = tiktoken.get_encoding("gpt2")
total_tokens = len(tokenizer.encode(text_data))

print("Data downloaded")
print("Characters:", total_characters)
print("Tokens:", total_tokens)

# Model initialization
GPT_CONFIG_124M = {
"vocab_size": 50257,   # Vocabulary size
"context_length": 256, # Shortened context length (orig: 1024)
"emb_dim": 768,        # Embedding dimension
"n_heads": 12,         # Number of attention heads
"n_layers": 12,        # Number of layers
"drop_rate": 0.1,      # Dropout rate
"qkv_bias": False      # Query-key-value bias
}

torch.manual_seed(123)
model = GPTModel(GPT_CONFIG_124M)
model.eval()
print ("Model initialized")


# Functions to transform from tokens to ids and from to ids to tokens
def text_to_token_ids(text, tokenizer):
encoded = tokenizer.encode(text, allowed_special={'<|endoftext|>'})
encoded_tensor = torch.tensor(encoded).unsqueeze(0) # add batch dimension
return encoded_tensor

def token_ids_to_text(token_ids, tokenizer):
flat = token_ids.squeeze(0) # remove batch dimension
return tokenizer.decode(flat.tolist())



# Define loss functions
def calc_loss_batch(input_batch, target_batch, model, device):
input_batch, target_batch = input_batch.to(device), target_batch.to(device)
logits = model(input_batch)
loss = torch.nn.functional.cross_entropy(logits.flatten(0, 1), target_batch.flatten())
return loss


def calc_loss_loader(data_loader, model, device, num_batches=None):
total_loss = 0.
if len(data_loader) == 0:
return float("nan")
elif num_batches is None:
num_batches = len(data_loader)
else:
# Reduce the number of batches to match the total number of batches in the data loader
# if num_batches exceeds the number of batches in the data loader
num_batches = min(num_batches, len(data_loader))
for i, (input_batch, target_batch) in enumerate(data_loader):
if i < num_batches:
loss = calc_loss_batch(input_batch, target_batch, model, device)
total_loss += loss.item()
else:
break
return total_loss / num_batches


# Apply Train/validation ratio and create dataloaders
train_ratio = 0.90
split_idx = int(train_ratio * len(text_data))
train_data = text_data[:split_idx]
val_data = text_data[split_idx:]

torch.manual_seed(123)

train_loader = create_dataloader_v1(
train_data,
batch_size=2,
max_length=GPT_CONFIG_124M["context_length"],
stride=GPT_CONFIG_124M["context_length"],
drop_last=True,
shuffle=True,
num_workers=0
)

val_loader = create_dataloader_v1(
val_data,
batch_size=2,
max_length=GPT_CONFIG_124M["context_length"],
stride=GPT_CONFIG_124M["context_length"],
drop_last=False,
shuffle=False,
num_workers=0
)


# Sanity checks
if total_tokens * (train_ratio) < GPT_CONFIG_124M["context_length"]:
print("Not enough tokens for the training loader. "
"Try to lower the `GPT_CONFIG_124M['context_length']` or "
"increase the `training_ratio`")

if total_tokens * (1-train_ratio) < GPT_CONFIG_124M["context_length"]:
print("Not enough tokens for the validation loader. "
"Try to lower the `GPT_CONFIG_124M['context_length']` or "
"decrease the `training_ratio`")

print("Train loader:")
for x, y in train_loader:
print(x.shape, y.shape)

print("\nValidation loader:")
for x, y in val_loader:
print(x.shape, y.shape)

train_tokens = 0
for input_batch, target_batch in train_loader:
train_tokens += input_batch.numel()

val_tokens = 0
for input_batch, target_batch in val_loader:
val_tokens += input_batch.numel()

print("Training tokens:", train_tokens)
print("Validation tokens:", val_tokens)
print("All tokens:", train_tokens + val_tokens)


# Indicate the device to use
if torch.cuda.is_available():
device = torch.device("cuda")
elif torch.backends.mps.is_available():
device = torch.device("mps")
else:
device = torch.device("cpu")

print(f"Using {device} device.")

model.to(device) # no assignment model = model.to(device) necessary for nn.Module classes



# Pre-calculate losses without starting yet
torch.manual_seed(123) # For reproducibility due to the shuffling in the data loader

with torch.no_grad(): # Disable gradient tracking for efficiency because we are not training, yet
train_loss = calc_loss_loader(train_loader, model, device)
val_loss = calc_loss_loader(val_loader, model, device)

print("Training loss:", train_loss)
print("Validation loss:", val_loss)


# Functions to train the data
def train_model_simple(model, train_loader, val_loader, optimizer, device, num_epochs,
eval_freq, eval_iter, start_context, tokenizer):
# Initialize lists to track losses and tokens seen
train_losses, val_losses, track_tokens_seen = [], [], []
tokens_seen, global_step = 0, -1

# Main training loop
for epoch in range(num_epochs):
model.train()  # Set model to training mode

for input_batch, target_batch in train_loader:
optimizer.zero_grad() # Reset loss gradients from previous batch iteration
loss = calc_loss_batch(input_batch, target_batch, model, device)
loss.backward() # Calculate loss gradients
optimizer.step() # Update model weights using loss gradients
tokens_seen += input_batch.numel()
global_step += 1

# Optional evaluation step
if global_step % eval_freq == 0:
train_loss, val_loss = evaluate_model(
model, train_loader, val_loader, device, eval_iter)
train_losses.append(train_loss)
val_losses.append(val_loss)
track_tokens_seen.append(tokens_seen)
print(f"Ep {epoch+1} (Step {global_step:06d}): "
f"Train loss {train_loss:.3f}, Val loss {val_loss:.3f}")

# Print a sample text after each epoch
generate_and_print_sample(
model, tokenizer, device, start_context
)

return train_losses, val_losses, track_tokens_seen


def evaluate_model(model, train_loader, val_loader, device, eval_iter):
model.eval()
with torch.no_grad():
train_loss = calc_loss_loader(train_loader, model, device, num_batches=eval_iter)
val_loss = calc_loss_loader(val_loader, model, device, num_batches=eval_iter)
model.train()
return train_loss, val_loss


def generate_and_print_sample(model, tokenizer, device, start_context):
model.eval()
context_size = model.pos_emb.weight.shape[0]
encoded = text_to_token_ids(start_context, tokenizer).to(device)
with torch.no_grad():
token_ids = generate_text(
model=model, idx=encoded,
max_new_tokens=50, context_size=context_size
)
decoded_text = token_ids_to_text(token_ids, tokenizer)
print(decoded_text.replace("\n", " "))  # Compact print format
model.train()


# Start training!
import time
start_time = time.time()

torch.manual_seed(123)
model = GPTModel(GPT_CONFIG_124M)
model.to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=0.0004, weight_decay=0.1)

num_epochs = 10
train_losses, val_losses, tokens_seen = train_model_simple(
model, train_loader, val_loader, optimizer, device,
num_epochs=num_epochs, eval_freq=5, eval_iter=5,
start_context="Every effort moves you", tokenizer=tokenizer
)

end_time = time.time()
execution_time_minutes = (end_time - start_time) / 60
print(f"Training completed in {execution_time_minutes:.2f} minutes.")



# Show graphics with the training process
import matplotlib.pyplot as plt
from matplotlib.ticker import MaxNLocator
import math
def plot_losses(epochs_seen, tokens_seen, train_losses, val_losses):
fig, ax1 = plt.subplots(figsize=(5, 3))
ax1.plot(epochs_seen, train_losses, label="Training loss")
ax1.plot(
epochs_seen, val_losses, linestyle="-.", label="Validation loss"
)
ax1.set_xlabel("Epochs")
ax1.set_ylabel("Loss")
ax1.legend(loc="upper right")
ax1.xaxis.set_major_locator(MaxNLocator(integer=True))
ax2 = ax1.twiny()
ax2.plot(tokens_seen, train_losses, alpha=0)
ax2.set_xlabel("Tokens seen")
fig.tight_layout()
plt.show()

# Compute perplexity from the loss values
train_ppls = [math.exp(loss) for loss in train_losses]
val_ppls = [math.exp(loss) for loss in val_losses]
# Plot perplexity over tokens seen
plt.figure()
plt.plot(tokens_seen, train_ppls, label='Training Perplexity')
plt.plot(tokens_seen, val_ppls, label='Validation Perplexity')
plt.xlabel('Tokens Seen')
plt.ylabel('Perplexity')
plt.title('Perplexity over Training')
plt.legend()
plt.show()

epochs_tensor = torch.linspace(0, num_epochs, len(train_losses))
plot_losses(epochs_tensor, tokens_seen, train_losses, val_losses)


torch.save({
"model_state_dict": model.state_dict(),
"optimizer_state_dict": optimizer.state_dict(),
},
"/tmp/model_and_optimizer.pth"
)

Functions to transform text <–> ids

이것은 어휘에서 텍스트를 ID로, 그리고 그 반대로 변환하는 데 사용할 수 있는 몇 가지 간단한 함수입니다. 이는 텍스트 처리의 시작과 예측의 끝에서 필요합니다:

# Functions to transform from tokens to ids and from to ids to tokens
def text_to_token_ids(text, tokenizer):
encoded = tokenizer.encode(text, allowed_special={'<|endoftext|>'})
encoded_tensor = torch.tensor(encoded).unsqueeze(0) # add batch dimension
return encoded_tensor

def token_ids_to_text(token_ids, tokenizer):
flat = token_ids.squeeze(0) # remove batch dimension
return tokenizer.decode(flat.tolist())

텍스트 생성 함수

이전 섹션에서는 가장 가능성이 높은 토큰을 로짓을 얻은 후에 가져오는 함수가 있었습니다. 그러나 이는 각 입력에 대해 항상 동일한 출력을 생성하게 되어 매우 결정적입니다.

다음 generate_text 함수는 top-k, temperature 및 multinomial 개념을 적용합니다.

• **top-k**는 상위 k개의 토큰을 제외한 모든 토큰의 확률을 -inf로 줄이기 시작한다는 것을 의미합니다. 따라서 k=3인 경우, 결정을 내리기 전에 가장 가능성이 높은 3개의 토큰만 -inf가 아닌 확률을 가집니다.
• **temperature**는 모든 확률이 온도 값으로 나누어진다는 것을 의미합니다. 값이 0.1이면 가장 높은 확률이 가장 낮은 확률에 비해 개선되며, 예를 들어 온도가 5이면 더 평평해집니다. 이는 LLM이 가지길 원하는 응답의 변화를 개선하는 데 도움이 됩니다.
• 온도를 적용한 후, softmax 함수가 다시 적용되어 남아 있는 모든 토큰의 총 확률이 1이 되도록 합니다.
• 마지막으로, 가장 큰 확률을 가진 토큰을 선택하는 대신, 함수 **multinomial**이 최종 확률에 따라 다음 토큰을 예측하는 데 적용됩니다. 따라서 토큰 1이 70%의 확률을 가졌다면, 토큰 2는 20%, 토큰 3은 10%의 확률을 가지며, 70%의 경우 토큰 1이 선택되고, 20%의 경우 토큰 2가, 10%의 경우 토큰 3이 선택됩니다.

# Generate text function
def generate_text(model, idx, max_new_tokens, context_size, temperature=0.0, top_k=None, eos_id=None):

# For-loop is the same as before: Get logits, and only focus on last time step
for _ in range(max_new_tokens):
idx_cond = idx[:, -context_size:]
with torch.no_grad():
logits = model(idx_cond)
logits = logits[:, -1, :]

# New: Filter logits with top_k sampling
if top_k is not None:
# Keep only top_k values
top_logits, _ = torch.topk(logits, top_k)
min_val = top_logits[:, -1]
logits = torch.where(logits < min_val, torch.tensor(float("-inf")).to(logits.device), logits)

# New: Apply temperature scaling
if temperature > 0.0:
logits = logits / temperature

# Apply softmax to get probabilities
probs = torch.softmax(logits, dim=-1)  # (batch_size, context_len)

# Sample from the distribution
idx_next = torch.multinomial(probs, num_samples=1)  # (batch_size, 1)

# Otherwise same as before: get idx of the vocab entry with the highest logits value
else:
idx_next = torch.argmax(logits, dim=-1, keepdim=True)  # (batch_size, 1)

if idx_next == eos_id:  # Stop generating early if end-of-sequence token is encountered and eos_id is specified
break

# Same as before: append sampled index to the running sequence
idx = torch.cat((idx, idx_next), dim=1)  # (batch_size, num_tokens+1)

return idx

Tip

top-k의 일반적인 대안은 top-p로, 핵심 샘플링이라고도 하며, 가장 높은 확률을 가진 k 샘플을 얻는 대신, 결과로 나온 어휘를 확률에 따라 정리하고 가장 높은 확률부터 가장 낮은 확률까지 합산하여 임계값에 도달할 때까지 진행합니다.

그런 다음, 상대 확률에 따라 어휘의 단어들만 고려됩니다.

이는 각 경우에 따라 최적의 k가 다를 수 있으므로 k 샘플의 수를 선택할 필요 없이 오직 임계값만 필요하게 합니다.

이 개선 사항은 이전 코드에 포함되어 있지 않음을 유의하세요.

Tip

생성된 텍스트를 개선하는 또 다른 방법은 이 예제에서 사용된 탐욕적 검색 대신 Beam search를 사용하는 것입니다.
탐욕적 검색과 달리, 각 단계에서 가장 확률이 높은 다음 단어를 선택하고 단일 시퀀스를 구축하는 대신, beam search는 각 단계에서 상위 𝑘 k의 점수가 높은 부분 시퀀스(이른바 “beams”)를 추적합니다. 여러 가능성을 동시에 탐색함으로써 효율성과 품질의 균형을 맞추어, 탐욕적 접근 방식으로 인해 조기 비최적 선택으로 놓칠 수 있는 더 나은 전체 시퀀스를 찾을 가능성을 높입니다.

이 개선 사항은 이전 코드에 포함되어 있지 않음을 유의하세요.

Loss functions

calc_loss_batch 함수는 단일 배치의 예측에 대한 교차 엔트로피를 계산합니다.
**calc_loss_loader**는 모든 배치의 교차 엔트로피를 가져와서 평균 교차 엔트로피를 계산합니다.

# Define loss functions
def calc_loss_batch(input_batch, target_batch, model, device):
input_batch, target_batch = input_batch.to(device), target_batch.to(device)
logits = model(input_batch)
loss = torch.nn.functional.cross_entropy(logits.flatten(0, 1), target_batch.flatten())
return loss

def calc_loss_loader(data_loader, model, device, num_batches=None):
total_loss = 0.
if len(data_loader) == 0:
return float("nan")
elif num_batches is None:
num_batches = len(data_loader)
else:
# Reduce the number of batches to match the total number of batches in the data loader
# if num_batches exceeds the number of batches in the data loader
num_batches = min(num_batches, len(data_loader))
for i, (input_batch, target_batch) in enumerate(data_loader):
if i < num_batches:
loss = calc_loss_batch(input_batch, target_batch, model, device)
total_loss += loss.item()
else:
break
return total_loss / num_batches

Tip

그래디언트 클리핑은 훈련 안정성을 향상시키기 위해 큰 신경망에서 그래디언트 크기에 대한 최대 임계값을 설정하는 기술입니다. 그래디언트가 이 미리 정의된 max_norm을 초과하면, 모델의 매개변수에 대한 업데이트가 관리 가능한 범위 내에 유지되도록 비례적으로 축소되어 폭발하는 그래디언트와 같은 문제를 방지하고 보다 통제되고 안정적인 훈련을 보장합니다.

이 개선 사항은 이전 코드에 포함되어 있지 않음을 유의하십시오.

다음 예제를 확인하십시오:

데이터 로딩

함수 create_dataloader_v1와 create_dataloader_v1는 이전 섹션에서 이미 논의되었습니다.

여기서 90%의 텍스트가 훈련에 사용되고 10%가 검증에 사용된다는 점을 주목하십시오. 두 세트는 2개의 서로 다른 데이터 로더에 저장됩니다.
때때로 데이터 세트의 일부는 모델 성능을 더 잘 평가하기 위해 테스트 세트로 남겨지기도 합니다.

두 데이터 로더는 동일한 배치 크기, 최대 길이, 스트라이드 및 작업자 수(이 경우 0)를 사용하고 있습니다.
주요 차이점은 각 데이터 로더에서 사용하는 데이터이며, 검증자는 마지막 데이터를 버리지 않으며 검증 목적에 필요하지 않기 때문에 데이터를 섞지 않습니다.

또한 스트라이드가 컨텍스트 길이만큼 크다는 것은 훈련 데이터에 사용되는 컨텍스트 간에 겹침이 없음을 의미합니다(과적합을 줄이지만 훈련 데이터 세트도 줄입니다).

더욱이, 이 경우 배치 크기가 2로 설정되어 데이터를 2개의 배치로 나누며, 이의 주요 목표는 병렬 처리를 허용하고 배치당 소비를 줄이는 것입니다.

train_ratio = 0.90
split_idx = int(train_ratio * len(text_data))
train_data = text_data[:split_idx]
val_data = text_data[split_idx:]

torch.manual_seed(123)

train_loader = create_dataloader_v1(
train_data,
batch_size=2,
max_length=GPT_CONFIG_124M["context_length"],
stride=GPT_CONFIG_124M["context_length"],
drop_last=True,
shuffle=True,
num_workers=0
)

val_loader = create_dataloader_v1(
val_data,
batch_size=2,
max_length=GPT_CONFIG_124M["context_length"],
stride=GPT_CONFIG_124M["context_length"],
drop_last=False,
shuffle=False,
num_workers=0
)

Sanity Checks

목표는 훈련을 위한 충분한 토큰이 있는지, 형태가 예상한 대로인지 확인하고, 훈련 및 검증에 사용된 토큰 수에 대한 정보를 얻는 것입니다:

# Sanity checks
if total_tokens * (train_ratio) < GPT_CONFIG_124M["context_length"]:
print("Not enough tokens for the training loader. "
"Try to lower the `GPT_CONFIG_124M['context_length']` or "
"increase the `training_ratio`")

if total_tokens * (1-train_ratio) < GPT_CONFIG_124M["context_length"]:
print("Not enough tokens for the validation loader. "
"Try to lower the `GPT_CONFIG_124M['context_length']` or "
"decrease the `training_ratio`")

print("Train loader:")
for x, y in train_loader:
print(x.shape, y.shape)

print("\nValidation loader:")
for x, y in val_loader:
print(x.shape, y.shape)

train_tokens = 0
for input_batch, target_batch in train_loader:
train_tokens += input_batch.numel()

val_tokens = 0
for input_batch, target_batch in val_loader:
val_tokens += input_batch.numel()

print("Training tokens:", train_tokens)
print("Validation tokens:", val_tokens)
print("All tokens:", train_tokens + val_tokens)

훈련 및 사전 계산을 위한 장치 선택

다음 코드는 사용할 장치를 선택하고 훈련 손실 및 검증 손실을 계산합니다(아직 아무것도 훈련하지 않은 상태에서) 시작점으로 삼습니다.

# Indicate the device to use

if torch.cuda.is_available():
device = torch.device("cuda")
elif torch.backends.mps.is_available():
device = torch.device("mps")
else:
device = torch.device("cpu")

print(f"Using {device} device.")

model.to(device) # no assignment model = model.to(device) necessary for nn.Module classes

# Pre-calculate losses without starting yet
torch.manual_seed(123) # For reproducibility due to the shuffling in the data loader

with torch.no_grad(): # Disable gradient tracking for efficiency because we are not training, yet
train_loss = calc_loss_loader(train_loader, model, device)
val_loss = calc_loss_loader(val_loader, model, device)

print("Training loss:", train_loss)
print("Validation loss:", val_loss)

Training functions

함수 generate_and_print_sample는 컨텍스트를 받아 모델이 그 시점에서 얼마나 좋은지에 대한 느낌을 얻기 위해 일부 토큰을 생성합니다. 이는 train_model_simple에 의해 각 단계에서 호출됩니다.

함수 evaluate_model은 훈련 함수에 지시된 만큼 자주 호출되며, 모델 훈련 시점에서 훈련 손실과 검증 손실을 측정하는 데 사용됩니다.

그런 다음 큰 함수 train_model_simple이 실제로 모델을 훈련합니다. 이 함수는 다음을 기대합니다:

• 훈련 데이터 로더(훈련을 위해 이미 분리되고 준비된 데이터)
• 검증자 로더
• 훈련 중 사용할 최적화기: 이는 그래디언트를 사용하고 손실을 줄이기 위해 매개변수를 업데이트하는 함수입니다. 이 경우, 보시다시피 AdamW가 사용되지만 더 많은 것이 있습니다.
• optimizer.zero_grad()는 각 라운드에서 그래디언트를 재설정하기 위해 호출되어 누적되지 않도록 합니다.
• lr 매개변수는 학습률로, 모델의 매개변수를 업데이트할 때 최적화 과정에서 취하는 단계의 크기를 결정합니다. 작은 학습률은 최적화기가 가중치에 대한 작은 업데이트를 수행하게 하여 더 정확한 수렴을 이끌 수 있지만 훈련 속도를 느리게 할 수 있습니다. 큰 학습률은 훈련 속도를 높일 수 있지만 손실 함수의 최소값을 넘어버릴 위험이 있습니다(손실 함수가 최소화되는 지점을 넘어 점프).
• Weight Decay는 큰 가중치에 대해 패널티를 부여하는 추가 항을 추가하여 손실 계산 단계를 수정합니다. 이는 최적화기가 데이터에 잘 맞추면서 모델을 단순하게 유지하여 머신러닝 모델에서 과적합을 방지하기 위해 어떤 단일 특성에 너무 많은 중요성을 부여하지 않도록 유도합니다.
• L2 정규화가 있는 SGD와 같은 전통적인 최적화기는 손실 함수의 그래디언트와 함께 가중치 감소를 결합합니다. 그러나 AdamW(Adam 최적화기의 변형)는 가중치 감소를 그래디언트 업데이트와 분리하여 더 효과적인 정규화를 이끌어냅니다.
• 훈련에 사용할 장치
• 에포크 수: 훈련 데이터를 반복하는 횟수
• 평가 빈도: evaluate_model을 호출하는 빈도
• 평가 반복: generate_and_print_sample을 호출할 때 모델의 현재 상태를 평가하는 데 사용할 배치 수
• 시작 컨텍스트: generate_and_print_sample을 호출할 때 사용할 시작 문장
• 토크나이저

# Functions to train the data
def train_model_simple(model, train_loader, val_loader, optimizer, device, num_epochs,
eval_freq, eval_iter, start_context, tokenizer):
# Initialize lists to track losses and tokens seen
train_losses, val_losses, track_tokens_seen = [], [], []
tokens_seen, global_step = 0, -1

# Main training loop
for epoch in range(num_epochs):
model.train()  # Set model to training mode

for input_batch, target_batch in train_loader:
optimizer.zero_grad() # Reset loss gradients from previous batch iteration
loss = calc_loss_batch(input_batch, target_batch, model, device)
loss.backward() # Calculate loss gradients
optimizer.step() # Update model weights using loss gradients
tokens_seen += input_batch.numel()
global_step += 1

# Optional evaluation step
if global_step % eval_freq == 0:
train_loss, val_loss = evaluate_model(
model, train_loader, val_loader, device, eval_iter)
train_losses.append(train_loss)
val_losses.append(val_loss)
track_tokens_seen.append(tokens_seen)
print(f"Ep {epoch+1} (Step {global_step:06d}): "
f"Train loss {train_loss:.3f}, Val loss {val_loss:.3f}")

# Print a sample text after each epoch
generate_and_print_sample(
model, tokenizer, device, start_context
)

return train_losses, val_losses, track_tokens_seen


def evaluate_model(model, train_loader, val_loader, device, eval_iter):
model.eval() # Set in eval mode to avoid dropout
with torch.no_grad():
train_loss = calc_loss_loader(train_loader, model, device, num_batches=eval_iter)
val_loss = calc_loss_loader(val_loader, model, device, num_batches=eval_iter)
model.train() # Back to training model applying all the configurations
return train_loss, val_loss


def generate_and_print_sample(model, tokenizer, device, start_context):
model.eval() # Set in eval mode to avoid dropout
context_size = model.pos_emb.weight.shape[0]
encoded = text_to_token_ids(start_context, tokenizer).to(device)
with torch.no_grad():
token_ids = generate_text(
model=model, idx=encoded,
max_new_tokens=50, context_size=context_size
)
decoded_text = token_ids_to_text(token_ids, tokenizer)
print(decoded_text.replace("\n", " "))  # Compact print format
model.train() # Back to training model applying all the configurations

Tip

학습 속도를 개선하기 위해 선형 워밍업 및 코사인 감소라는 몇 가지 관련 기술이 있습니다.

선형 워밍업은 초기 학습 속도와 최대 학습 속도를 정의하고 각 에포크 후에 일관되게 업데이트하는 것입니다. 이는 훈련을 작은 가중치 업데이트로 시작하면 모델이 훈련 단계에서 큰 불안정한 업데이트를 만날 위험이 줄어들기 때문입니다.
코사인 감소는 워밍업 단계 이후에 반 코사인 곡선을 따라 학습 속도를 점진적으로 줄이는 기술로, 가중치 업데이트를 느리게 하여 손실 최소값을 초과할 위험을 최소화하고 후속 단계에서 훈련의 안정성을 보장합니다.

이러한 개선 사항은 이전 코드에 포함되어 있지 않다는 점에 유의하세요.

Start training

import time
start_time = time.time()

torch.manual_seed(123)
model = GPTModel(GPT_CONFIG_124M)
model.to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=0.0004, weight_decay=0.1)

num_epochs = 10
train_losses, val_losses, tokens_seen = train_model_simple(
model, train_loader, val_loader, optimizer, device,
num_epochs=num_epochs, eval_freq=5, eval_iter=5,
start_context="Every effort moves you", tokenizer=tokenizer
)

end_time = time.time()
execution_time_minutes = (end_time - start_time) / 60
print(f"Training completed in {execution_time_minutes:.2f} minutes.")

Print training evolution

다음 함수를 사용하면 모델이 훈련되는 동안의 진화를 출력할 수 있습니다.

import matplotlib.pyplot as plt
from matplotlib.ticker import MaxNLocator
import math
def plot_losses(epochs_seen, tokens_seen, train_losses, val_losses):
fig, ax1 = plt.subplots(figsize=(5, 3))
ax1.plot(epochs_seen, train_losses, label="Training loss")
ax1.plot(
epochs_seen, val_losses, linestyle="-.", label="Validation loss"
)
ax1.set_xlabel("Epochs")
ax1.set_ylabel("Loss")
ax1.legend(loc="upper right")
ax1.xaxis.set_major_locator(MaxNLocator(integer=True))
ax2 = ax1.twiny()
ax2.plot(tokens_seen, train_losses, alpha=0)
ax2.set_xlabel("Tokens seen")
fig.tight_layout()
plt.show()

# Compute perplexity from the loss values
train_ppls = [math.exp(loss) for loss in train_losses]
val_ppls = [math.exp(loss) for loss in val_losses]
# Plot perplexity over tokens seen
plt.figure()
plt.plot(tokens_seen, train_ppls, label='Training Perplexity')
plt.plot(tokens_seen, val_ppls, label='Validation Perplexity')
plt.xlabel('Tokens Seen')
plt.ylabel('Perplexity')
plt.title('Perplexity over Training')
plt.legend()
plt.show()

epochs_tensor = torch.linspace(0, num_epochs, len(train_losses))
plot_losses(epochs_tensor, tokens_seen, train_losses, val_losses)

모델 저장

나중에 훈련을 계속하고 싶다면 모델 + 옵티마이저를 저장할 수 있습니다:

# Save the model and the optimizer for later training
torch.save({
"model_state_dict": model.state_dict(),
"optimizer_state_dict": optimizer.state_dict(),
},
"/tmp/model_and_optimizer.pth"
)
# Note that this model with the optimizer occupied close to 2GB

# Restore model and optimizer for training
checkpoint = torch.load("/tmp/model_and_optimizer.pth", map_location=device)

model = GPTModel(GPT_CONFIG_124M)
model.load_state_dict(checkpoint["model_state_dict"])
optimizer = torch.optim.AdamW(model.parameters(), lr=5e-4, weight_decay=0.1)
optimizer.load_state_dict(checkpoint["optimizer_state_dict"])
model.train(); # Put in training mode

모델만 사용하려는 경우:

# Save the model
torch.save(model.state_dict(), "model.pth")

# Load it
model = GPTModel(GPT_CONFIG_124M)

model.load_state_dict(torch.load("model.pth", map_location=device))

model.eval() # Put in eval mode

GPT2 가중치 로드

로컬에서 GPT2 가중치를 로드하는 두 개의 간단한 스크립트가 있습니다. 두 경우 모두 로컬에서 리포지토리 https://github.com/rasbt/LLMs-from-scratch을 클론할 수 있습니다. 그런 다음:

• 스크립트 https://github.com/rasbt/LLMs-from-scratch/blob/main/ch05/01_main-chapter-code/gpt_generate.py는 모든 가중치를 다운로드하고 OpenAI의 형식을 우리 LLM에서 기대하는 형식으로 변환합니다. 이 스크립트는 필요한 구성과 프롬프트 “Every effort moves you“로 준비되어 있습니다.
• 스크립트 https://github.com/rasbt/LLMs-from-scratch/blob/main/ch05/02_alternative_weight_loading/weight-loading-hf-transformers.ipynb는 로컬에서 GPT2 가중치를 로드할 수 있게 해줍니다(단지 CHOOSE_MODEL 변수를 변경하면 됩니다) 그리고 몇 가지 프롬프트에서 텍스트를 예측합니다.

참고문헌

• https://www.manning.com/books/build-a-large-language-model-from-scratch

Tip

AWS 해킹 배우기 및 연습하기:HackTricks Training AWS Red Team Expert (ARTE)
GCP 해킹 배우기 및 연습하기: HackTricks Training GCP Red Team Expert (GRTE) Azure 해킹 배우기 및 연습하기: HackTricks Training Azure Red Team Expert (AzRTE)

HackTricks 지원하기

• 구독 계획 확인하기!
• **💬 디스코드 그룹 또는 텔레그램 그룹에 참여하거나 트위터 🐦 @hacktricks_live를 팔로우하세요.
• HackTricks 및 HackTricks Cloud 깃허브 리포지토리에 PR을 제출하여 해킹 트릭을 공유하세요.