#!/usr/bin/env python3 # SPDX-License-Identifier: BUSL-1.1 # Copyright (c) 2024-2026 Lucas Ricardo Mella Chillemi """ pretrain_local.py — Continual pretrain de PamparV3 en GPU local (GTX 1650 4GB). Entrena con datos textbook en formato CLM (next token prediction). Optimizado para VRAM limitada: AMP fp16 + gradient accumulation + checkpointing. Uso: # Pretrain con datos generados (cuando la generación termine) python scripts/pretrain_local.py # Con opciones custom python scripts/pretrain_local.py --epochs 8 --lr 1e-4 --grad-accum 8 # Reanudar entrenamiento interrumpido python scripts/pretrain_local.py --resume # Esperar a que la generación termine antes de entrenar python scripts/pretrain_local.py --wait-for-data 200 Detener con Ctrl-C — guarda checkpoint antes de salir. """ import argparse import json import logging import math import random import sys import time from pathlib import Path from typing import Optional import sentencepiece as spm import torch import torch.nn.functional as F import torch.nn.utils as nn_utils ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from pampar.coder.v3 import PRESET_V3, ConfigV3, PamparV3 logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("pretrain_local") # ───────────────────────────────────────────────────────────────────────────── # Dataset # ───────────────────────────────────────────────────────────────────────────── class TextbookDataset: """ Carga textbook JSONL, tokeniza y sirve chunks aleatorios para CLM. Divide textos largos en chunks solapados de max_seq_len+1 tokens. El +1 es para tener input ([:L]) y target ([1:L+1]). """ def __init__( self, ruta_jsonl: Path, tokenizer: spm.SentencePieceProcessor, max_seq_len: int = 512, ) -> None: self.tok = tokenizer self.max_seq_len = max_seq_len self.chunks: list[list[int]] = [] self._cargar(ruta_jsonl) def _cargar(self, ruta: Path) -> None: """Lee el JSONL y tokeniza en chunks.""" n_textos = 0 for linea in ruta.read_text(encoding="utf-8").splitlines(): if not linea.strip(): continue try: obj = json.loads(linea) texto = obj.get("text", "") except json.JSONDecodeError: continue if not texto or len(texto) < 50: continue ids = self.tok.Encode(texto) n_textos += 1 # Chunks solapados (50% overlap) step = self.max_seq_len // 2 for i in range(0, max(1, len(ids) - self.max_seq_len), step): chunk = ids[i : i + self.max_seq_len + 1] if len(chunk) >= 32: self.chunks.append(chunk) logger.info( "Dataset: %d textos → %d chunks (seq_len=%d)", n_textos, len(self.chunks), self.max_seq_len, ) def __len__(self) -> int: return len(self.chunks) def get_batch( self, batch_size: int, device: torch.device, ) -> torch.Tensor: """Devuelve un batch aleatorio [B, L+1] padded.""" indices = random.sample( range(len(self.chunks)), min(batch_size, len(self.chunks)) ) seleccionados = [self.chunks[i] for i in indices] max_len = min( max(len(c) for c in seleccionados), self.max_seq_len + 1, ) padded = [] for chunk in seleccionados: trunc = chunk[:max_len] pad = [0] * (max_len - len(trunc)) padded.append(trunc + pad) return torch.tensor(padded, dtype=torch.long, device=device) @property def total_tokens(self) -> int: """Total de tokens en el dataset (sin padding).""" return sum(len(c) for c in self.chunks) # ───────────────────────────────────────────────────────────────────────────── # LR Scheduler # ───────────────────────────────────────────────────────────────────────────── def cosine_lr( step: int, warmup_steps: int, total_steps: int, lr_max: float, lr_min: float = 1e-6, ) -> float: """Cosine schedule con warmup lineal.""" if step < warmup_steps: return lr_max * (step + 1) / warmup_steps progreso = (step - warmup_steps) / max(1, total_steps - warmup_steps) return lr_min + 0.5 * (lr_max - lr_min) * (1 + math.cos(math.pi * progreso)) # ───────────────────────────────────────────────────────────────────────────── # Training Loop # ───────────────────────────────────────────────────────────────────────────── def entrenar( modelo: PamparV3, dataset: TextbookDataset, optimizer: torch.optim.Optimizer, device: torch.device, *, epochs: int, batch_size: int, grad_accum: int, max_grad_norm: float, lr_max: float, guardar_cada: int, ruta_ckpt: Path, paso_inicio: int = 0, mejor_loss: float = float("inf"), use_amp: bool = False, ) -> None: """Bucle principal de continual pretrain. AMP opcional (fp32 por defecto).""" steps_per_epoch = max(1, len(dataset) // (batch_size * grad_accum)) total_steps = steps_per_epoch * epochs warmup_steps = min(total_steps // 10, 100) logger.info( "Config: epochs=%d batch=%d grad_accum=%d effective_batch=%d amp=%s", epochs, batch_size, grad_accum, batch_size * grad_accum, use_amp, ) logger.info( "Steps: %d/epoch %d total warmup=%d lr_max=%.1e", steps_per_epoch, total_steps, warmup_steps, lr_max, ) logger.info( "Dataset: %d chunks ~%dK tokens", len(dataset), dataset.total_tokens // 1000 ) scaler = torch.amp.GradScaler("cuda", enabled=use_amp) paso = paso_inicio mejor = mejor_loss t_inicio = time.time() try: for epoch in range(1, epochs + 1): losses_epoch: list[float] = [] modelo.train() for micro_step in range(steps_per_epoch * grad_accum): # ── LR schedule ────────────────────────────────────────── lr = cosine_lr(paso, warmup_steps, total_steps, lr_max) for pg in optimizer.param_groups: pg["lr"] = lr # ── Forward + loss ─────────────────────────────────────── tokens = dataset.get_batch(batch_size, device) input_ids = tokens[:, :-1] targets = tokens[:, 1:] with torch.amp.autocast("cuda", dtype=torch.float16, enabled=use_amp): logits, loss, _info = modelo(input_ids, targets=targets) loss = loss / grad_accum # ── Backward ───────────────────────────────────────────── scaler.scale(loss).backward() loss_val = float(loss.detach()) * grad_accum losses_epoch.append(loss_val) # ── Optimizer step cada grad_accum micro-steps ──────────── if (micro_step + 1) % grad_accum == 0: scaler.unscale_(optimizer) nn_utils.clip_grad_norm_(modelo.parameters(), max_grad_norm) scaler.step(optimizer) scaler.update() optimizer.zero_grad(set_to_none=True) paso += 1 # ── Log cada 10 pasos ──────────────────────────────── if paso % 10 == 0: avg_recent = sum(losses_epoch[-grad_accum * 10 :]) / min( len(losses_epoch), grad_accum * 10 ) ppl = math.exp(min(avg_recent, 20.0)) elapsed = time.time() - t_inicio eta_s = ( (total_steps - paso) / max(1, (paso - paso_inicio) / elapsed) if elapsed > 0 else 0 ) eta_h = eta_s / 3600 logger.info( "epoch %d | paso %4d/%d | loss=%.3f ppl=%.1f | lr=%.1e | ETA=%.1fh", epoch, paso, total_steps, avg_recent, ppl, lr, eta_h, ) # ── VRAM log (solo paso 1) ─────────────────────────── if paso == paso_inicio + 1: alloc = torch.cuda.max_memory_allocated() / 1e9 logger.info("VRAM pico: %.2f GB / 4.00 GB", alloc) # ── Guardar periódico ───────────────────────────────── if guardar_cada > 0 and paso % guardar_cada == 0: _guardar_checkpoint( modelo, optimizer, paso, mejor, ruta_ckpt.parent / f"v3_pretrain_step{paso}.pt", ) # ── Fin de epoch ───────────────────────────────────────────── avg_epoch = sum(losses_epoch) / len(losses_epoch) ppl_epoch = math.exp(min(avg_epoch, 20.0)) elapsed = time.time() - t_inicio hh, mm = int(elapsed // 3600), int((elapsed % 3600) // 60) logger.info( "═══ Epoch %d/%d loss=%.3f ppl=%.1f [%02dh%02dm] ═══", epoch, epochs, avg_epoch, ppl_epoch, hh, mm, ) # Guardar checkpoint de epoch _guardar_checkpoint( modelo, optimizer, paso, mejor, ruta_ckpt.parent / f"v3_pretrain_epoch{epoch}.pt", ) # Guardar best if avg_epoch < mejor: mejor = avg_epoch _guardar_checkpoint( modelo, optimizer, paso, mejor, ruta_ckpt, ) logger.info("★ Nuevo mejor loss: %.3f", mejor) except KeyboardInterrupt: logger.info("\nInterrumpido — guardando checkpoint final...") # Guardar checkpoint final _guardar_checkpoint( modelo, optimizer, paso, mejor, ruta_ckpt.parent / "v3_pretrain_last.pt", ) elapsed = time.time() - t_inicio logger.info( "Pretrain completado — %d pasos, mejor loss=%.3f, tiempo=%.1f min", paso - paso_inicio, mejor, elapsed / 60, ) def _guardar_checkpoint( modelo: PamparV3, optimizer: torch.optim.Optimizer, paso: int, mejor_loss: float, ruta: Path, ) -> None: """Guarda checkpoint con modelo + optimizer + metadata.""" import dataclasses ruta.parent.mkdir(parents=True, exist_ok=True) torch.save( { "modelo": modelo.state_dict(), "optimizer": optimizer.state_dict(), "paso_global": paso, "config": dataclasses.asdict(modelo.config), "mejor_loss": mejor_loss, "tipo": "pretrain_local", }, ruta, ) logger.info("✓ Checkpoint → %s (paso %d)", ruta.name, paso) # ───────────────────────────────────────────────────────────────────────────── # Esperar datos # ───────────────────────────────────────────────────────────────────────────── def esperar_datos(ruta_jsonl: Path, min_ejemplos: int, intervalo: int = 30) -> None: """Espera hasta que el JSONL tenga al menos min_ejemplos líneas.""" logger.info("Esperando ≥%d ejemplos en %s...", min_ejemplos, ruta_jsonl.name) while True: if ruta_jsonl.exists(): n = sum( 1 for line in ruta_jsonl.read_text(encoding="utf-8").splitlines() if line.strip() ) if n >= min_ejemplos: logger.info("Datos listos: %d ejemplos", n) return logger.info( " %d/%d ejemplos — esperando %ds...", n, min_ejemplos, intervalo ) else: logger.info(" Archivo no existe aún — esperando %ds...", intervalo) time.sleep(intervalo) # ───────────────────────────────────────────────────────────────────────────── # CLI # ───────────────────────────────────────────────────────────────────────────── def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser( description="Continual pretrain de PamparV3 en GPU local", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) # Rutas p.add_argument( "--checkpoint-base", type=Path, default=ROOT / "checkpoints" / "v3_ghidra_v9.pt", help="Checkpoint base del que partir", ) p.add_argument( "--checkpoint-out", type=Path, default=ROOT / "checkpoints" / "v3_pretrain_best.pt", help="Ruta del checkpoint de salida (best)", ) p.add_argument( "--data", type=Path, default=ROOT / "data" / "textbook_v3" / "textbook_pretrain.jsonl", help="Datos de textbook JSONL", ) p.add_argument( "--tokenizer", type=Path, default=ROOT / "data" / "tokenizer" / "pampar_48k.model", help="SentencePiece 48K", ) # Hiperparámetros p.add_argument("--epochs", type=int, default=5, help="Número de epochs") p.add_argument("--lr", type=float, default=1e-4, help="Learning rate máximo") p.add_argument("--batch-size", type=int, default=2, help="Micro-batch size") p.add_argument( "--grad-accum", type=int, default=4, help="Gradient accumulation steps" ) p.add_argument( "--seq-len", type=int, default=512, help="Longitud máxima de secuencia" ) p.add_argument("--max-grad-norm", type=float, default=0.5, help="Gradient clipping") p.add_argument("--weight-decay", type=float, default=0.1, help="Weight decay") p.add_argument( "--guardar-cada", type=int, default=200, help="Guardar cada N pasos. 0=solo epochs", ) # Control p.add_argument( "--resume", action="store_true", help="Reanudar desde checkpoint de salida" ) p.add_argument( "--wait-for-data", type=int, default=0, help="Esperar hasta N ejemplos en el JSONL antes de empezar", ) p.add_argument( "--amp", action="store_true", help="Usar AMP fp16 (puede causar NaN en esta arquitectura)", ) return p.parse_args() def main() -> None: args = _parse_args() # ── Validar CUDA ────────────────────────────────────────────────────────── if not torch.cuda.is_available(): logger.error("CUDA no disponible. Este script requiere GPU.") sys.exit(1) device = torch.device("cuda") gpu_name = torch.cuda.get_device_name(0) vram_gb = torch.cuda.get_device_properties(0).total_memory / 1e9 logger.info("GPU: %s (%.1f GB VRAM)", gpu_name, vram_gb) # ── Esperar datos si se pide ────────────────────────────────────────────── if args.wait_for_data > 0: esperar_datos(args.data, args.wait_for_data) # ── Validar archivos ────────────────────────────────────────────────────── if not args.data.exists(): logger.error("Datos no encontrados: %s", args.data) sys.exit(1) if not args.tokenizer.exists(): logger.error("Tokenizer no encontrado: %s", args.tokenizer) sys.exit(1) # ── Tokenizer ───────────────────────────────────────────────────────────── tok = spm.SentencePieceProcessor() tok.Load(str(args.tokenizer)) logger.info("Tokenizer: vocab=%d", tok.GetPieceSize()) # ── Dataset ─────────────────────────────────────────────────────────────── dataset = TextbookDataset(args.data, tok, max_seq_len=args.seq_len) if len(dataset) == 0: logger.error("Dataset vacío — ¿generación de datos incompleta?") sys.exit(1) # ── Modelo ──────────────────────────────────────────────────────────────── ruta_base = args.checkpoint_base paso_inicio = 0 mejor_loss = float("inf") if args.resume and args.checkpoint_out.exists(): ruta_base = args.checkpoint_out logger.info("Reanudando desde %s", ruta_base) if not ruta_base.exists(): logger.error("Checkpoint base no encontrado: %s", ruta_base) sys.exit(1) logger.info("Cargando checkpoint: %s", ruta_base.name) payload = torch.load(ruta_base, map_location="cpu", weights_only=False) config = ConfigV3(**payload["config"]) if "config" in payload else PRESET_V3 modelo = PamparV3(config).to(device) modelo.load_state_dict(payload["modelo"]) if args.resume and "paso_global" in payload: paso_inicio = int(payload["paso_global"]) mejor_loss = float(payload.get("mejor_loss", float("inf"))) logger.info( "Reanudando desde paso %d, mejor_loss=%.3f", paso_inicio, mejor_loss ) n_params = sum(p.numel() for p in modelo.parameters() if p.requires_grad) logger.info( "PamparV3 — %.1fM params, gradient checkpointing=%s", n_params / 1e6, config.use_checkpoint, ) # ── Optimizer ───────────────────────────────────────────────────────────── optimizer = torch.optim.AdamW( modelo.parameters(), lr=args.lr, betas=(0.9, 0.95), weight_decay=args.weight_decay, eps=1e-8, ) if args.resume and "optimizer" in payload: try: optimizer.load_state_dict(payload["optimizer"]) logger.info("Optimizer restaurado") except Exception as e: logger.warning("No se pudo restaurar optimizer: %s", e) del payload torch.cuda.empty_cache() # ── Entrenar ────────────────────────────────────────────────────────────── logger.info("=" * 60) logger.info("CONTINUAL PRETRAIN — PamparV3 108M") logger.info("=" * 60) entrenar( modelo=modelo, dataset=dataset, optimizer=optimizer, device=device, epochs=args.epochs, batch_size=args.batch_size, grad_accum=args.grad_accum, max_grad_norm=args.max_grad_norm, lr_max=args.lr, guardar_cada=args.guardar_cada, ruta_ckpt=args.checkpoint_out, paso_inicio=paso_inicio, mejor_loss=mejor_loss, use_amp=args.amp, ) if __name__ == "__main__": main()