LLM 만들기 Pretraining

LLM을 만들기 위해서는 방대한 train data가 필요하다.

data가 준비되면 전처리과정부터 LLM모델 학습하고 모델을 완성한다.

 

전처리과정

전처리 과정은 모델이 텍스트의 노이즈에 방해받지 않고 패턴을 명확히 학습하게 도와줍니다.

데이터의 질(Quality)이 양보다 중요합니다.

 

Cleaning

• 불필요한 공백 제거: 연속된 공백, 탭(\t), 줄바꿈(\n)이 너무 많으면 모델이 혼란을 겪을 수 있으므로 하나로 통일하거나 제거합니다.
• 특수문자 처리: 학습 목적에 따라 이메일 주소, URL, HTML 태그(<div>, <a> 등)를 정규표현식으로 제거합니다.
• 오탈자 및 노이즈 제거: 같이 한글의 경우, 깨진 글자나 자음/모음만 있는 경우(예: 'ㅋㅋㅋ', 'ㅠㅠ')를 필터링합니다.

    text = re.sub(r'<[^>]*>', '', text) # 태그 제거
    text = re.sub(r'http\S+', '', text) # URL 제거
    text = re.sub(r'[^a-zA-Z0-9가-힣ㄱ-ㅎㅏ-ㅣ\s.,!?\'\"]', '', text) #불필요한 특수문자 제거
    text = re.sub(r'\s+', ' ', text).strip() #연속된 공백 및 줄바꿈 정리

 

정규화 (Normalization)

• 대소문자 통합: 영문 데이터의 경우 대문자를 소문자로 통일하기도 하지만, 최근 대형 언어 모델(GPT 등)은 대소문자를 구분하여 학습하는 추세입니다.
• 유니코드 정규화: 한글은 '가'가 가(하나의 코드) 또는 ㄱ+ㅏ(조합형)로 저장될 수 있습니다. 이를 하나로 통일(NFC 정규화)해야 같은 단어로 인식합니다.

text = unicodedata.normalize('NFC', text)

 

문장 경계 식별 및 특수 토큰 추가

• 문서 구분: 여러 문서를 하나의 파일로 합칠 때, 각 문서 사이에 <|endoftext|>와 같은 EOS(End Of Sentence/Text) 토큰을 삽입하여 모델이 "여기서 이야기가 끝났다"는 것을 배우게 합니다.
• 문장 분리: 너무 긴 텍스트는 max_length에 맞게 적절히 자르거나, 문장 단위로 끊어서 입력 데이터를 구성합니다.

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

토큰화 (Tokenization)

컴퓨터는 텍스트(글자)를 직접 이해할 수 없습니다. 따라서 tokenizer.encode()를 통해 단어 혹은 단어의 일부를 고유한 숫자(ID)로 변환합니다.

• BPE(Byte Pair Encoding): 최근 GPT 계열에서 가장 많이 쓰이는 방식입니다. 자주 등장하는 단어 뭉치를 하나의 토큰으로 묶어 효율성을 높입니다.
• Vocabulary 제한: 너무 드물게 나타나는 단어는 UNK(Unknown) 토큰으로 처리하거나 제거하여 사전(Vocab) 크기를 관리합니다.

tokenizer = tiktoken.get_encoding("gpt2") #GPT-2, GPT-3.5/4 기준 BPE 알고리즘 사용

 

 

데이터 로더 정의

텍스트 데이터를 학습용 데이터셋으로 변환하여 PyTorch의 DataLoader를 생성하는 과정

 

슬라이딩 윈도우 (Sliding Window)

• max_length=32: 모델이 한 번에 볼 수 있는 단어(토큰)의 개수입니다.
• stride=4: 데이터를 추출할 때 4칸씩 이동하며 샘플을 만듭니다.
  • 값이 작을수록 데이터 양이 많아지지만, 중복된 데이터가 생성되어 학습 시간이 늘어납니다.
  • 값이 크면 데이터 양은 적어지지만, 중복이 적은 다양한 샘플을 학습할 수 있습니다.

입력(Input)과 타겟(Target) 구성

• Next Token Prediction: 언어 모델의 핵심 원리입니다.
  • input_chunk: [A, B, C, D]
  • target_chunk: [B, C, D, E]
  • 모델은 A를 보고 B를 예측하고, A, B를 보고 C를 예측하도록 학습됩니다.
  • 한 단어씩 만들어 가는 자동회귀(autoregressive) LLM

DataLoader

• Batching: 128개의 데이터를 하나로 묶어 GPU가 한 번에 계산하게 합니다. 이는 학습 속도를 비약적으로 높여줍니다.
• Shuffle: 데이터를 매 학습(Epoch)마다 무작위로 섞어 모델이 데이터의 순서를 외우지 못하게 방지합니다.

