Iโm trying to fine-tune using things like the Harry Potter novels.
I performed fine tuning several times with llama-3-Korean-Bllossom-8B.
I wanted to create a creative llama model rather than an instructional one.
Let me show you an example.
I made one sample from Chapter 1: The Surviving Child to Chapter 2: In Front of the Vanishing Window.
The results of doing as gpt said were disastrous.
In order to divide the scenes into chunks, there were times when scenes were cut randomly and the scenes were messed around, saying they had to be stitched together.
So I cut one sample like that.
The jsonl file contains {โtextโ: โChapter 1 The Child Who Survived\nThe Dursleys living at 4 Privet Drive were people who were very proud of being normal.
Everything.โ}
I created a sample like this.
Here is the code:
-- coding: utf-8 --
โโ"
train_auto.py
QLoRA continuation/style learning automation script
Goal:
- train.jsonl {โtextโ: โโฆโ} Automatic data analysis
- Automatic learning amount determination based on data size/token/block
- Prevents the problem of small data lasting dozens of epochs
- Test creation of the same prompt before and after learning
- Automatically save before_sample.txt / after_sample.txt / train_report.txt
- Automatically generate verification prompts for short/long posts
Note:
- This script is for learning โwriting style/sequence.โ
- Not for learning chat/RP format.
โโ"
import gc
import json
import math
import random
import inspection
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import torch
from datasets import Dataset
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from transformers import (
AutoModelForCausalLM;
Auto Tokenizer,
BitsAndBytesConfig,
DataCollatorForLanguageModeling,
EarlyStoppingCallback,
GenerationConfig,
Trainer,
TrainerCallback,
TrainingArguments,
)
===========================================================
Default settings
===========================================================
MODEL_NAME = โLlama-3.1-8Bโ
DATA_PATH = โtrain.jsonlโ
OUTPUT_DIR = โblossom-8b-qlora-story-autoโ
MAX_SEQ_LENGTH = 1536
VALID_RATIO = 0.05
RANDOM_SEED = 42
MIN_CHARS = 40
Auto tuning on/off
AUTO_TUNE = True
Verification of generation before/after learning on/off
AUTO_GENERATION_TEST = True
GENERATION_TEST_DIR = โgeneration_checkโ
GENERATION_MAX_NEW_TOKENS = 120
If you want to specify your own verification prompt, put it here.
If left blank, it is automatically extracted from train.jsonl.
GENERATION_TEST_PROMPTS: List[str] =
Saving/evaluating too small data only takes time, so it is automatically turned off.
SAVE_TOTAL_LIMIT = 2
USE_BF16 = torch.cuda.is_available() and torch.cuda.is_bf16_supported()
===========================================================
data structure
===========================================================
@dataclass
classDataStats:
file_size_bytes: int
file_size_label: str
samples: int
total_chars: int
total_tokens: int
avg_chars: float
avg_tokens: float
min_chars: int
max_chars: int
min_tokens: int
max_tokens: int
estimated_blocks: int
skipped_short: int
skipped_duplicate: int
skipped_json: int
@dataclass
class TrainPlan:
profile: str
mode: str
train_blocks: int
eval_blocks: int
max_steps: int
num_train_epochs: int
gradient_accumulation_steps: int
optimize_updates: int
learning_rate: float
warmup_ratio: float
eval_strategy: str
save_strategy: str
eval_steps: Optional[int]
save_steps: Optional[int]
load_best_model_at_end: bool
early_stopping: bool
@dataclass
class PromptItem:
prompt: str
position: str
source_chars: int
line_no: int
===========================================================
utility
===========================================================
def set_seed(seed: int) โ None:
random.seed(seed)
torch.manual_seed(seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(seed)
def file_size_label(path: str) โ Tuple[int, str]:
size = Path(path).stat().st_size
if size < 1024:
return size, f"{size} B"
if size < 1024 * 1024:
return size, f"{size / 1024:.1f} KB"
return size, f"{size / (1024 * 1024):.2f} MB"
def ceil_div(a: int, b: int) โ int:
return (a + b - 1) // b
def normalize_text(text: str) โ str:
return " โ.join(str(text).replace(โ\r\n", โ\nโ).replace(โ\rโ, โ\nโ).split()).strip()
===========================================================
Data load/analysis
===========================================================
def load_jsonl(path: str, tokenizer: AutoTokenizer) โ Tuple[Dataset, DataStats, List[Dict[str, Any]]]:
rows: List[Dict[str, Any]] =
raw_items: List[Dict[str, Any]] =
seen = set()
skipped_short = 0
skipped_duplicate = 0
skipped_json = 0
total_chars = 0
total_tokens = 0
min_chars: Optional[int] = None
max_chars = 0
min_tokens: Optional[int] = None
max_tokens = 0
with open(path, โrโ, encoding=โutf-8-sigโ) as f:
for line_no, line in enumerate(f, start=1):
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
skipped_json += 1
print(f"[skip] JSON parsing failed: line {line_no}")
continue
text = str(obj.get(โtextโ, โโ)).replace(โ\r\nโ, โ\nโ).replace(โ\rโ, โ\nโ).strip()
text = โ\nโ.join(x.rstrip() for x in text.splitlines()).strip()
if len(text) < MIN_CHARS:
skipped_short += 1
continue
if text in seen:
skipped_duplicate += 1
continue
token_count = len(tokenizer.encode(text, add_special_tokens=False))
char_count = len(text)
seen.add(text)
rows.append({"text": text, "chars": char_count, "tokens": token_count})
raw_items.append({"text": text, "chars": char_count, "tokens": token_count, "line_no": line_no})
total_chars += char_count
total_tokens += token_count
min_chars = char_count if min_chars is None else min(min_chars, char_count)
max_chars = max(max_chars, char_count)
min_tokens = token_count if min_tokens is None else min(min_tokens, token_count)
max_tokens = max(max_tokens, token_count)
if not rows:
raise ValueError(f"ํ์ต ๊ฐ๋ฅํ text๊ฐ ์์ต๋๋ค: {path}")
size_bytes, size_label = file_size_label(path)
stats = DataStats(
file_size_bytes=size_bytes,
file_size_label=size_label,
samples=len(rows),
total_chars=total_chars,
total_tokens=total_tokens,
avg_chars=total_chars / len(rows),
avg_tokens=total_tokens / len(rows),
min_chars=min_chars or 0,
max_chars=max_chars,
min_tokens=min_tokens or 0,
max_tokens=max_tokens,
estimated_blocks=max(1, ceil_div(total_tokens, MAX_SEQ_LENGTH)),
skipped_short=skipped_short,
skipped_duplicate=skipped_duplicate,
skipped_json=skipped_json,
)
print_data_stats(stats)
return Dataset.from_list(rows), stats, raw_items
def print_data_stats(stats: DataStats) โ None:
print(โ=โ * 60)
print(โ[data stats]โ)
print(f"ํ์ผ ํฌ๊ธฐ : {stats.file_size_label}โ)
print(f"์ํ ์ : {stats.samples:,}โ)
print(f"์ด ๋ฌธ์ ์ : {stats.total_chars:,}โ)
print(f"์ด ํ ํฐ ์ : {stats.total_tokens:,}โ)
print(f"ํ๊ท ๋ฌธ์ ์ : {stats.avg_chars:,.1f}โ)
print(f"ํ๊ท ํ ํฐ ์ : {stats.avg_tokens:,.1f}โ)
print(f"์ต์/์ต๋ ๋ฌธ์ ์ : {stats.min_chars:,} / {stats.max_chars:,}โ)
print(f"์ต์/์ต๋ ํ ํฐ ์ : {stats.min_tokens:,} / {stats.max_tokens:,}โ)
print(f"์์ ๋ธ๋ก : {stats.estimated_blocks:,}๊ฐ @ MAX_SEQ_LENGTH={MAX_SEQ_LENGTH}โ)
print(f"์ ์ธ : ์งง์={stats.skipped_short:,}, ์ค๋ณต={stats.skipped_duplicate:,}, JSON์ค๋ฅ={stats.skipped_json:,}โ)
print(โ=โ * 60)
============================================================
ํ ํฐํ/ํจํน
============================================================
def tokenize_for_continuation(examples: Dict[str, List[str]], tokenizer: AutoTokenizer) โ Dict[str, Any]:
texts = [text + tokenizer.eos_token for text in examples[โtextโ]]
return tokenizer(texts, add_special_tokens=False)
def pack_blocks(examples: Dict[str, List[List[int]]]) โ Dict[str, List[List[int]]]:
โโ"
MAX_SEQ_LENGTH ๋จ์ packing.
๋ง์ง๋ง remainder๋ฅผ ๋ฒ๋ฆฌ์ง ์๋๋ค.
KB๊ธ ๋ฐ์ดํฐ๊ฐ 1536ํ ํฐ ๋ฏธ๋ง์ด๋ผ๋ ์ด์ ๋ก 0๋ธ๋ก์ด ๋๋ ๋ฌธ์ ๋ฅผ ๋ง๋๋ค.
โโ"
result: Dict[str, List[List[int]]] = {}
for key in ("input_ids", "attention_mask"):
concatenated: List[int] = []
for item in examples[key]:
concatenated.extend(item)
if not concatenated:
result[key] = []
continue
result[key] = [
concatenated[i : i + MAX_SEQ_LENGTH]
for i in range(0, len(concatenated), MAX_SEQ_LENGTH)
]
return result
============================================================
์๋ ํ์ต ๊ณํ
============================================================
def choose_profile(stats: DataStats, train_blocks: int) โ Tuple[str, int, int, float, float, int]:
โโ"
๋ฐํ:
profile, target_optimizer_updates, grad_accum, learning_rate, warmup_ratio, save_eval_steps
๊ธฐ์ค:
- ํ์ผ ํฌ๊ธฐ ํ๋๋ง ๋ณด์ง ์๋๋ค.
- ์ด ํ ํฐ ์์ packed block ์๋ฅผ ๊ฐ์ด ๋ณธ๋ค.
- KB๊ธ ํ์ผ์ด block ์ ๋๋ฌธ์ large๋ก ํ์ง ์๊ฒ ๋ฐฉ์งํ๋ค.
"""
size_mb = stats.file_size_bytes / (1024 * 1024)
tokens = stats.total_tokens
blocks = train_blocks
rules = [
# profile, max_size_mb, max_tokens, max_blocks, updates, grad_accum, lr, warmup, save_eval
("tiny / KB๊ธ", 0.10, 12_000, 8, 8, 1, 1.0e-4, 0.00, 0),
("small / ์ํ", 0.50, 50_000, 32, 16, 1, 9.0e-5, 0.00, 0),
("medium / ์คํ", 2.00, 200_000, 128, 32, 2, 8.0e-5, 0.02, 50),
("large / MB๊ธ", 10.00, 1_000_000, 768, 64, 4, 7.0e-5, 0.03, 100),
("xlarge / ๋ํ", 50.00, 5_000_000, 4_096, 128, 8, 6.0e-5, 0.03, 200),
]
for profile, max_size, max_tokens, max_blocks, updates, grad_accum, lr, warmup, save_eval in rules:
if size_mb <= max_size and tokens <= max_tokens and blocks <= max_blocks:
return profile, updates, grad_accum, lr, warmup, save_eval
return "huge / ์ด๋ํ", 0, 8, 5.0e-5, 0.03, 500
def build_train_plan(stats: DataStats, train_blocks: int, eval_blocks: int) โ TrainPlan:
train_blocks = max(1, int(train_blocks))
eval_blocks = int(eval_blocks)
profile, target_updates, grad_accum, lr, warmup, save_eval = choose_profile(stats, train_blocks)
if target_updates > 0:
# ์์~๋ํ ๋๋ถ๋ถ์ update ์ ๊ธฐ์ค์ผ๋ก ์ ํ
mode = "max_steps"
max_steps = max(1, target_updates * grad_accum)
num_epochs = 1
optimizer_updates = target_updates
else:
# ์ง์ง ์ด๋ํ๋ง epoch ๊ธฐ๋ฐ
mode = "epoch"
max_steps = -1
num_epochs = 2
optimizer_updates = max(1, (train_blocks * num_epochs) // grad_accum)
# ์์ ๋ฐ์ดํฐ๋ eval/save๊ฐ ์คํ๋ ค ์๊ฐ ๋ญ๋น
if save_eval <= 0 or train_blocks <= 32 or eval_blocks <= 0:
eval_strategy = "no"
save_strategy = "no"
eval_steps = None
save_steps = None
load_best = False
early_stopping = False
else:
eval_strategy = "steps"
save_strategy = "steps"
eval_steps = save_eval
save_steps = save_eval
load_best = True
early_stopping = True
return TrainPlan(
profile=profile,
mode=mode,
train_blocks=train_blocks,
eval_blocks=eval_blocks,
max_steps=max_steps,
num_train_epochs=num_epochs,
gradient_accumulation_steps=grad_accum,
optimizer_updates=optimizer_updates,
learning_rate=lr,
warmup_ratio=warmup,
eval_strategy=eval_strategy,
save_strategy=save_strategy,
eval_steps=eval_steps,
save_steps=save_steps,
load_best_model_at_end=load_best,
early_stopping=early_stopping,
)
def print_train_plan(plan: TrainPlan) โ None:
print(โ=โ * 60)
print(โ[auto training plan]โ)
print(f"profile : {plan.profile}โ)
print(f"mode : {plan.mode}โ)
print(f"train/eval blocks : {plan.train_blocks:,} / {plan.eval_blocks:,}โ)
print(f"epochs : {plan.num_train_epochs}โ)
print(f"max steps : {plan.max_steps}โ)
print(f"grad accumulation : {plan.gradient_accumulation_steps}โ)
print(f"optimizer updates : {plan.optimizer_updates:,}โ)
print(f"learning rate : {plan.learning_rate}โ)
print(f"warmup ratio : {plan.warmup_ratio}โ)
print(f"eval/save : {plan.eval_strategy} / {plan.save_strategy}โ)
print(โ=โ * 60)
============================================================
๊ฒ์ฆ ํ๋กฌํํธ ์๋ ์์ฑ
============================================================
def safe_cut_at_sentence(text: str, min_len: int = 35, max_len: int = 100) โ str:
text = normalize_text(text)
if not text:
return โโ
if len(text) <= max_len:
return text
endings = ["๋ค.", "์.", "๊น?", "๊น.", "!", "?", ".", "โ", "\""]
best = -1
for ending in endings:
pos = text.rfind(ending, min_len, max_len)
if pos >= 0:
best = max(best, pos + len(ending))
if best >= min_len:
return text[:best].strip()
cut = text.rfind(" ", min_len, max_len)
if cut < min_len:
cut = max_len
return text[:cut].strip()
def make_segment_prompt(text: str, ratio: float, target_len: int = 95) โ str:
โโ"
๊ธด ๊ธ์ ์/์ค/ํ๋ฐ ์์น์์ ์์ฐ์ค๋ฌ์ด ์งง์ ์ด์ด์ฐ๊ธฐ ํ๋กฌํํธ๋ฅผ ๋ง๋ ๋ค.
๊ธด ์๋ฌธ ์ ์ฒด๋ฅผ ๋ฃ์ง ์๋๋ค.
โโ"
text = normalize_text(text)
if not text:
return โโ
n = len(text)
start = int(n * ratio)
start = max(0, min(start, max(0, n - target_len - 1)))
window_start = max(0, start - 140)
window_end = min(n, start + target_len + 180)
window = text[window_start:window_end]
local_start = start - window_start
# ๋ฌธ์ฅ ์ค๊ฐ์์ ์์ํ๋ ๊ฒ์ ์ค์ธ๋ค.
prev_candidates = []
for ending in ["๋ค.", "์.", "!", "?", "."]:
prev_candidates.append(window.rfind(ending, 0, local_start))
prev = max(prev_candidates)
if prev >= 0 and local_start - prev <= 120:
local_start = prev + 2
segment = window[local_start : local_start + target_len + 80].strip()
return safe_cut_at_sentence(segment, min_len=30, max_len=target_len)
def build_prompt_items(raw_items: List[Dict[str, Any]], max_prompts: int = 6) โ List[PromptItem]:
if GENERATION_TEST_PROMPTS:
manual: List[PromptItem] =
for idx, p in enumerate(GENERATION_TEST_PROMPTS, start=1):
prompt = safe_cut_at_sentence(p, min_len=20, max_len=100)
if prompt:
manual.append(PromptItem(prompt=prompt, position=โmanualโ, source_chars=len(p), line_no=idx))
return manual[:max_prompts]
candidates: List[PromptItem] = []
for item in raw_items:
text = item["text"]
chars = int(item["chars"])
line_no = int(item["line_no"])
if chars < 600:
positions = [("beginning", 0.00)]
elif chars < 2500:
positions = [("beginning", 0.00), ("middle", 0.45)]
else:
positions = [("beginning", 0.00), ("middle", 0.45), ("late", 0.78)]
for pos_name, ratio in positions:
prompt = make_segment_prompt(text, ratio=ratio)
if len(prompt) >= 20:
candidates.append(PromptItem(prompt=prompt, position=pos_name, source_chars=chars, line_no=line_no))
if not candidates:
return [
PromptItem("์ค๋์ ์ด์ํ๊ฒ ๋ง์์ด ๋ณต์กํ๋ค.", "fallback", 0, 0),
PromptItem("๋๋ ์กฐ์ฉํ ๊ณ ๊ฐ๋ฅผ ๋ค์๋ค.", "fallback", 0, 0),
PromptItem("๊ทธ๋ ๋ฌธ๋ ์ด์ํ ์๊ฐ์ด ๋ค์๋ค.", "fallback", 0, 0),
]
def score(item: PromptItem) -> Tuple[int, int, int]:
# ๊ธด ๊ธ์ middle/late๋ฅผ ์ฐ์ ํ๋ณด
pos_score = {"middle": 0, "late": 1, "beginning": 2}.get(item.position, 3)
return (pos_score, -item.source_chars, item.line_no)
candidates.sort(key=score)
selected: List[PromptItem] = []
seen_prompt_keys = set()
seen_positions = set()
# 1์ฐจ: position ๋ค์์ฑ ํ๋ณด
for item in candidates:
if len(selected) >= max_prompts:
break
key = item.prompt[:80]
if key in seen_prompt_keys:
continue
if item.position in seen_positions and len(seen_positions) < 3:
continue
selected.append(item)
seen_prompt_keys.add(key)
seen_positions.add(item.position)
# 2์ฐจ: ๋จ์ ์นธ ์ฑ์ฐ๊ธฐ
for item in candidates:
if len(selected) >= max_prompts:
break
key = item.prompt[:80]
if key in seen_prompt_keys:
continue
selected.append(item)
seen_prompt_keys.add(key)
return selected[:max_prompts]
============================================================
์์ฑ ๊ฒ์ฆ/๋ฆฌํฌํธ
============================================================
def generate_samples(
model: AutoModelForCausalLM,
tokenizer: AutoTokenizer,
prompts: List[PromptItem],
label: str,
out_dir: str,
) โ Path:
out_path = Path(out_dir)
out_path.mkdir(parents=True, exist_ok=True)
file_path = out_path / f"{label}_sample.txt"
was_training = model.training
model.eval()
gen_config = GenerationConfig(
max_new_tokens=GENERATION_MAX_NEW_TOKENS,
do_sample=True,
temperature=0.72,
top_p=0.88,
repetition_penalty=1.18,
no_repeat_ngram_size=4,
pad_token_id=tokenizer.pad_token_id,
eos_token_id=tokenizer.eos_token_id,
)
lines: List[str] = []
lines.append("=" * 80)
lines.append(f"[{label} generation sample]")
lines.append("=" * 80)
with torch.no_grad():
for idx, item in enumerate(prompts, start=1):
prompt = item.prompt
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
input_len = inputs["input_ids"].shape[-1]
output_ids = model.generate(**inputs, generation_config=gen_config)
continuation = tokenizer.decode(output_ids[0][input_len:], skip_special_tokens=True).strip()
full_text = tokenizer.decode(output_ids[0], skip_special_tokens=True)
lines.append("")
lines.append("-" * 80)
lines.append(
f"[prompt {idx}] position={item.position}, source_chars={item.source_chars}, line={item.line_no}"
)
lines.append(prompt)
lines.append("")
lines.append(f"[continuation {idx}]")
lines.append(continuation if continuation else "(์์ฑ๋ continuation ์์)")
lines.append("")
lines.append(f"[full output {idx}]")
lines.append(full_text)
file_path.write_text("\n".join(lines), encoding="utf-8")
if was_training:
model.train()
print(f"[generation] saved: {file_path.resolve()}")
return file_path
def simple_repetition_score(text: str) โ Dict[str, Any]:
words = normalize_text(text).split()
if not words:
return {โwordsโ: 0, โunique_wordsโ: 0, โunique_ratioโ: 0.0, โdot_countโ: text.count(โ.โ)}
unique_words = len(set(words))
return {
"words": len(words),
"unique_words": unique_words,
"unique_ratio": round(unique_words / max(1, len(words)), 4),
"dot_count": text.count("."),
}
def write_train_report(
stats: DataStats,
plan: TrainPlan,
prompts: List[PromptItem],
before_path: Optional[Path],
after_path: Optional[Path],
elapsed_seconds: Optional[float],
) โ Path:
report_dir = Path(GENERATION_TEST_DIR)
report_dir.mkdir(parents=True, exist_ok=True)
report_path = report_dir / โtrain_report.txtโ
lines: List[str] = []
lines.append("=" * 80)
lines.append("[train report]")
lines.append("=" * 80)
lines.append("")
lines.append("[config]")
lines.append(f"model : {MODEL_NAME}")
lines.append(f"data path : {DATA_PATH}")
lines.append(f"output dir : {OUTPUT_DIR}")
lines.append(f"max seq length : {MAX_SEQ_LENGTH}")
lines.append(f"bf16/fp16 : {USE_BF16} / {not USE_BF16}")
lines.append("")
lines.append("[data stats]")
for k, v in asdict(stats).items():
lines.append(f"{k:24}: {v}")
lines.append("")
lines.append("[train plan]")
for k, v in asdict(plan).items():
lines.append(f"{k:24}: {v}")
lines.append("")
if elapsed_seconds is not None:
lines.append("[time]")
lines.append(f"elapsed_seconds : {elapsed_seconds:.1f}")
lines.append(f"elapsed_minutes : {elapsed_seconds / 60:.2f}")
lines.append("")
lines.append("[generation files]")
lines.append(f"before : {before_path}")
lines.append(f"after : {after_path}")
lines.append("")
lines.append("[prompts]")
for idx, item in enumerate(prompts, start=1):
lines.append(
f"{idx}. position={item.position}, source_chars={item.source_chars}, "
f"line={item.line_no} | {item.prompt}"
)
lines.append("")
lines.append("[judge guide]")
lines.append("- before/after continuation์ ๋ฌธ์ฒด ๋ณํ๊ฐ ์๋์ง ๋ณธ๋ค.")
lines.append("- after๊ฐ ์๋ฌธ์ ๊ทธ๋๋ก ๋ฒ ๋ผ๋ฉด ๊ณผ์ ํฉ ๊ฐ๋ฅ์ฑ์ด ์๋ค.")
lines.append("- after๊ฐ ์ (.) ๋ฐ๋ณต, ๊ฐ์ ๊ตฌ์ ๋ฐ๋ณต, ํ์๋ช
๋ฐ๋ณต์ ๋ณด์ด๋ฉด update ์๋ฅผ ์ค์ธ๋ค.")
lines.append("- after๊ฐ before์ ๊ฑฐ์ ๊ฐ์ผ๋ฉด update ์๋ฅผ ๋๋ฆฌ๊ฑฐ๋ ๋ฐ์ดํฐ๋์ ๋๋ฆฐ๋ค.")
lines.append("- ์งง์ ๊ธ์ beginning continuation, ๊ธด ๊ธ์ middle/late continuation ๋ณํ๊น์ง ๊ฐ์ด ๋ณธ๋ค.")
lines.append("- KB๊ธ ๋ฐ์ดํฐ๋ ์ง์ ์ฃผ์
๋ณด๋ค ๋ฌธ์ฒด ํํธ ์ ๋๋ก ๋ณด๋ ๊ฒ์ด ์์ ํ๋ค.")
report_path.write_text("\n".join(lines), encoding="utf-8")
print(f"[report] saved: {report_path.resolve()}")
return report_path
============================================================
TrainingArguments ํธํ ์ฒ๋ฆฌ
============================================================
def add_training_args_compat(kwargs: Dict[str, Any]) โ Dict[str, Any]:
sig = inspect.signature(TrainingArguments.init)
eval_value = kwargs.pop("_eval_strategy_value", "steps")
kwargs = {k: v for k, v in kwargs.items() if v is not None}
if "eval_strategy" in sig.parameters:
kwargs["eval_strategy"] = eval_value
elif "evaluation_strategy" in sig.parameters:
kwargs["evaluation_strategy"] = eval_value
if "optim" in sig.parameters:
kwargs["optim"] = "paged_adamw_8bit"
if "gradient_checkpointing" in sig.parameters:
kwargs["gradient_checkpointing"] = True
return kwargs
class PeriodicCleanupCallback(TrainerCallback):
def init(self, every_n_steps: int = 50):
self.every_n_steps = every_n_steps
def on_step_end(self, args, state, control, **kwargs):
if state.global_step and state.global_step % self.every_n_steps == 0:
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
============================================================
main
============================================================
def main() โ None:
set_seed(RANDOM_SEED)
tokenizer = AutoTokenizer.from_pretrained(
MODEL_NAME,
use_fast=True,
trust_remote_code=True,
)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "right"
dataset, stats, raw_items = load_jsonl(DATA_PATH, tokenizer)
split = dataset.train_test_split(
test_size=VALID_RATIO,
seed=RANDOM_SEED,
shuffle=True,
)
def _tokenize(examples):
return tokenize_for_continuation(examples, tokenizer)
tokenized = split.map(
_tokenize,
batched=True,
remove_columns=split["train"].column_names,
desc="Tokenizing",
)
train_dataset = tokenized["train"].map(
pack_blocks,
batched=True,
desc="Packing train blocks",
)
eval_dataset = tokenized["test"].map(
pack_blocks,
batched=True,
desc="Packing eval blocks",
)
if len(train_dataset) == 0:
raise ValueError("train_dataset์ด 0๋ธ๋ก์
๋๋ค. ๋ฐ์ดํฐ ๋๋ MAX_SEQ_LENGTH ์ค์ ์ ํ์ธํ์ธ์.")
if len(eval_dataset) == 0:
print("[warn] eval block์ด 0๊ฐ๋ผ์ eval์ ๋๋๋ค.")
print("=" * 60)
print("[packed blocks]")
print(f"train blocks : {len(train_dataset):,}")
print(f"eval blocks : {len(eval_dataset):,}")
print(f"tokens per block : {MAX_SEQ_LENGTH:,}")
print("=" * 60)
plan = build_train_plan(stats, len(train_dataset), len(eval_dataset)) if AUTO_TUNE else TrainPlan(
profile="manual",
mode="epoch",
train_blocks=len(train_dataset),
eval_blocks=len(eval_dataset),
max_steps=-1,
num_train_epochs=50,
gradient_accumulation_steps=8,
optimizer_updates=max(1, (len(train_dataset) * 50) // 8),
learning_rate=8e-5,
warmup_ratio=0.03,
eval_strategy="steps",
save_strategy="steps",
eval_steps=100,
save_steps=100,
load_best_model_at_end=True,
early_stopping=True,
)
print_train_plan(plan)
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
bnb_4bit_compute_dtype=torch.bfloat16 if USE_BF16 else torch.float16,
)
model = AutoModelForCausalLM.from_pretrained(
MODEL_NAME,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True,
)
model.config.use_cache = False
model = prepare_model_for_kbit_training(model)
peft_config = LoraConfig(
r=16,
lora_alpha=32,
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM",
target_modules=[
"q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj",
],
)
model = get_peft_model(model, peft_config)
model.print_trainable_parameters()
prompts = build_prompt_items(raw_items, max_prompts=6) if AUTO_GENERATION_TEST else []
before_path: Optional[Path] = None
after_path: Optional[Path] = None
if AUTO_GENERATION_TEST:
before_path = generate_samples(
model=model,
tokenizer=tokenizer,
prompts=prompts,
label="before",
out_dir=GENERATION_TEST_DIR,
)
collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)
training_args_kwargs = dict(
output_dir=OUTPUT_DIR,
_eval_strategy_value=plan.eval_strategy,
per_device_train_batch_size=1,
gradient_accumulation_steps=plan.gradient_accumulation_steps,
num_train_epochs=plan.num_train_epochs,
max_steps=plan.max_steps,
learning_rate=plan.learning_rate,
lr_scheduler_type="cosine",
warmup_ratio=plan.warmup_ratio,
weight_decay=0.01,
logging_steps=10,
eval_steps=plan.eval_steps,
save_strategy=plan.save_strategy,
save_steps=plan.save_steps,
save_total_limit=SAVE_TOTAL_LIMIT,
load_best_model_at_end=plan.load_best_model_at_end,
metric_for_best_model="eval_loss" if plan.load_best_model_at_end else None,
greater_is_better=False if plan.load_best_model_at_end else None,
bf16=USE_BF16,
fp16=not USE_BF16,
report_to="none",
remove_unused_columns=False,
dataloader_num_workers=0,
)
training_args = TrainingArguments(**add_training_args_compat(training_args_kwargs))
callbacks: List[TrainerCallback] = [PeriodicCleanupCallback(every_n_steps=50)]
if plan.early_stopping:
callbacks.insert(0, EarlyStoppingCallback(early_stopping_patience=3))
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
eval_dataset=eval_dataset if plan.eval_strategy != "no" else None,
data_collator=collator,
callbacks=callbacks,
)
start_time = time.time()
trainer.train()
elapsed = time.time() - start_time
if AUTO_GENERATION_TEST:
after_path = generate_samples(
model=model,
tokenizer=tokenizer,
prompts=prompts,
label="after",
out_dir=GENERATION_TEST_DIR,
)
write_train_report(
stats=stats,
plan=plan,
prompts=prompts,
before_path=before_path,
after_path=after_path,
elapsed_seconds=elapsed,
)
model.save_pretrained(OUTPUT_DIR)
tokenizer.save_pretrained(OUTPUT_DIR)
print(f"[done] saved adapter/tokenizer to: {Path(OUTPUT_DIR).resolve()}")
if AUTO_GENERATION_TEST:
print(f"[done] generation check dir: {Path(GENERATION_TEST_DIR).resolve()}")
if name == โmainโ:
main()
I was trying to use the train.jsonl file universally. This is an attempt to automatically process cases where the number of samples is large but the amount of text is small, or, conversely, when the number of samples is small but the amount of text is large.
It was a learning experience.
However, it ended up being an explanatory model rather than the creative model I wanted.
Chetgpt says itโs my fault.
Iโm insisting that I chose the wrong base model.
Chet gpt is the one who recommended llama-3-Korean-Bllossom-8B, and now he is saying that I chose the wrong base model, which is absurd.
What do you guys think?