"""Train and measure a response-supervised CODELOCK LoRA adapter on one GPU.""" from __future__ import annotations import argparse import hashlib import importlib.metadata import json import os import time from collections import Counter from datetime import datetime, timezone from pathlib import Path import torch from peft import LoraConfig, PeftModel, get_peft_model from transformers import (AutoModelForCausalLM, AutoTokenizer, Trainer, TrainerCallback, TrainingArguments, set_seed) from codelock_affect.finetuning import ( ADAPTER_SYSTEM, CompletionCollator, build_supervised_records, dataset_digest, encode_record, state_prompt, ) def now(): return datetime.now(timezone.utc).isoformat() def write_json(path, data): temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") temporary.replace(path) def retention_probes(model, tokenizer): """Small development checks, separate from the held-out loss measurement.""" tasks = [ ("Reply with only the result of 17 + 25.", "42"), ("Reply with only the result of 9 * 7.", "63"), ("Reply with only the result of 100 - 37.", "63"), ("Reply with only the result of 81 / 9.", "9"), ("Return exactly these words in this order: bed, desk, window", "bed, desk, window"), ("Return exactly this identifier and nothing else: C-01", "C-01"), ("The recorded balance is 100. Reply with only that number.", "100"), ("The order status is pending. Reply with only that status.", "pending"), ] model.eval() rows = [] for i, (question, expected) in enumerate(tasks): levels = {name: float(i % 5 == j) for j, name in enumerate(("pain", "fear", "anxiety", "relief"))} messages = [{"role": "system", "content": ADAPTER_SYSTEM}, {"role": "user", "content": state_prompt(question, levels)}] text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True, enable_thinking=False) batch = tokenizer(text, return_tensors="pt").to(model.device) with torch.inference_mode(): output = model.generate(**batch, max_new_tokens=32, do_sample=False, use_cache=True, pad_token_id=tokenizer.pad_token_id) answer = tokenizer.decode(output[0, batch.input_ids.shape[1]:], skip_special_tokens=True).strip() rows.append({"prompt": question, "expected": expected, "answer": answer, "correct": " ".join(answer.lower().split()) == " ".join(expected.lower().split())}) return {"correct": sum(row["correct"] for row in rows), "total": len(rows), "scope": "development checks; not a broad capability benchmark", "rows": rows} class Progress(TrainerCallback): def __init__(self, output): self.output = output def on_log(self, args, state, control, logs=None, **kwargs): row = {"utc": now(), "step": state.global_step, "epoch": state.epoch, "gpu_memory_allocated_mib": torch.cuda.memory_allocated() // 1048576, **(logs or {})} with (self.output / "metrics.jsonl").open("a", encoding="utf-8") as stream: stream.write(json.dumps(row) + "\n") write_json(self.output / "progress.json", row) print(json.dumps(row), flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--model", required=True) parser.add_argument("--config", default="configs/adapter-training.json") parser.add_argument("--data", default="data/affect_pairs.jsonl") parser.add_argument("--output", required=True) parser.add_argument("--resume") parser.add_argument("--adapter", help="Continue from an existing LoRA adapter") parser.add_argument("--skip-test", action="store_true", help="Use validation and development probes for continuing rounds") args = parser.parse_args() os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") os.environ.setdefault("WANDB_DISABLED", "true") cfg = json.loads(Path(args.config).read_text(encoding="utf-8")) output = Path(args.output) if output.exists() and any(output.iterdir()) and not args.resume: raise ValueError("Use a new output directory or explicitly resume a checkpoint") output.mkdir(parents=True, exist_ok=True) started = time.monotonic() if not torch.cuda.is_available() or not torch.cuda.is_bf16_supported(): raise RuntimeError("This configuration requires a BF16-capable CUDA GPU") set_seed(cfg["seed"]) source = Path(args.data).read_bytes() pairs = [json.loads(line) for line in source.decode().splitlines() if line.strip()] records = build_supervised_records(pairs, cfg.get("split_seed", cfg["seed"])) tokenizer = AutoTokenizer.from_pretrained(args.model, trust_remote_code=False) if tokenizer.pad_token_id is None: tokenizer.pad_token = tokenizer.eos_token encoded = {split: [encode_record(tokenizer, row, cfg["max_length"]) for row in records if row["split"] == split] for split in ("train", "validation", "test")} metadata = { "started_utc": now(), "complete": False, "method": "supervised_lora", "base_weights_updated": False, "adapter_weights_updated": True, "source_dataset_sha256": hashlib.sha256(source).hexdigest(), "supervised_dataset_sha256": dataset_digest(records), "split_counts": dict(Counter(row["split"] for row in records)), "scenario_counts": {split: len({r["scenario_id"] for r in records if r["split"] == split}) for split in encoded}, "template_disjoint": False, "config": cfg, "cuda_device": torch.cuda.get_device_name(), "runtime": {name: importlib.metadata.version(name) for name in ("torch", "transformers", "peft", "accelerate", "safetensors")}, "max_encoded_length": max(len(row["input_ids"]) for data in encoded.values() for row in data), "loss_scope": "assistant completion and EOS; prompt and padding masked", "warm_start": Path(args.adapter).parent.name if args.adapter else None, "warm_start_sha256": hashlib.sha256((Path(args.adapter) / "adapter_model.safetensors").read_bytes()).hexdigest() if args.adapter else None, } write_json(output / "run.json", metadata) with (output / "supervised-data.jsonl").open("w", encoding="utf-8") as stream: for row in records: stream.write(json.dumps(row, sort_keys=True) + "\n") model = AutoModelForCausalLM.from_pretrained( args.model, torch_dtype=torch.bfloat16, attn_implementation="sdpa", trust_remote_code=False, ).to("cuda") model.config.use_cache = False lora = LoraConfig(r=cfg["lora_rank"], lora_alpha=cfg["lora_alpha"], lora_dropout=cfg["lora_dropout"], target_modules=cfg["target_modules"], bias="none", task_type="CAUSAL_LM") model = (PeftModel.from_pretrained(model, args.adapter, is_trainable=True) if args.adapter else get_peft_model(model, lora)) model.enable_input_require_grads() metadata["trainable_parameters"], metadata["total_parameters"] = model.get_nb_trainable_parameters() write_json(output / "run.json", metadata) print(json.dumps({"event": "model_ready", **metadata}), flush=True) training_args = TrainingArguments( output_dir=str(output / "checkpoints"), num_train_epochs=cfg["epochs"], learning_rate=cfg["learning_rate"], per_device_train_batch_size=cfg["batch_size"], per_device_eval_batch_size=cfg["batch_size"], gradient_accumulation_steps=cfg["gradient_accumulation_steps"], weight_decay=cfg["weight_decay"], max_grad_norm=cfg["max_grad_norm"], warmup_ratio=cfg["warmup_ratio"], lr_scheduler_type="cosine", bf16=True, gradient_checkpointing=True, gradient_checkpointing_kwargs={"use_reentrant": False}, logging_strategy="steps", logging_steps=5, logging_first_step=True, eval_strategy="epoch", save_strategy="epoch", save_total_limit=cfg.get("save_total_limit", 3), load_best_model_at_end=True, metric_for_best_model="eval_loss", greater_is_better=False, report_to=[], seed=cfg["seed"], data_seed=cfg["seed"], remove_unused_columns=False, dataloader_num_workers=0, optim="adamw_torch", disable_tqdm=True, ) trainer = Trainer(model=model, args=training_args, processing_class=tokenizer, train_dataset=encoded["train"], eval_dataset=encoded["validation"], data_collator=CompletionCollator(tokenizer.pad_token_id), callbacks=[Progress(output)]) baseline = trainer.evaluate(metric_key_prefix="before_validation") write_json(output / "baseline-validation.json", baseline) probes_before = retention_probes(model, tokenizer) write_json(output / "probes-before.json", probes_before) result = trainer.train(resume_from_checkpoint=args.resume) # The held-out test split is evaluated once after validation selects a checkpoint. final_validation = trainer.evaluate(metric_key_prefix="validation") final_test = None if args.skip_test else trainer.evaluate(encoded["test"], metric_key_prefix="test") probes_after = retention_probes(model, tokenizer) write_json(output / "probes-after.json", probes_after) selected_step = Path(trainer.state.best_model_checkpoint).name adapter = output / "adapter" model.peft_config["default"].base_model_name_or_path = "Qwen/Qwen3-8B" model.save_pretrained(adapter, safe_serialization=True) tokenizer.save_pretrained(adapter) (adapter / "README.md").write_text( "# CODELOCK-8B / LoRA adapter\n\n" "Response-supervised adapter for CODELOCK's four-channel state interface. " "Load the compatible backbone, then attach this adapter with PEFT.\n\n" "See the accompanying run record for data hashes, checkpoint selection, " "validation loss and development checks.\n", encoding="utf-8") metadata.update({"complete": True, "completed_utc": now(), "elapsed_seconds": time.monotonic() - started, "optimizer_steps": trainer.state.global_step, "selected_checkpoint": selected_step, "best_validation_loss": trainer.state.best_metric, "before_validation": baseline, "after_validation": final_validation, "test": final_test, "test_evaluated": not args.skip_test, "probes_before": probes_before, "probes_after": probes_after, "train": result.metrics, "peak_cuda_memory_mib": torch.cuda.max_memory_allocated() // 1048576}) write_json(output / "run.json", metadata) trainer.save_state() print(json.dumps({"event": "training_complete", **metadata}), flush=True) if __name__ == "__main__": main()