import torch
from torch.utils.data import Dataset, DataLoader
import tiktoken  # 토큰화를 위한 라이브러리

class MyDataset(Dataset):
    def __init__(self, txt, max_length, stride):
        self.input_ids = []
        self.target_ids = []

        # 1. 토크나이저 초기화 (GPT-2, GPT-3.5/4 기준 BPE 알고리즘 사용)
        tokenizer = tiktoken.get_encoding("gpt2")

        # 2. 토큰화 (전체 텍스트를 정수 ID 리스트로 변환)
        # <|endoftext|> 특수 토큰을 추가하여 문서의 시작을 알림
        token_ids = tokenizer.encode("<|endoftext|>" + txt, allowed_special={"<|endoftext|>"})

        print(f"# of tokens in txt: {len(token_ids)}")

        # 3. 슬라이딩 윈도우 방식으로 데이터 생성
        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]

# 4. 파일 읽기 및 데이터 로더 생성
try:
    with open("Alice.txt", 'r', encoding='utf-8-sig') as file:
        txt = file.read()

    dataset = MyDataset(txt, max_length=32, stride=4)
    train_loader = DataLoader(dataset, batch_size=128, shuffle=True, drop_last=True)
    
    print(f"Total batches: {len(train_loader)}")

except FileNotFoundError:
    print("파일을 찾을 수 없습니다. 파일명을 확인해주세요.")

 

모델 정의

GPT(Generative Pre-trained Transformer) 아키텍처의 핵심 구성 요소

 

MultiHeadAttention

• 인과적 마스킹(Causal Masking): register_buffer로 등록된 mask를 사용하여 모델이 미래의 토큰을 보지 못하게 차단합니다. 이는 문장 생성 모델의 핵심입니다.
• 병렬 처리: 입력을 여러 개의 '헤드'로 나누어 다양한 문맥적 관계를 동시에 학습합니다.

LayerNorm & TransformerBlock

• Pre-LayerNorm 구조: norm을 어텐션과 피드포워드 이전에 적용하는 방식으로, 최신 트랜스포머 모델들이 학습 안정성을 위해 채택하는 표준 방식입니다.
• 잔차 연결(Residual Connection): x + shortcut을 통해 기울기 소실 문제를 방지하고 깊은 층에서도 학습이 잘 되도록 합니다.

GPT Model

• 토큰 및 위치 임베딩: 단어의 의미(tok_emb)와 문장 내 순서 정보(pos_emb)를 더해 입력값으로 사용합니다.
• 최종 출력층: out_head를 통해 다음에 올 토큰의 확률(Logits)을 사전 크기(VOCAB_SIZE)만큼 출력합니다.

import torch
import torch.nn as nn
import torch.nn.functional as F

# 2.0+ 버전 최적화: Flash Attention 활용을 위해 MultiHeadAttention 수정
class MultiHeadAttention(nn.Module):
    def __init__(self, d_in, d_out, num_heads, context_length, drop_rate, qkv_bias):
        super().__init__()
        self.num_heads = num_heads
        self.head_dim = d_out // num_heads
        
        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)
        self.dropout = drop_rate

    def forward(self, x):
        b, n, d = x.shape
        # Q, K, V 생성 및 (b, heads, n, head_dim)으로 변환
        q = self.W_query(x).view(b, n, self.num_heads, self.head_dim).transpose(1, 2)
        k = self.W_key(x).view(b, n, self.num_heads, self.head_dim).transpose(1, 2)
        v = self.W_value(x).view(b, n, self.num_heads, self.head_dim).transpose(1, 2)

        # 팁 1: F.scaled_dot_product_attention 사용 (Flash Attention 자동 적용)
        # 메모리 효율과 속도가 비약적으로 향상되며 마스킹도 내부에서 처리 가능
        context_vec = F.scaled_dot_product_attention(
            q, k, v, is_causal=True, dropout_p=self.dropout if self.training else 0.0
        )
        
        context_vec = context_vec.transpose(1, 2).contiguous().view(b, n, -1)
        return self.out_proj(context_vec)

# 팁 2: 가중치 초기화 함수 정의 (GPT-2 스타일)
def _init_weights(module):
    if isinstance(module, (nn.Linear, nn.Embedding)):
        torch.nn.init.normal_(module.weight, mean=0.0, std=0.02)
        if isinstance(module, nn.Linear) and module.bias is not None:
            torch.nn.init.zeros_(module.bias)

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['num_layers'])]
        )

        self.final_norm = LayerNorm(cfg['emb_dim'])
        self.out_head = nn.Linear(cfg['emb_dim'], cfg['vocab_size'], bias=False)
        
        # 가중치 초기화 적용
        self.apply(_init_weights)

    def forward(self, in_idx):
        b, n = in_idx.shape
        device = in_idx.device
        x = self.tok_emb(in_idx) + self.pos_emb(torch.arange(n, device=device))
        x = self.drop_emb(x)
        x = self.trf_blocks(x)
        logits = self.out_head(self.final_norm(x))
        return logits

 

