Spaces:
Paused
Paused
Download src/train.py from VSRAJHAVEL/BioGraph-AI: direct link, hf CLI and curl.
- Browser
- Download file 14.7 kB
-
https://huggingface.co/spaces/VSRAJHAVEL/BioGraph-AI/resolve/main/src/train.py
- Command line
-
hf download hf://spaces/VSRAJHAVEL/BioGraph-AI/src/train.py
-
curl -L -o train.py https://huggingface.co/spaces/VSRAJHAVEL/BioGraph-AI/resolve/main/src/train.py
14.7 kB
| """ | |
| Training and Validation Pipeline for GNNs and Classical Baselines. | |
| Includes CPU thread limits, masked loss, early stopping, LR scheduling, | |
| metrics computation, and multi-seed benchmarking. | |
| """ | |
| import os | |
| import sys | |
| # Ensure project root is in sys.path | |
| sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) | |
| import time | |
| import json | |
| import random | |
| import logging | |
| import argparse | |
| from typing import Dict, List, Tuple, Any, Optional | |
| import yaml | |
| import numpy as np | |
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| from torch.optim import Adam | |
| from torch.optim.lr_scheduler import ReduceLROnPlateau | |
| from torch_geometric.loader import DataLoader | |
| from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score, roc_auc_score, average_precision_score | |
| from src.data import get_dataloaders, NODE_DIM, EDGE_DIM | |
| from src.models.gnn import build_gnn_model, BaseGNN | |
| from src.models.baseline import ClassicalFingerprintBaseline | |
| # Enforce 4 CPU threads matching hardware physical cores | |
| torch.set_num_threads(4) | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") | |
| logger = logging.getLogger(__name__) | |
| def set_seed(seed: int = 42): | |
| """Set random seed for reproducibility.""" | |
| random.seed(seed) | |
| np.random.seed(seed) | |
| torch.manual_seed(seed) | |
| if torch.cuda.is_available(): | |
| torch.cuda.manual_seed_all(seed) | |
| def compute_loss(preds: torch.Tensor, targets: torch.Tensor, task_type: str) -> torch.Tensor: | |
| """Compute task loss with NaN masking for missing multi-task targets.""" | |
| mask = ~torch.isnan(targets) | |
| if mask.sum() == 0: | |
| return torch.tensor(0.0, requires_grad=True, device=preds.device) | |
| if task_type == 'regression': | |
| return F.mse_loss(preds[mask], targets[mask]) | |
| else: | |
| # Binary or multi-task classification | |
| return F.binary_cross_entropy_with_logits(preds[mask], targets[mask]) | |
| def evaluate_gnn( | |
| model: nn.Module, | |
| loader: DataLoader, | |
| task_type: str, | |
| device: torch.device | |
| ) -> Tuple[Dict[str, float], np.ndarray, np.ndarray]: | |
| """ | |
| Evaluate GNN model on a DataLoader. | |
| Returns: | |
| metrics: dictionary of evaluation metrics | |
| all_preds: numpy array of predictions | |
| all_targets: numpy array of targets | |
| """ | |
| model.eval() | |
| preds_list = [] | |
| targets_list = [] | |
| with torch.no_grad(): | |
| for batch in loader: | |
| batch = batch.to(device) | |
| out = model(batch.x, batch.edge_index, edge_attr=batch.edge_attr, batch=batch.batch) | |
| if task_type != 'regression': | |
| out = torch.sigmoid(out) | |
| preds_list.append(out.cpu().numpy()) | |
| targets_list.append(batch.y.cpu().numpy()) | |
| preds = np.concatenate(preds_list, axis=0) if preds_list else np.empty((0, 1)) | |
| targets = np.concatenate(targets_list, axis=0) if targets_list else np.empty((0, 1)) | |
| num_tasks = targets.shape[1] if targets.ndim > 1 else 1 | |
| metrics: Dict[str, float] = {} | |
| if task_type == 'regression': | |
| rmse_list, mae_list, r2_list = [], [], [] | |
| for t in range(num_tasks): | |
| yt = targets[:, t] | |
| yp = preds[:, t] | |
| mask = ~np.isnan(yt) | |
| if np.sum(mask) > 1: | |
| rmse_list.append(np.sqrt(mean_squared_error(yt[mask], yp[mask]))) | |
| mae_list.append(mean_absolute_error(yt[mask], yp[mask])) | |
| r2_list.append(r2_score(yt[mask], yp[mask])) | |
| metrics['rmse'] = float(np.mean(rmse_list)) if rmse_list else 0.0 | |
| metrics['mae'] = float(np.mean(mae_list)) if mae_list else 0.0 | |
| metrics['r2'] = float(np.mean(r2_list)) if r2_list else 0.0 | |
| else: | |
| roc_list, pr_list = [], [] | |
| for t in range(num_tasks): | |
| yt = targets[:, t] | |
| yp = preds[:, t] | |
| mask = ~np.isnan(yt) | |
| if len(np.unique(yt[mask])) > 1: | |
| roc_list.append(roc_auc_score(yt[mask], yp[mask])) | |
| pr_list.append(average_precision_score(yt[mask], yp[mask])) | |
| metrics['roc_auc'] = float(np.mean(roc_list)) if roc_list else 0.5 | |
| metrics['pr_auc'] = float(np.mean(pr_list)) if pr_list else 0.0 | |
| return metrics, preds, targets | |
| def train_single_seed( | |
| model_name: str, | |
| config: Dict[str, Any], | |
| seed: int = 42, | |
| quick: bool = False, | |
| override_epochs: Optional[int] = None, | |
| save_dir: str = 'results/checkpoints' | |
| ) -> Dict[str, Any]: | |
| """Train a single model architecture on a specific random seed.""" | |
| set_seed(seed) | |
| device = torch.device('cpu') | |
| os.makedirs(save_dir, exist_ok=True) | |
| os.makedirs('results/history', exist_ok=True) | |
| dataset_name = config['dataset'] | |
| task_type = config['task_type'] | |
| num_tasks = config['num_tasks'] | |
| primary_metric = config['metric'] | |
| higher_is_better = config.get('higher_is_better', False) | |
| train_cfg = config.get('training', {}) | |
| model_cfg = config.get('model', {}) | |
| batch_size = train_cfg.get('batch_size', 32) | |
| max_epochs = override_epochs or (2 if quick else train_cfg.get('epochs', 50)) | |
| lr = train_cfg.get('lr', 0.001) | |
| weight_decay = train_cfg.get('weight_decay', 1e-4) | |
| patience = train_cfg.get('patience', 15) | |
| grad_clip = train_cfg.get('grad_clip', 1.0) | |
| # 1. Classical Baseline Branch | |
| if model_name.lower() == 'baseline': | |
| logger.info(f"Training Classical Baseline (Random Forest) on {dataset_name.upper()} (Seed={seed})...") | |
| _, _, _, meta = get_dataloaders(dataset_name, batch_size=batch_size, quick=quick) | |
| baseline = ClassicalFingerprintBaseline( | |
| task_type=task_type, | |
| num_tasks=num_tasks, | |
| n_estimators=20 if quick else 100, | |
| random_state=seed | |
| ) | |
| t0 = time.time() | |
| baseline.fit(meta['train']) | |
| train_time = time.time() - t0 | |
| val_metrics = baseline.evaluate(meta['val']) | |
| test_metrics = baseline.evaluate(meta['test']) | |
| ckpt_path = os.path.join(save_dir, f"{dataset_name}_baseline_best.pkl") | |
| baseline.save(ckpt_path) | |
| result = { | |
| 'model_name': 'baseline', | |
| 'dataset_name': dataset_name, | |
| 'seed': seed, | |
| 'train_time': train_time, | |
| 'val_metrics': val_metrics, | |
| 'test_metrics': test_metrics, | |
| 'checkpoint_path': ckpt_path | |
| } | |
| return result | |
| # 2. GNN Training Branch | |
| train_loader, val_loader, test_loader, meta = get_dataloaders(dataset_name, batch_size=batch_size, quick=quick) | |
| model = build_gnn_model( | |
| model_name=model_name, | |
| in_channels=NODE_DIM, | |
| edge_dim=EDGE_DIM, | |
| num_tasks=num_tasks, | |
| hidden_dim=model_cfg.get('hidden_dim', 64), | |
| num_layers=model_cfg.get('num_layers', 3), | |
| dropout=model_cfg.get('dropout', 0.2), | |
| residual=model_cfg.get('residual', True) | |
| ).to(device) | |
| optimizer = Adam(model.parameters(), lr=lr, weight_decay=weight_decay) | |
| scheduler = ReduceLROnPlateau( | |
| optimizer, | |
| mode='max' if higher_is_better else 'min', | |
| factor=train_cfg.get('lr_factor', 0.5), | |
| patience=train_cfg.get('lr_patience', 5), | |
| min_lr=1e-5 | |
| ) | |
| best_val_score = -float('inf') if higher_is_better else float('inf') | |
| best_epoch = 0 | |
| patience_counter = 0 | |
| best_model_state = None | |
| history = {'train_loss': [], 'val_metric': [], 'epoch_times': []} | |
| t0 = time.time() | |
| for epoch in range(1, max_epochs + 1): | |
| ep_t0 = time.time() | |
| model.train() | |
| total_loss = 0.0 | |
| num_batches = 0 | |
| for batch in train_loader: | |
| batch = batch.to(device) | |
| optimizer.zero_grad() | |
| out = model(batch.x, batch.edge_index, edge_attr=batch.edge_attr, batch=batch.batch) | |
| loss = compute_loss(out, batch.y, task_type) | |
| loss.backward() | |
| if grad_clip > 0: | |
| torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=grad_clip) | |
| optimizer.step() | |
| total_loss += loss.item() | |
| num_batches += 1 | |
| avg_loss = total_loss / max(num_batches, 1) | |
| val_metrics, _, _ = evaluate_gnn(model, val_loader, task_type, device) | |
| val_score = val_metrics[primary_metric] | |
| scheduler.step(val_score) | |
| ep_time = time.time() - ep_t0 | |
| history['train_loss'].append(avg_loss) | |
| history['val_metric'].append(val_score) | |
| history['epoch_times'].append(ep_time) | |
| # Check improvement | |
| is_improved = (val_score > best_val_score) if higher_is_better else (val_score < best_val_score) | |
| if is_improved: | |
| best_val_score = val_score | |
| best_epoch = epoch | |
| patience_counter = 0 | |
| best_model_state = {k: v.cpu().clone() for k, v in model.state_dict().items()} | |
| else: | |
| patience_counter += 1 | |
| if epoch % 5 == 0 or epoch == 1 or epoch == max_epochs: | |
| logger.info(f"[{model_name.upper()}|Epoch {epoch:02d}/{max_epochs:02d}] Loss: {avg_loss:.4f} | Val {primary_metric.upper()}: {val_score:.4f} (Best: {best_val_score:.4f}) | Time: {ep_time:.2f}s") | |
| if patience_counter >= patience and not quick: | |
| logger.info(f"Early stopping triggered at epoch {epoch}. Best epoch was {best_epoch}.") | |
| break | |
| total_train_time = time.time() - t0 | |
| # Load best model for test evaluation | |
| if best_model_state is not None: | |
| model.load_state_dict(best_model_state) | |
| val_metrics, _, _ = evaluate_gnn(model, val_loader, task_type, device) | |
| test_metrics, _, _ = evaluate_gnn(model, test_loader, task_type, device) | |
| # Save checkpoint | |
| best_ckpt_path = os.path.join(save_dir, f"{dataset_name}_{model_name.lower()}_best.pt") | |
| ckpt_payload = { | |
| 'model_name': model_name.lower(), | |
| 'dataset_name': dataset_name, | |
| 'model_state_dict': model.state_dict(), | |
| 'config': config, | |
| 'val_metrics': val_metrics, | |
| 'test_metrics': test_metrics, | |
| 'node_dim': NODE_DIM, | |
| 'edge_dim': EDGE_DIM, | |
| 'num_tasks': num_tasks, | |
| 'model_kwargs': { | |
| 'in_channels': NODE_DIM, | |
| 'edge_dim': EDGE_DIM, | |
| 'num_tasks': num_tasks, | |
| 'hidden_dim': model_cfg.get('hidden_dim', 64), | |
| 'num_layers': model_cfg.get('num_layers', 3), | |
| 'dropout': model_cfg.get('dropout', 0.2), | |
| 'residual': model_cfg.get('residual', True) | |
| } | |
| } | |
| torch.save(ckpt_payload, best_ckpt_path) | |
| logger.info(f"Saved best checkpoint to {best_ckpt_path} (Test {primary_metric.upper()}: {test_metrics[primary_metric]:.4f})") | |
| # Save history | |
| history_path = f"results/history/{dataset_name}_{model_name.lower()}_history.json" | |
| with open(history_path, 'w') as f: | |
| json.dump(history, f) | |
| return { | |
| 'model_name': model_name.lower(), | |
| 'dataset_name': dataset_name, | |
| 'seed': seed, | |
| 'train_time': total_train_time, | |
| 'best_epoch': best_epoch, | |
| 'val_metrics': val_metrics, | |
| 'test_metrics': test_metrics, | |
| 'checkpoint_path': best_ckpt_path | |
| } | |
| def train_dataset( | |
| dataset_name: str, | |
| models: List[str], | |
| config_path: Optional[str] = None, | |
| seeds: Optional[List[int]] = None, | |
| quick: bool = False, | |
| override_epochs: Optional[int] = None | |
| ) -> Dict[str, Any]: | |
| """Train multiple models on a given dataset across seeds, aggregating mean and std.""" | |
| if config_path is None: | |
| config_path = f"configs/{dataset_name.lower()}.yaml" | |
| with open(config_path, 'r') as f: | |
| config = yaml.safe_load(f) | |
| active_seeds = seeds or ([42] if quick else config.get('training', {}).get('seeds', [42, 123, 999])) | |
| results_summary: Dict[str, Any] = {} | |
| for model_name in models: | |
| seed_results = [] | |
| for seed in active_seeds: | |
| logger.info(f"\n==================================================") | |
| logger.info(f"Training {model_name.upper()} on {dataset_name.upper()} | Seed {seed}") | |
| logger.info(f"==================================================") | |
| res = train_single_seed( | |
| model_name=model_name, | |
| config=config, | |
| seed=seed, | |
| quick=quick, | |
| override_epochs=override_epochs | |
| ) | |
| seed_results.append(res) | |
| # Aggregate metrics across seeds | |
| metric_keys = seed_results[0]['test_metrics'].keys() | |
| agg_metrics = {} | |
| for k in metric_keys: | |
| vals = [r['test_metrics'][k] for r in seed_results] | |
| agg_metrics[k] = { | |
| 'mean': float(np.mean(vals)), | |
| 'std': float(np.std(vals)), | |
| 'values': vals | |
| } | |
| avg_time = float(np.mean([r['train_time'] for r in seed_results])) | |
| results_summary[model_name] = { | |
| 'aggregated_test_metrics': agg_metrics, | |
| 'mean_train_time_sec': avg_time, | |
| 'runs': seed_results | |
| } | |
| return results_summary | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser(description="Train GNNs and Baselines for Molecular Property Prediction") | |
| parser.add_argument("--config", type=str, default=None, help="Path to YAML configuration") | |
| parser.add_argument("--dataset", type=str, default="esol", help="Dataset name: esol, bbbp, tox21") | |
| parser.add_argument("--model", type=str, default="all", help="Model name: baseline, gcn, gat, gin, mpnn, or all") | |
| parser.add_argument("--quick", action="store_true", help="Run rapid smoke test (trimmed dataset, 2 epochs)") | |
| parser.add_argument("--epochs", type=int, default=None, help="Override number of training epochs") | |
| parser.add_argument("--seeds", type=int, nargs="+", default=None, help="Override seeds, e.g., --seeds 42") | |
| args = parser.parse_args() | |
| cfg_path = args.config if args.config is not None else f"configs/{args.dataset.lower()}.yaml" | |
| all_models = ['baseline', 'gcn', 'gat', 'gin', 'mpnn'] | |
| models_to_train = all_models if args.model.lower() == 'all' else [args.model.lower()] | |
| summary = train_dataset( | |
| dataset_name=args.dataset, | |
| models=models_to_train, | |
| config_path=cfg_path, | |
| seeds=args.seeds, | |
| quick=args.quick, | |
| override_epochs=args.epochs | |
| ) | |
| print("\n" + "="*50) | |
| print(f"BENCHMARK SUMMARY FOR {args.dataset.upper()}") | |
| print("="*50) | |
| for m, d in summary.items(): | |
| metrics_str = ", ".join([f"{k.upper()}: {v['mean']:.4f} ± {v['std']:.4f}" for k, v in d['aggregated_test_metrics'].items()]) | |
| print(f"{m.upper():<10} | {metrics_str} | Avg Time: {d['mean_train_time_sec']:.1f}s") | |