RAM_Altman_GPT / src /train.py
brunuu's picture
1.0
7de15fc verified
Raw History Blame Contribute Delete
3.17 kB
import math
import time
import torch
import torch.nn.functional as F
from transformers import GPT2LMHeadModel, GPT2TokenizerFast
from config import BATCH_SIZE, LEARNING_RATE, STEPS
REPLY_PREFIX = "RAM-ALTMAN:"
HOOK_TOKENS = 8
HOOK_WEIGHT = 4.0
LOG_EVERY = 250
def read_segments(data_file: str) -> tuple[list[str], list[bool]]:
texts, is_reply = [], []
with open(data_file, encoding="utf-8") as f:
for line in f:
if line.startswith(REPLY_PREFIX):
texts += [REPLY_PREFIX, line.removeprefix(REPLY_PREFIX)]
is_reply += [False, True]
else:
texts.append(line)
is_reply.append(False)
return texts, is_reply
def reply_weights(length: int) -> list[float]:
# The hook weighs more than the rant
if length <= HOOK_TOKENS:
return [1.0] * length
return [HOOK_WEIGHT] * HOOK_TOKENS + [1.0] * (length - HOOK_TOKENS)
def load_token_ids(hf_tokenizer: GPT2TokenizerFast, data_file: str, device: str) -> tuple[torch.Tensor, torch.Tensor]:
texts, is_reply = read_segments(data_file)
ids, weights = [], []
for segment_ids, reply in zip(hf_tokenizer(texts)["input_ids"], is_reply):
ids += segment_ids
weights += reply_weights(len(segment_ids)) if reply else [0.0] * len(segment_ids)
return torch.tensor(ids, device=device), torch.tensor(weights, device=device)
def cosine_lr(step: int) -> float:
return LEARNING_RATE * 0.5 * (1 + math.cos(math.pi * step / STEPS))
def random_batch(token_ids: torch.Tensor, weights: torch.Tensor, context_length: int) -> tuple[torch.Tensor, torch.Tensor]:
starts = torch.randint(0, len(token_ids) - context_length - 1, (BATCH_SIZE,))
inputs = torch.stack([token_ids[i:i + context_length] for i in starts])
input_weights = torch.stack([weights[i:i + context_length] for i in starts])
return inputs, input_weights
def weighted_loss(logits: torch.Tensor, inputs: torch.Tensor, input_weights: torch.Tensor) -> torch.Tensor:
predictions = logits[:, :-1].reshape(-1, logits.size(-1))
targets = inputs[:, 1:].reshape(-1)
target_weights = input_weights[:, 1:].reshape(-1)
token_losses = F.cross_entropy(predictions, targets, reduction="none")
return (token_losses * target_weights).sum() / target_weights.sum().clamp_min(1e-8)
def train_loop(model: GPT2LMHeadModel, token_ids: torch.Tensor, weights: torch.Tensor, context_length: int) -> None:
optimizer = torch.optim.AdamW(model.parameters(), lr=LEARNING_RATE, weight_decay=0.01)
start = time.time()
for step in range(1, STEPS + 1):
for group in optimizer.param_groups:
group["lr"] = cosine_lr(step)
inputs, input_weights = random_batch(token_ids, weights, context_length)
loss = weighted_loss(model(input_ids=inputs).logits, inputs, input_weights)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
if step == 1 or step % LOG_EVERY == 0:
print(f"step {step}/{STEPS} - loss {loss.item():.3f} <> {time.time() - start:.0f}s", flush=True)