프로젝트 1 — 생성한 데이터로 상담 모델 미세조정

import json, re, time, urllib.request
import numpy as np
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, Trainer, TrainingArguments, set_seed
from peft import LoraConfig, get_peft_model
from datasets import Dataset
from sentence_transformers import SentenceTransformer

# Ollama 가 GPU 에 올려 둔 모델을 내린다. 학습에 GPU 를 쓰려고
for m in ("qwen2.5:3b", "qwen2.5vl:3b"):
    urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", headers={"Content-Type": "application/json"},
                                                  data=json.dumps({"model": m, "keep_alive": 0}).encode())).read()

train_rows = [json.loads(line) for line in open("out_sft_train.jsonl", encoding="utf-8")]
test = [json.loads(line) for line in open("care_test.jsonl", encoding="utf-8")]
text = open("care_guide.md", encoding="utf-8").read()
sections = [(m.group(1), m.group(2).strip()) for m in re.finditer(r"## (.+)\n\n(.+)", text)]
SYSTEM = train_rows[0]["messages"][0]["content"]
norm = lambda s: s.replace(" ", "").replace(",", "")
print("학습", len(train_rows), "쌍 · 시험", len(test), "개 · GPU:", torch.cuda.is_available())

name = "Qwen/Qwen2.5-0.5B-Instruct"
tok = AutoTokenizer.from_pretrained(name)
device = "cuda" if torch.cuda.is_available() else "cpu"
set_seed(42)
model = AutoModelForCausalLM.from_pretrained(name, dtype=torch.float16 if device == "cuda" else torch.float32).to(device)


def ask(m, question, context=None):
    """질문(과 자료)을 채팅 템플릿으로 묻고 답을 받는다. 탐욕적 선택이라 다시 돌려도 같다"""
    user = f"[자료]\n{context}\n\n[질문]\n{question}" if context else question
    msgs = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": user}]
    enc = tok.apply_chat_template(msgs, 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().replace("\n", " ")


# RAG 의 절 찾기: e5 로 질문과 가장 가까운 절 하나
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
S = emb.encode([f"passage: {t}: {b}" for t, b in sections], normalize_embeddings=True)
results = {"미세조정 전": [], "RAG(절 하나)": [], "미세조정 뒤": []}
answers = {}
for t in test:
    best = sections[int(np.argmax(S @ emb.encode("query: " + t["질문"], normalize_embeddings=True)))]
    for setting, a in [("미세조정 전", ask(model, t["질문"])), ("RAG(절 하나)", ask(model, t["질문"], f"{best[0]}: {best[1]}"))]:
        results[setting].append(norm(t["열쇠"]) in norm(a))
        answers[(setting, t["질문"])] = a
print({k: f"{sum(v)}/{len(test)}" for k, v in results.items() if v})
for t in test[:3]:
    print("Q:", t["질문"], f"(열쇠 {t['열쇠']})")
    print("   전:", answers[("미세조정 전", t["질문"])][:70])
    print("   RAG:", answers[("RAG(절 하나)", t["질문"])][:70])

def encode(batch):
    """질문 부분은 -100(채점 안 함), 답 부분만 채점하게 labels 를 만든다(5차시와 같다)"""
    input_ids, labels = [], []
    for msgs in batch["messages"]:
        p = tok(tok.apply_chat_template(msgs[:2], add_generation_prompt=True, tokenize=False), add_special_tokens=False)["input_ids"]
        f = tok(tok.apply_chat_template(msgs, 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):
    """가장 긴 줄에 맞춰 채우고, 채운 자리는 보지도 채점하지도 않는다"""
    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}


data = Dataset.from_list([{"messages": r["messages"]} for r in train_rows]).map(encode, batched=True, remove_columns=["messages"])
first = data[0]
print("토큰 수:", len(first["input_ids"]), "· 채점하는 부분:", repr(tok.decode([t for t in first["labels"] if t != -100])))

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_care_ckpt", per_device_train_batch_size=8, num_train_epochs=8, learning_rate=5e-4,
                         logging_steps=5, save_strategy="no", report_to="none", seed=42,
                         fp16=torch.cuda.is_available(), disable_tqdm=True)
trainer = Trainer(model=model, args=args, train_dataset=data, data_collator=collate)
start = time.perf_counter()
trainer.train()
losses = [h["loss"] for h in trainer.state.log_history if "loss" in h]
print(f"학습 시간: {time.perf_counter() - start:.0f}초 · 손실: {losses[0]:.3f} → {losses[-1]:.3f}")
model = trainer.model.eval()
model.save_pretrained("out_lora_care")                    # 어댑터만 저장한다

for t in test:
    a = ask(model, t["질문"])
    results["미세조정 뒤"].append(norm(t["열쇠"]) in norm(a))
    answers[("미세조정 뒤", t["질문"])] = a

import pandas as pd
table = pd.DataFrame({"열쇠": [t["열쇠"] for t in test],
                      **{k: ["O" if x else "X" for x in v] for k, v in results.items()}})
table.index = [t["질문"][:18] for t in test]
print(table.to_string())
print({k: f"{sum(v)}/{len(test)}" for k, v in results.items()})
for t, ok in zip(test, results["미세조정 뒤"]):
    if not ok:
        print("뒤에도 틀림:", t["질문"], "→", answers[("미세조정 뒤", t["질문"])][:80])