|
|
import argparse |
|
|
import json |
|
|
import inspect |
|
|
import math |
|
|
import time |
|
|
from pathlib import Path |
|
|
from typing import Any, Dict, Optional, Tuple, List |
|
|
|
|
|
import torch |
|
|
import yaml |
|
|
from datasets import load_dataset, DatasetDict |
|
|
from huggingface_hub import snapshot_download |
|
|
from transformers import ( |
|
|
AutoModelForCausalLM, |
|
|
AutoTokenizer, |
|
|
PreTrainedTokenizerFast, |
|
|
TrainingArguments, |
|
|
Trainer, |
|
|
TrainerCallback, |
|
|
default_data_collator, |
|
|
set_seed, |
|
|
) |
|
|
from transformers.trainer_utils import get_last_checkpoint |
|
|
from peft import ( |
|
|
LoraConfig, |
|
|
get_peft_model, |
|
|
prepare_model_for_kbit_training, |
|
|
PeftModel, |
|
|
) |
|
|
|
|
|
try: |
|
|
from transformers import BitsAndBytesConfig |
|
|
except ImportError: |
|
|
BitsAndBytesConfig = None |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _dtype_from_str(s: str) -> torch.dtype: |
|
|
s = (s or "").lower() |
|
|
if s in ("float16", "fp16"): |
|
|
return torch.float16 |
|
|
if s in ("bfloat16", "bf16"): |
|
|
return torch.bfloat16 |
|
|
if s in ("float32", "fp32"): |
|
|
return torch.float32 |
|
|
raise ValueError(f"Unknown torch_dtype: {s}") |
|
|
|
|
|
def _now_iso() -> str: |
|
|
return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime()) |
|
|
|
|
|
def _safe_exp(x: float) -> float: |
|
|
x = min(float(x), 50.0) |
|
|
return float(math.exp(x)) |
|
|
|
|
|
def _ensure_dir(p: Path) -> Path: |
|
|
p.mkdir(parents=True, exist_ok=True) |
|
|
return p |
|
|
|
|
|
def _looks_like_model_dir(p: Path) -> bool: |
|
|
if not p.exists() or not p.is_dir(): |
|
|
return False |
|
|
if (p / "config.json").exists(): |
|
|
return True |
|
|
if any(p.glob("*.safetensors")) or any(p.glob("pytorch_model*.bin")): |
|
|
return True |
|
|
return False |
|
|
|
|
|
def _detect_text_field(example: Dict[str, Any]) -> Optional[str]: |
|
|
for k, v in example.items(): |
|
|
if isinstance(v, str) and v.strip(): |
|
|
return k |
|
|
return None |
|
|
|
|
|
def _load_tokenizer(base_dir: Path, use_fast: bool, trust_remote_code: bool): |
|
|
try: |
|
|
return AutoTokenizer.from_pretrained( |
|
|
str(base_dir), |
|
|
use_fast=use_fast, |
|
|
trust_remote_code=trust_remote_code, |
|
|
) |
|
|
except ValueError as e: |
|
|
if "TokenizersBackend" not in str(e): |
|
|
raise |
|
|
tok_file = base_dir / "tokenizer.json" |
|
|
tok_cfg_path = base_dir / "tokenizer_config.json" |
|
|
if not tok_file.exists(): |
|
|
raise |
|
|
|
|
|
tok_kwargs: Dict[str, Any] = {} |
|
|
if tok_cfg_path.exists(): |
|
|
with tok_cfg_path.open("r", encoding="utf-8") as f: |
|
|
tok_cfg = json.load(f) |
|
|
for key in ("bos_token", "eos_token", "pad_token", "unk_token", "model_max_length"): |
|
|
if tok_cfg.get(key) is not None: |
|
|
tok_kwargs[key] = tok_cfg[key] |
|
|
extra = tok_cfg.get("additional_special_tokens") or tok_cfg.get("extra_special_tokens") |
|
|
if extra: |
|
|
tok_kwargs["additional_special_tokens"] = extra |
|
|
|
|
|
return PreTrainedTokenizerFast(tokenizer_file=str(tok_file), **tok_kwargs) |
|
|
|
|
|
def _infer_target_modules(model) -> List[str]: |
|
|
names = set() |
|
|
for n, _ in model.named_modules(): |
|
|
names.add(n.split(".")[-1]) |
|
|
|
|
|
for group in [ |
|
|
["q_proj", "k_proj", "v_proj", "o_proj"], |
|
|
["Wqkv", "out_proj"], |
|
|
["query_key_value", "dense"], |
|
|
["c_attn", "c_proj"], |
|
|
]: |
|
|
if all(x in names for x in group): |
|
|
return group |
|
|
|
|
|
fallback = [x for x in ["q_proj", "k_proj", "v_proj", "o_proj", "c_attn", "c_proj", "out_proj", "dense"] if x in names] |
|
|
if fallback: |
|
|
return fallback |
|
|
|
|
|
raise ValueError("Could not auto-infer target_modules. Set peft.target_modules explicitly.") |
|
|
|
|
|
def _choose_attn_impl(cfg: Dict[str, Any]) -> Optional[str]: |
|
|
return cfg.get("model", {}).get("attn_implementation", None) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class JsonlLoggerCallback(TrainerCallback): |
|
|
def __init__(self, run_dir: Path): |
|
|
self.run_dir = run_dir |
|
|
self.train_log_path = _ensure_dir(run_dir / "logs") / "train.jsonl" |
|
|
self.eval_log_path = _ensure_dir(run_dir / "logs") / "eval.jsonl" |
|
|
self.start_time = None |
|
|
|
|
|
def _eta(self, global_step: int, max_steps: int) -> Optional[str]: |
|
|
if self.start_time is None or global_step <= 0 or max_steps <= 0: |
|
|
return None |
|
|
elapsed = time.time() - self.start_time |
|
|
sec_per_step = elapsed / global_step |
|
|
remaining = max(0, max_steps - global_step) * sec_per_step |
|
|
h = int(remaining // 3600) |
|
|
m = int((remaining % 3600) // 60) |
|
|
s = int(remaining % 60) |
|
|
return f"{h:02d}:{m:02d}:{s:02d}" |
|
|
|
|
|
def on_train_begin(self, args, state, control, **kwargs): |
|
|
self.start_time = time.time() |
|
|
|
|
|
def on_log(self, args, state, control, logs=None, **kwargs): |
|
|
if not logs: |
|
|
return |
|
|
|
|
|
max_steps = int(state.max_steps) if getattr(state, "max_steps", None) else 0 |
|
|
progress_pct = (100.0 * state.global_step / max_steps) if max_steps > 0 else None |
|
|
epoch_pct = None |
|
|
if state.epoch is not None and args.num_train_epochs and args.num_train_epochs > 0: |
|
|
epoch_pct = 100.0 * (float(state.epoch) / float(args.num_train_epochs)) |
|
|
|
|
|
payload = { |
|
|
"ts": _now_iso(), |
|
|
"event": "train_log", |
|
|
"step": int(state.global_step), |
|
|
"epoch": round(float(state.epoch), 4) if state.epoch is not None else None, |
|
|
"progress_pct": round(progress_pct, 2) if progress_pct is not None else None, |
|
|
"epoch_pct": round(epoch_pct, 2) if epoch_pct is not None else None, |
|
|
"eta": self._eta(int(state.global_step), max_steps), |
|
|
"max_grad_norm": getattr(args, "max_grad_norm", None), |
|
|
**logs, |
|
|
} |
|
|
|
|
|
with self.train_log_path.open("a", encoding="utf-8") as f: |
|
|
f.write(json.dumps(payload, ensure_ascii=False) + "\n") |
|
|
|
|
|
def on_evaluate(self, args, state, control, metrics=None, **kwargs): |
|
|
if not metrics: |
|
|
return |
|
|
eval_loss = metrics.get("eval_loss", None) |
|
|
ppl = _safe_exp(eval_loss) if eval_loss is not None else None |
|
|
|
|
|
payload = { |
|
|
"ts": _now_iso(), |
|
|
"event": "eval", |
|
|
"step": int(state.global_step), |
|
|
"epoch": float(state.epoch) if state.epoch is not None else None, |
|
|
**metrics, |
|
|
"perplexity": ppl, |
|
|
} |
|
|
with self.eval_log_path.open("a", encoding="utf-8") as f: |
|
|
f.write(json.dumps(payload, ensure_ascii=False) + "\n") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def build_datasets(cfg: Dict[str, Any], tokenizer) -> Tuple[Any, Any]: |
|
|
data_cfg = cfg["data"] |
|
|
train_path = data_cfg["train_jsonl"] |
|
|
eval_path = data_cfg.get("eval_jsonl", None) |
|
|
split_ratio = float(data_cfg.get("eval_split_ratio", 0.0)) |
|
|
text_field = data_cfg.get("text_field", "text") |
|
|
block_size = int(data_cfg.get("block_size", 2048)) |
|
|
shuffle = bool(data_cfg.get("shuffle", True)) |
|
|
num_proc = int(data_cfg.get("num_proc", 4)) |
|
|
|
|
|
pack_mode = str(data_cfg.get("pack_mode", "drop")).lower().strip() |
|
|
if pack_mode not in ("drop", "pad"): |
|
|
raise ValueError(f"data.pack_mode must be 'drop' or 'pad', got: {pack_mode}") |
|
|
|
|
|
eos_id = tokenizer.eos_token_id |
|
|
if eos_id is None: |
|
|
raise ValueError("Tokenizer has no eos_token_id; CPT packing needs an EOS delimiter.") |
|
|
|
|
|
if tokenizer.pad_token_id is None: |
|
|
|
|
|
tokenizer.pad_token = tokenizer.eos_token |
|
|
pad_id = tokenizer.pad_token_id |
|
|
|
|
|
ds = load_dataset("json", data_files={"train": train_path}) |
|
|
|
|
|
if eval_path: |
|
|
ds_eval = load_dataset("json", data_files={"eval": eval_path}) |
|
|
dsd = DatasetDict({"train": ds["train"], "eval": ds_eval["eval"]}) |
|
|
else: |
|
|
if 0.0 < split_ratio < 1.0: |
|
|
split = ds["train"].train_test_split(test_size=split_ratio, seed=int(cfg["run"].get("seed", 42))) |
|
|
dsd = DatasetDict({"train": split["train"], "eval": split["test"]}) |
|
|
else: |
|
|
dsd = DatasetDict({"train": ds["train"], "eval": None}) |
|
|
|
|
|
if text_field not in dsd["train"].column_names: |
|
|
auto_field = _detect_text_field(dsd["train"][0]) |
|
|
if not auto_field: |
|
|
raise ValueError(f"Could not find text field. Columns: {dsd['train'].column_names}") |
|
|
text_field = auto_field |
|
|
|
|
|
def tokenize_fn(examples): |
|
|
out = tokenizer( |
|
|
examples[text_field], |
|
|
add_special_tokens=False, |
|
|
truncation=False, |
|
|
padding=False, |
|
|
) |
|
|
if "token_type_ids" in out: |
|
|
del out["token_type_ids"] |
|
|
|
|
|
out["input_ids"] = [ids + [eos_id] for ids in out["input_ids"]] |
|
|
out["attention_mask"] = [m + [1] for m in out["attention_mask"]] |
|
|
return out |
|
|
|
|
|
tokenized_train = dsd["train"].map( |
|
|
tokenize_fn, |
|
|
batched=True, |
|
|
num_proc=num_proc, |
|
|
remove_columns=dsd["train"].column_names, |
|
|
desc="Tokenizing train", |
|
|
) |
|
|
|
|
|
tokenized_eval = None |
|
|
if dsd["eval"] is not None: |
|
|
tokenized_eval = dsd["eval"].map( |
|
|
tokenize_fn, |
|
|
batched=True, |
|
|
num_proc=num_proc, |
|
|
remove_columns=dsd["eval"].column_names, |
|
|
desc="Tokenizing eval", |
|
|
) |
|
|
|
|
|
def group_texts(examples): |
|
|
concatenated = {k: sum(examples[k], []) for k in examples.keys()} |
|
|
total_length = len(concatenated["input_ids"]) |
|
|
|
|
|
if total_length == 0: |
|
|
return {"input_ids": [], "attention_mask": [], "labels": []} |
|
|
|
|
|
full_len = (total_length // block_size) * block_size |
|
|
blocks_input, blocks_attn, blocks_labels = [], [], [] |
|
|
|
|
|
|
|
|
for i in range(0, full_len, block_size): |
|
|
chunk = concatenated["input_ids"][i:i + block_size] |
|
|
attn = concatenated["attention_mask"][i:i + block_size] |
|
|
blocks_input.append(chunk) |
|
|
blocks_attn.append(attn) |
|
|
blocks_labels.append(chunk.copy()) |
|
|
|
|
|
|
|
|
remainder = total_length - full_len |
|
|
if remainder > 0 and pack_mode == "pad": |
|
|
chunk = concatenated["input_ids"][full_len:full_len + remainder] |
|
|
attn = concatenated["attention_mask"][full_len:full_len + remainder] |
|
|
|
|
|
pad_len = block_size - remainder |
|
|
chunk_padded = chunk + [pad_id] * pad_len |
|
|
attn_padded = attn + [0] * pad_len |
|
|
|
|
|
labels = chunk_padded.copy() |
|
|
labels[-pad_len:] = [-100] * pad_len |
|
|
|
|
|
blocks_input.append(chunk_padded) |
|
|
blocks_attn.append(attn_padded) |
|
|
blocks_labels.append(labels) |
|
|
|
|
|
return { |
|
|
"input_ids": blocks_input, |
|
|
"attention_mask": blocks_attn, |
|
|
"labels": blocks_labels, |
|
|
} |
|
|
|
|
|
tokenized_train = tokenized_train.map( |
|
|
group_texts, |
|
|
batched=True, |
|
|
num_proc=num_proc, |
|
|
desc=f"Packing train blocks (mode={pack_mode})", |
|
|
) |
|
|
if tokenized_eval is not None: |
|
|
tokenized_eval = tokenized_eval.map( |
|
|
group_texts, |
|
|
batched=True, |
|
|
num_proc=num_proc, |
|
|
desc=f"Packing eval blocks (mode={pack_mode})", |
|
|
) |
|
|
|
|
|
if len(tokenized_train) == 0: |
|
|
raise ValueError( |
|
|
"Train dataset is empty after packing. " |
|
|
"Either increase data, reduce block_size, or set data.pack_mode='pad'." |
|
|
) |
|
|
|
|
|
if shuffle: |
|
|
tokenized_train = tokenized_train.shuffle(seed=int(cfg["run"].get("seed", 42))) |
|
|
|
|
|
return tokenized_train, tokenized_eval |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _select_model_loader(base_dir: Path): |
|
|
cfg_path = base_dir / "config.json" |
|
|
if not cfg_path.exists(): |
|
|
return {"kind": "causal", "arch": None} |
|
|
with cfg_path.open("r", encoding="utf-8") as f: |
|
|
cfg = json.load(f) |
|
|
arch = cfg.get("architectures") or [] |
|
|
arch_name = arch[0] if arch else None |
|
|
if any("ForConditionalGeneration" in a for a in arch): |
|
|
return {"kind": "conditional", "arch": arch_name} |
|
|
return {"kind": "causal", "arch": arch_name} |
|
|
|
|
|
def _resolve_model_class(arch_name: str): |
|
|
import transformers |
|
|
cls = getattr(transformers, arch_name, None) |
|
|
if cls is None: |
|
|
raise ValueError(f"Model class '{arch_name}' is not available in installed transformers.") |
|
|
return cls |
|
|
|
|
|
|
|
|
def load_base_model_and_tokenizer(cfg: Dict[str, Any], base_dir: Path): |
|
|
model_cfg = cfg["model"] |
|
|
trust_remote_code = bool(model_cfg.get("trust_remote_code", True)) |
|
|
use_fast = bool(model_cfg.get("tokenizer_use_fast", True)) |
|
|
device_map = model_cfg.get("device_map", "auto") |
|
|
|
|
|
tokenizer = _load_tokenizer(base_dir, use_fast=use_fast, trust_remote_code=trust_remote_code) |
|
|
if tokenizer.pad_token is None: |
|
|
tokenizer.pad_token = tokenizer.eos_token |
|
|
|
|
|
torch_dtype = _dtype_from_str(model_cfg.get("torch_dtype", "bfloat16")) |
|
|
use_4bit = bool(model_cfg.get("use_4bit", False)) |
|
|
|
|
|
quant_cfg = None |
|
|
if use_4bit: |
|
|
if BitsAndBytesConfig is None: |
|
|
raise ImportError("BitsAndBytesConfig is not available in this transformers version.") |
|
|
quant_cfg = BitsAndBytesConfig( |
|
|
load_in_4bit=True, |
|
|
bnb_4bit_quant_type=str(model_cfg.get("bnb_4bit_quant_type", "nf4")), |
|
|
bnb_4bit_use_double_quant=bool(model_cfg.get("bnb_4bit_use_double_quant", True)), |
|
|
bnb_4bit_compute_dtype=_dtype_from_str(model_cfg.get("bnb_4bit_compute_dtype", "bfloat16")), |
|
|
) |
|
|
|
|
|
attn_impl = _choose_attn_impl(cfg) |
|
|
model_meta = _select_model_loader(base_dir) |
|
|
|
|
|
try: |
|
|
if model_meta["kind"] == "conditional": |
|
|
model_cls = _resolve_model_class(model_meta["arch"]) if model_meta["arch"] else None |
|
|
if model_cls is None: |
|
|
raise ValueError("Conditional model architecture not specified in config.json.") |
|
|
model = model_cls.from_pretrained( |
|
|
str(base_dir), |
|
|
device_map=device_map, |
|
|
trust_remote_code=trust_remote_code, |
|
|
low_cpu_mem_usage=True, |
|
|
torch_dtype=(torch_dtype if not use_4bit else None), |
|
|
quantization_config=quant_cfg, |
|
|
attn_implementation=attn_impl, |
|
|
) |
|
|
else: |
|
|
model = AutoModelForCausalLM.from_pretrained( |
|
|
str(base_dir), |
|
|
device_map=device_map, |
|
|
trust_remote_code=trust_remote_code, |
|
|
low_cpu_mem_usage=True, |
|
|
torch_dtype=(torch_dtype if not use_4bit else None), |
|
|
quantization_config=quant_cfg, |
|
|
attn_implementation=attn_impl, |
|
|
) |
|
|
except Exception as e: |
|
|
if attn_impl is not None: |
|
|
print(f"[warn] attn_implementation='{attn_impl}' failed: {e}") |
|
|
print("[warn] Falling back to default attention implementation.") |
|
|
if model_meta["kind"] == "conditional": |
|
|
model_cls = _resolve_model_class(model_meta["arch"]) if model_meta["arch"] else None |
|
|
if model_cls is None: |
|
|
raise ValueError("Conditional model architecture not specified in config.json.") |
|
|
model = model_cls.from_pretrained( |
|
|
str(base_dir), |
|
|
device_map=device_map, |
|
|
trust_remote_code=trust_remote_code, |
|
|
low_cpu_mem_usage=True, |
|
|
torch_dtype=(torch_dtype if not use_4bit else None), |
|
|
quantization_config=quant_cfg, |
|
|
) |
|
|
else: |
|
|
model = AutoModelForCausalLM.from_pretrained( |
|
|
str(base_dir), |
|
|
device_map=device_map, |
|
|
trust_remote_code=trust_remote_code, |
|
|
low_cpu_mem_usage=True, |
|
|
torch_dtype=(torch_dtype if not use_4bit else None), |
|
|
quantization_config=quant_cfg, |
|
|
) |
|
|
|
|
|
return model, tokenizer |
|
|
|
|
|
|
|
|
def apply_peft(cfg: Dict[str, Any], model): |
|
|
peft_cfg = cfg["peft"] |
|
|
model_cfg = cfg["model"] |
|
|
tr_cfg = cfg["train"] |
|
|
|
|
|
if not bool(peft_cfg.get("enabled", True)): |
|
|
return model, None |
|
|
|
|
|
use_4bit = bool(model_cfg.get("use_4bit", False)) |
|
|
gradient_checkpointing = bool(tr_cfg.get("gradient_checkpointing", True)) |
|
|
|
|
|
if gradient_checkpointing and hasattr(model, "gradient_checkpointing_enable"): |
|
|
model.gradient_checkpointing_enable() |
|
|
if hasattr(model, "config"): |
|
|
model.config.use_cache = False |
|
|
|
|
|
if use_4bit: |
|
|
model = prepare_model_for_kbit_training( |
|
|
model, |
|
|
use_gradient_checkpointing=gradient_checkpointing, |
|
|
) |
|
|
|
|
|
target_modules = peft_cfg.get("target_modules", "auto") |
|
|
if target_modules == "auto": |
|
|
target_modules = _infer_target_modules(model) |
|
|
|
|
|
lora_config = LoraConfig( |
|
|
r=int(peft_cfg.get("r", 16)), |
|
|
lora_alpha=int(peft_cfg.get("lora_alpha", 32)), |
|
|
lora_dropout=float(peft_cfg.get("lora_dropout", 0.05)), |
|
|
bias=str(peft_cfg.get("bias", "none")), |
|
|
task_type="CAUSAL_LM", |
|
|
target_modules=target_modules, |
|
|
) |
|
|
model = get_peft_model(model, lora_config) |
|
|
return model, lora_config |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def merge_adapter(cfg: Dict[str, Any], base_dir: Path, adapter_dir: Path, final_dir: Path): |
|
|
print(f"--- Merge: {adapter_dir} + {base_dir} -> {final_dir} ---") |
|
|
|
|
|
model_cfg = cfg["model"] |
|
|
merge_cfg = cfg.get("merge", {}) |
|
|
trust_remote_code = bool(model_cfg.get("trust_remote_code", True)) |
|
|
use_fast = bool(model_cfg.get("tokenizer_use_fast", True)) |
|
|
|
|
|
merged_dtype = _dtype_from_str(merge_cfg.get("merged_dtype", "float16")) |
|
|
max_shard_size = str(merge_cfg.get("max_shard_size", "2GB")) |
|
|
|
|
|
model_meta = _select_model_loader(base_dir) |
|
|
if model_meta["kind"] == "conditional": |
|
|
base_cls = _resolve_model_class(model_meta["arch"]) if model_meta["arch"] else None |
|
|
if base_cls is None: |
|
|
raise ValueError("Conditional model architecture not specified in config.json.") |
|
|
base = base_cls.from_pretrained( |
|
|
str(base_dir), |
|
|
torch_dtype=merged_dtype, |
|
|
device_map="cpu", |
|
|
low_cpu_mem_usage=True, |
|
|
trust_remote_code=trust_remote_code, |
|
|
) |
|
|
else: |
|
|
base = AutoModelForCausalLM.from_pretrained( |
|
|
str(base_dir), |
|
|
torch_dtype=merged_dtype, |
|
|
device_map="cpu", |
|
|
low_cpu_mem_usage=True, |
|
|
trust_remote_code=trust_remote_code, |
|
|
) |
|
|
|
|
|
merged = PeftModel.from_pretrained(base, str(adapter_dir)) |
|
|
merged = merged.merge_and_unload() |
|
|
|
|
|
_ensure_dir(final_dir) |
|
|
|
|
|
|
|
|
if hasattr(merged, '_weight_conversions'): |
|
|
merged._weight_conversions = [] |
|
|
merged.save_pretrained( |
|
|
str(final_dir), |
|
|
safe_serialization=True, |
|
|
max_shard_size=max_shard_size |
|
|
) |
|
|
|
|
|
tok = _load_tokenizer(base_dir, use_fast=use_fast, trust_remote_code=trust_remote_code) |
|
|
if tok.pad_token is None: |
|
|
tok.pad_token = tok.eos_token |
|
|
tok.save_pretrained(str(final_dir)) |
|
|
|
|
|
print("--- Merge complete ---") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def main(): |
|
|
ap = argparse.ArgumentParser() |
|
|
ap.add_argument("--config", required=True, help="Path to YAML config") |
|
|
ap.add_argument("--merge-only", action="store_true", help="Skip training, just merge adapter") |
|
|
args = ap.parse_args() |
|
|
|
|
|
with open(args.config, "r", encoding="utf-8") as f: |
|
|
cfg = yaml.safe_load(f) |
|
|
|
|
|
run_dir = _ensure_dir(Path(cfg["run"]["run_dir"])) |
|
|
_ensure_dir(run_dir / "logs") |
|
|
|
|
|
with (run_dir / "config_resolved.yaml").open("w", encoding="utf-8") as f: |
|
|
yaml.safe_dump(cfg, f, sort_keys=False) |
|
|
|
|
|
model_cfg = cfg["model"] |
|
|
repo_id = str(model_cfg["repo_id"]).strip() |
|
|
repo_path = Path(repo_id) |
|
|
|
|
|
|
|
|
if repo_path.exists() and repo_path.is_dir(): |
|
|
base_dir = repo_path |
|
|
if not _looks_like_model_dir(base_dir): |
|
|
raise ValueError(f"model.repo_id points to a directory, but it doesn't look like a HF model dir: {base_dir}") |
|
|
else: |
|
|
|
|
|
base_dir = _ensure_dir(run_dir / model_cfg.get("base_local_dir", "base_model")) |
|
|
if not _looks_like_model_dir(base_dir): |
|
|
print(f"Base model not found at {base_dir}, downloading from {repo_id} ...") |
|
|
snapshot_download( |
|
|
repo_id=repo_id, |
|
|
revision=model_cfg.get("revision", None), |
|
|
local_dir=str(base_dir), |
|
|
local_dir_use_symlinks=False, |
|
|
) |
|
|
|
|
|
ckpt_dir = _ensure_dir(run_dir / "checkpoints") |
|
|
best_adapter_dir = _ensure_dir(run_dir / "best_adapter") |
|
|
|
|
|
merge_cfg = cfg.get("merge", {}) or {} |
|
|
if merge_cfg.get("output_dir"): |
|
|
od = Path(str(merge_cfg["output_dir"])) |
|
|
final_dir = od if od.is_absolute() else (run_dir / od) |
|
|
else: |
|
|
final_dir = run_dir / "final_model" |
|
|
|
|
|
|
|
|
if args.merge_only: |
|
|
if not _looks_like_model_dir(best_adapter_dir): |
|
|
raise FileNotFoundError(f"Adapter not found at {best_adapter_dir}") |
|
|
merge_adapter(cfg, base_dir, best_adapter_dir, final_dir) |
|
|
return |
|
|
|
|
|
|
|
|
set_seed(int(cfg["run"].get("seed", 42))) |
|
|
|
|
|
model, tokenizer = load_base_model_and_tokenizer(cfg, base_dir) |
|
|
model, _ = apply_peft(cfg, model) |
|
|
|
|
|
train_ds, eval_ds = build_datasets(cfg, tokenizer) |
|
|
|
|
|
tr_cfg = cfg["train"] |
|
|
|
|
|
dtype = _dtype_from_str(model_cfg.get("torch_dtype", "bfloat16")) |
|
|
use_fp16 = (dtype == torch.float16) |
|
|
use_bf16 = (dtype == torch.bfloat16) |
|
|
|
|
|
max_steps = int(tr_cfg.get("max_steps", 0)) |
|
|
num_train_epochs = float(tr_cfg.get("num_train_epochs", 1)) |
|
|
|
|
|
|
|
|
ta_params = inspect.signature(TrainingArguments.__init__).parameters |
|
|
eval_key = "eval_strategy" if "eval_strategy" in ta_params else "evaluation_strategy" |
|
|
|
|
|
desired_ta_kwargs = dict( |
|
|
output_dir=str(ckpt_dir), |
|
|
max_steps=max_steps if max_steps > 0 else -1, |
|
|
num_train_epochs=num_train_epochs, |
|
|
|
|
|
per_device_train_batch_size=int(tr_cfg.get("per_device_train_batch_size", 1)), |
|
|
per_device_eval_batch_size=int(tr_cfg.get("per_device_eval_batch_size", tr_cfg.get("per_device_train_batch_size", 1))), |
|
|
gradient_accumulation_steps=int(tr_cfg.get("gradient_accumulation_steps", 1)), |
|
|
|
|
|
learning_rate=float(tr_cfg.get("learning_rate", 2e-5)), |
|
|
weight_decay=float(tr_cfg.get("weight_decay", 0.0)), |
|
|
warmup_ratio=float(tr_cfg.get("warmup_ratio", 0.0)), |
|
|
lr_scheduler_type=str(tr_cfg.get("lr_scheduler_type", "cosine")), |
|
|
|
|
|
optim=str(tr_cfg.get("optim", "paged_adamw_8bit" if bool(model_cfg.get("use_4bit", False)) else "adamw_torch")), |
|
|
max_grad_norm=float(tr_cfg.get("max_grad_norm", 1.0)), |
|
|
|
|
|
logging_steps=int(tr_cfg.get("logging_steps", 10)), |
|
|
|
|
|
save_strategy=str(tr_cfg.get("save_strategy", "steps")), |
|
|
save_steps=int(tr_cfg.get("save_steps", 200)), |
|
|
save_total_limit=int(tr_cfg.get("save_total_limit", 3)), |
|
|
|
|
|
eval_steps=int(tr_cfg.get("eval_steps", 200)), |
|
|
|
|
|
load_best_model_at_end=bool(tr_cfg.get("load_best_model_at_end", True)) if eval_ds is not None else False, |
|
|
metric_for_best_model="eval_loss", |
|
|
greater_is_better=False, |
|
|
|
|
|
fp16=use_fp16, |
|
|
bf16=use_bf16, |
|
|
|
|
|
report_to=[], |
|
|
remove_unused_columns=False, |
|
|
save_safetensors=True, |
|
|
overwrite_output_dir=False, |
|
|
) |
|
|
|
|
|
|
|
|
desired_ta_kwargs[eval_key] = str(tr_cfg.get("evaluation_strategy", "steps" if eval_ds is not None else "no")) |
|
|
ta_kwargs = {k: v for k, v in desired_ta_kwargs.items() if k in ta_params} |
|
|
|
|
|
training_args = TrainingArguments(**ta_kwargs) |
|
|
|
|
|
trainer_params = inspect.signature(Trainer.__init__).parameters |
|
|
desired_trainer_kwargs = dict( |
|
|
model=model, |
|
|
args=training_args, |
|
|
train_dataset=train_ds, |
|
|
eval_dataset=eval_ds, |
|
|
tokenizer=tokenizer, |
|
|
processing_class=tokenizer, |
|
|
data_collator=default_data_collator, |
|
|
callbacks=[JsonlLoggerCallback(run_dir)], |
|
|
) |
|
|
trainer_kwargs = {k: v for k, v in desired_trainer_kwargs.items() if k in trainer_params} |
|
|
trainer = Trainer(**trainer_kwargs) |
|
|
|
|
|
|
|
|
resume_from = tr_cfg.get("resume_from_checkpoint", None) |
|
|
if resume_from == "auto": |
|
|
last = get_last_checkpoint(str(ckpt_dir)) |
|
|
resume_from = last if last else None |
|
|
if resume_from: |
|
|
print(f"Resuming from {resume_from}") |
|
|
|
|
|
print("Starting training...") |
|
|
trainer.train(resume_from_checkpoint=resume_from) |
|
|
|
|
|
trainer.save_model(str(best_adapter_dir)) |
|
|
print(f"Saved best adapter -> {best_adapter_dir}") |
|
|
|
|
|
if eval_ds is not None: |
|
|
metrics = trainer.evaluate() |
|
|
eval_loss = metrics.get("eval_loss", None) |
|
|
metrics["perplexity"] = _safe_exp(eval_loss) if eval_loss is not None else None |
|
|
with (run_dir / "eval_final.json").open("w", encoding="utf-8") as f: |
|
|
json.dump(metrics, f, indent=2) |
|
|
print(f"Final eval_loss={eval_loss}, ppl={metrics['perplexity']}") |
|
|
|
|
|
if bool(cfg.get("merge", {}).get("enabled", False)): |
|
|
del trainer, model |
|
|
torch.cuda.empty_cache() |
|
|
merge_adapter(cfg, base_dir, best_adapter_dir, final_dir) |
|
|
else: |
|
|
print("Merge disabled. Run with --merge-only later if needed.") |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|