실습 3 — LoRA 로 SFT 하기

import json, time
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, Trainer, TrainingArguments, set_seed
from peft import LoraConfig, get_peft_model
from datasets import Dataset

name = "Qwen/Qwen2.5-0.5B-Instruct"
tok = AutoTokenizer.from_pretrained(name)
SYSTEM = "당신은 온라인 쇼핑몰 고객센터 상담원입니다."
with open("sft_shop.jsonl", encoding="utf-8") as f:
    rows = [json.loads(line) for line in f]
train_rows = [r for r in rows if r["나눔"] == "train"]
test_rows = [r for r in rows if r["나눔"] == "test"]

def messages(q, a=None):
    m = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": q}]
    return m + ([{"role": "assistant", "content": a}] if a is not None else [])

def encode(batch):
    input_ids, labels = [], []
    for q, a in zip(batch["질문"], batch["답변"]):
        p = tok(tok.apply_chat_template(messages(q), add_generation_prompt=True, tokenize=False), add_special_tokens=False)["input_ids"]
        f = tok(tok.apply_chat_template(messages(q, a), tokenize=False), add_special_tokens=False)["input_ids"]
        input_ids.append(f)
        labels.append([-100] * len(p) + f[len(p):])
    return {"input_ids": input_ids, "labels": labels}

def collate(features):
    """4장과 같다: 가장 긴 줄에 맞춰 채우고, 채운 자리는 보지도 채점하지도 않는다."""
    L = max(len(x["input_ids"]) for x in features)
    ids = torch.full((len(features), L), tok.pad_token_id)
    lab = torch.full((len(features), L), -100)
    att = torch.zeros((len(features), L), dtype=torch.long)
    for i, x in enumerate(features):
        n = len(x["input_ids"])
        ids[i, :n], lab[i, :n], att[i, :n] = torch.tensor(x["input_ids"]), torch.tensor(x["labels"]), 1
    return {"input_ids": ids, "labels": lab, "attention_mask": att}

sft_data = Dataset.from_list(train_rows).map(encode, batched=True, remove_columns=list(train_rows[0].keys()))
print("학습 쌍:", len(sft_data), "/ GPU:", torch.cuda.is_available())

set_seed(42)
model = AutoModelForCausalLM.from_pretrained(name, dtype=torch.float16)
model = get_peft_model(model, LoraConfig(r=8, lora_alpha=16, lora_dropout=0.05,
                       target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], task_type="CAUSAL_LM"))
args = TrainingArguments(
    "out_lora_ckpt",
    per_device_train_batch_size=8,
    num_train_epochs=8,                       # 72쌍을 여덟 번 돈다
    learning_rate=5e-4,                       # LoRA 는 학습률을 크게 잡는다
    logging_steps=9,
    save_strategy="no",
    report_to="none",
    seed=42,
    fp16=torch.cuda.is_available(),           # 혼합 정밀도
    disable_tqdm=True,
)
trainer = Trainer(model=model, args=args, train_dataset=sft_data, data_collator=collate)
if torch.cuda.is_available():
    torch.cuda.reset_peak_memory_stats()      # 최대 메모리 기록을 0 으로
start = time.perf_counter()
trainer.train()
print(f"학습 시간: {time.perf_counter() - start:.0f}초")
if torch.cuda.is_available():
    print(f"최대 GPU 메모리: {torch.cuda.max_memory_allocated() / 1024**2:,.0f} MB")
# 학습 기록에서 손실만 골라 처음과 끝을 본다
losses = [h["loss"] for h in trainer.state.log_history if "loss" in h]
print("손실:", round(losses[0], 3), "→", round(losses[-1], 3))

model = trainer.model.eval()

def ask(m, q):
    enc = tok.apply_chat_template(messages(q), add_generation_prompt=True, return_tensors="pt", return_dict=True).to(m.device)
    with torch.no_grad():
        out = m.generate(**enc, max_new_tokens=60, do_sample=False)
    return tok.decode(out[0][enc["input_ids"].shape[1]:], skip_special_tokens=True).strip()

hits = 0
for r in test_rows:
    a = ask(model, r["질문"])
    ok = r["정답열쇠"] in a
    hits += ok
    print("O" if ok else "X", r["질문"], "→", a.replace("\n", " "))
print(f"맞힌 수: {hits} / {len(test_rows)}")

model.save_pretrained("out_lora_shop")       # 어댑터만 저장된다
import os
size = sum(os.path.getsize(os.path.join("out_lora_shop", f)) for f in os.listdir("out_lora_shop"))
print("저장한 파일:", sorted(os.listdir("out_lora_shop")), f"/ {size / 1024**2:.1f} MB")