````instructions # Python ML/LLM Instructions > Para desarrollo de modelos de ML/LLM con PyTorch. ## Type Hints (OBLIGATORIO) ```python from typing import Optional, Tuple, Dict, List, Union, Literal from torch import Tensor import torch.nn as nn def forward( self, input_ids: Tensor, attention_mask: Optional[Tensor] = None, labels: Optional[Tensor] = None, ) -> Tuple[Tensor, Optional[Tensor]]: """ Forward pass del modelo. Args: input_ids: Token IDs, shape (batch, seq_len). attention_mask: Máscara de atención, shape (batch, seq_len). labels: Labels para calcular loss, shape (batch, seq_len). Returns: Tuple de (logits, loss). Loss es None si labels no se proporcionan. """ ... ```` ## Docstrings (Google Style) ```python class PampaRCoderV2(nn.Module): """ Modelo principal PAMPAr-Coder V2 con arquitectura cerebral. Attributes: config: Configuración del modelo. embedding: Capa de embedding de tokens. talamo: Orquestador central TálamoBrodmann. territorios: Lista de 4 BloqueTerrritorial. Example: >>> config = ConfigPampaRCoderV2.from_preset("1.5B") >>> model = PampaRCoderV2(config) >>> output = model(input_ids) """ ``` ## PyTorch Patterns ### Model Definition ```python class MiModulo(nn.Module): def __init__(self, config: ConfigPampaRCoderV2): super().__init__() self.config = config # Inicializar layers aquí def forward(self, x: Tensor) -> Tensor: # Forward pass return x def _init_weights(self, module: nn.Module) -> None: """Inicialización de pesos.""" if isinstance(module, nn.Linear): nn.init.normal_(module.weight, std=0.02) if module.bias is not None: nn.init.zeros_(module.bias) ``` ### Training Loop ```python from torch.cuda.amp import autocast, GradScaler from tqdm import tqdm def train_epoch( model: nn.Module, dataloader: DataLoader, optimizer: Optimizer, scheduler: LRScheduler, scaler: GradScaler, device: torch.device, accumulation_steps: int = 4, ) -> float: """Entrena una época completa.""" model.train() total_loss = 0.0 for step, batch in enumerate(tqdm(dataloader)): batch = {k: v.to(device) for k, v in batch.items()} with autocast(dtype=torch.bfloat16): outputs = model(**batch) loss = outputs.loss / accumulation_steps scaler.scale(loss).backward() if (step + 1) % accumulation_steps == 0: scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) scaler.step(optimizer) scaler.update() scheduler.step() optimizer.zero_grad() total_loss += loss.item() * accumulation_steps return total_loss / len(dataloader) ``` ### Checkpoint Saving/Loading ```python def save_checkpoint( model: nn.Module, optimizer: Optimizer, scheduler: LRScheduler, epoch: int, loss: float, path: str, ) -> None: """Guarda checkpoint completo.""" torch.save({ "model_state_dict": model.state_dict(), "optimizer_state_dict": optimizer.state_dict(), "scheduler_state_dict": scheduler.state_dict(), "epoch": epoch, "loss": loss, }, path) def load_checkpoint( path: str, model: nn.Module, optimizer: Optional[Optimizer] = None, scheduler: Optional[LRScheduler] = None, ) -> Dict: """Carga checkpoint.""" checkpoint = torch.load(path, map_location="cpu") model.load_state_dict(checkpoint["model_state_dict"]) if optimizer: optimizer.load_state_dict(checkpoint["optimizer_state_dict"]) if scheduler: scheduler.load_state_dict(checkpoint["scheduler_state_dict"]) return checkpoint ``` ## Memory Optimization ### Gradient Checkpointing ```python # Para modelos grandes (>500M params) model.gradient_checkpointing_enable() # Manual control: from torch.utils.checkpoint import checkpoint class Block(nn.Module): def forward(self, x): if self.training and self.gradient_checkpointing: return checkpoint(self._forward_impl, x, use_reentrant=False) return self._forward_impl(x) ``` ### Efficient Attention ```python # Usar Flash Attention cuando sea posible from torch.nn.functional import scaled_dot_product_attention # O xformers para backwards compatibility try: from xformers.ops import memory_efficient_attention HAS_XFORMERS = True except ImportError: HAS_XFORMERS = False ``` ### Tensor Operations ```python # BIEN: operaciones in-place cuando sea seguro x.add_(bias) # En lugar de x = x + bias # BIEN: evitar concatenaciones innecesarias # MAL: # outputs = [] # for block in self.blocks: # outputs.append(block(x)) # return torch.cat(outputs, dim=-1) # BIEN: usar stack si las dimensiones son iguales outputs = torch.stack([block(x) for block in self.blocks], dim=0) ``` ## Testing (pytest) ```python import pytest import torch from pampar.coder.v2.modelo import PampaRCoderV2 from pampar.coder.v2.config import ConfigPampaRCoderV2 @pytest.fixture def config(): """Configuración pequeña para tests.""" return ConfigPampaRCoderV2( vocab_size=1000, hidden_size=64, num_layers=2, num_heads=4, ) @pytest.fixture def model(config): """Modelo pequeño para tests.""" return PampaRCoderV2(config) class TestPampaRCoderV2: def test_forward_shape(self, model, config): """Verifica output shape.""" batch_size, seq_len = 2, 16 input_ids = torch.randint(0, config.vocab_size, (batch_size, seq_len)) output = model(input_ids) assert output.logits.shape == (batch_size, seq_len, config.vocab_size) def test_gradient_flow(self, model): """Verifica que gradientes fluyen correctamente.""" input_ids = torch.randint(0, 1000, (1, 8)) labels = input_ids.clone() output = model(input_ids, labels=labels) output.loss.backward() for name, param in model.named_parameters(): if param.requires_grad: assert param.grad is not None, f"No gradient for {name}" ``` ## Logging ```python import logging # Configurar al inicio del script logging.basicConfig( format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO, ) logger = logging.getLogger(__name__) # Usar en el código logger.info(f"Epoch {epoch}: loss={loss:.4f}") logger.warning(f"GPU memory high: {memory_used:.1f}GB") logger.error(f"Checkpoint save failed: {e}") ``` ``` ```