모델학습

학습 속도를 높여주고, 학습 후반부의 수렴 안정성을 크게 향상시키기위해 Mixed Precision(혼합 정밀도), Gradient Clipping(기울기 클리핑), Learning Rate Scheduler(학습률 스케줄러)를 적용

 

torch.amp.GradScaler (Mixed Precision)

• 연산을 16비트와 32비트를 혼합하여 수행합니다. GPU 메모리 사용량을 줄이고 연산 속도를 대폭 향상시킵니다. PyTorch AMP 가이드에서 상세 내용을 확인할 수 있습니다.

clip_grad_norm_ (Gradient Clipping)

• LLM은 학습 중 특정 구간에서 기울기가 갑자기 튀는 현상이 잦습니다. 이를 강제로 일정 수치(max_norm=1.0) 이하로 제한하여 모델이 망가지는 것을 방지합니다.

CosineAnnealingLR 스케줄러

• 학습 초기에는 큰 학습률로 빠르게 배우고, 끝날수록 학습률을 줄여 손실 함수(Loss)의 최저점에 안정적으로 도달하게 합니다.

torch.compile

• PyTorch 2.0+의 핵심 기능으로, 모델의 연산 그래프를 분석해 GPU 커널 수준에서 최적화합니다.

출력 및 저장 최적화

• zfill(3)을 사용하여 파일명이 model_001.pth, model_010.pth 순으로 정렬되기 쉽게 만들었습니다.

import torch
import torch.nn.functional as F

# 1. 환경 설정 및 최적화 준비
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")

torch.manual_seed(123)
model = GPTModel() # 이전에 정의한 모델
model.to(device)

# 컴파일
if torch.__version__ >= "2.0.0" and device.type == "cuda":
    model = torch.compile(model)

optimizer = torch.optim.AdamW(model.parameters(), lr=0.0004, weight_decay=0.1)

# 팁 1: 학습률 스케줄러 추가 (학습 후반부에 lr을 점진적으로 줄여 정교하게 학습)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)

# 팁 2: Mixed Precision 가속기 (NVIDIA GPU Tensor Core 활용)
scaler = torch.amp.GradScaler('cuda') if device.type == "cuda" else None

tokens_seen, global_step = 0, -1
losses = []

for epoch in range(100):
    model.train()
    epoch_loss = 0
    
    for input_batch, target_batch in train_loader:
        input_batch, target_batch = input_batch.to(device), target_batch.to(device)
        optimizer.zero_grad()

        # 팁 3: 혼합 정밀도 컨텍스트 (FP16/BF16 연산으로 속도 향상)
        if scaler:
            with torch.amp.autocast('cuda'):
                logits = model(input_batch)
                loss = F.cross_entropy(logits.flatten(0, 1), target_batch.flatten())
            
            scaler.scale(loss).backward()
            
            # 팁 4: Gradient Clipping (기울기 폭주 방지, GPT 학습 필수)
            scaler.unscale_(optimizer)
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            
            scaler.step(optimizer)
            scaler.update()
        else:
            # CPU 등 가속기가 없는 경우 일반 학습
            logits = model(input_batch)
            loss = F.cross_entropy(logits.flatten(0, 1), target_batch.flatten())
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            optimizer.step()

        epoch_loss += loss.item()
        tokens_seen += input_batch.numel()
        global_step += 1

        if global_step % 100 == 0: # 출력 주기를 좀 더 촘촘하게 변경
            print(f"Step {global_step} | Tokens: {tokens_seen} | Loss: {loss.item():.4f}")

    # 에폭 종료 후 처리
    avg_loss = epoch_loss / len(train_loader)
    losses.append(avg_loss)
    scheduler.step() # 스케줄러 업데이트
    
    current_lr = optimizer.param_groups[0]['lr']
    print(f"Epoch: {epoch + 1}/100 | Avg Loss: {avg_loss:.4f} | LR: {current_lr:.6f}")
    
    # 모델 저장 (State Dict만 저장하는 권장 방식)
    torch.save(model.state_dict(), f"model_{str(epoch + 1).zfill(3)}.pth")

 

 

최종 Loss 확인

 

모델구성 중 주요 개념

 

활성화함수 GELU

Transformer

 

Masked Attention