#!/usr/bin/env python3 # train_retrieval.py — with detailed logging """ Train the dense retrieval model using the custom Contrastive Learning pipeline defined in src.embeddings. Dataset: data/finfact_raw_truefalse.csv - claim : the claim text (anchor) - evidence : Python-list string of evidence sentences (positives) Goal: Given a claim, the retriever should rank its supporting evidence sentences higher than evidence from other claims. """ import argparse import ast import os import re import time import pandas as pd import torch from loguru import logger from torch.utils.data import DataLoader try: from src.retrieval.embeddings import ( SENTENCE_TRANSFORMERS_AVAILABLE, TORCH_AVAILABLE, ContrastiveEmbeddingModel, CryptoEmbeddingTrainer, RetrievalDataset, ) except ImportError as e: raise ImportError(f"Could not import embedding modules: {e}") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def parse_evidence_list(raw: str) -> list[str]: """ Parse the evidence column which may be stored as: 1. A Python list literal: ['sentence 1', 'sentence 2', ...] 2. A plain multi-sentence string separated by period / newline Returns a de-duplicated list of non-empty strings (min 10 chars). """ raw = str(raw).strip() sentences = [] # Try ast.literal_eval first (handles proper Python list literals) try: parsed = ast.literal_eval(raw) if isinstance(parsed, list): sentences = [str(s).strip() for s in parsed] else: sentences = [str(parsed).strip()] except (ValueError, SyntaxError): # Fallback: split on newlines or double-spaces parts = re.split(r"\n{1,}|\.\s{2,}", raw) sentences = [p.strip() for p in parts] # Filter short / empty strings sentences = [s for s in sentences if len(s) >= 10] # Remove duplicates while preserving order seen = set() unique = [] for s in sentences: key = s.lower() if key not in seen: seen.add(key) unique.append(s) return unique def load_finfact_dataset(csv_path: str): """ Load finfact_raw_truefalse.csv and return list of {'claim': str, 'evidences': List[str]} dicts. Only rows with at least one valid evidence sentence are kept. """ df = pd.read_csv(csv_path) required = {"claim", "evidence"} missing = required - set(df.columns) if missing: raise ValueError( f"CSV is missing required columns: {missing}. Found: {df.columns.tolist()}" ) records = [] skipped = 0 total_rows = len(df) logger.info(f"Parsing {total_rows} rows from CSV...") for idx, (_, row) in enumerate(df.iterrows()): claim = str(row["claim"]).strip() if not claim or claim.lower() == "nan": skipped += 1 continue if pd.isna(row["evidence"]): logger.debug(f" Row {idx}: skipped (NaN evidence)") skipped += 1 continue evs = parse_evidence_list(row["evidence"]) if not evs: logger.debug(f" Row {idx}: skipped (no valid evidence sentences)") skipped += 1 continue records.append({"claim": claim, "evidences": evs}) if (idx + 1) % 500 == 0: logger.info( f" Parsed {idx + 1}/{total_rows} rows | kept={len(records)} skipped={skipped}" ) all_evs = [ev for r in records for ev in r["evidences"]] ev_counts = [len(r["evidences"]) for r in records] logger.info( f"\n{'=' * 50}\n" f" CSV loaded: {csv_path}\n" f" Total rows : {total_rows}\n" f" Kept records : {len(records)}\n" f" Skipped : {skipped}\n" f" Total sentences : {len(all_evs)}\n" f" Avg ev/claim : {len(all_evs) / max(1, len(records)):.1f}\n" f" Min ev/claim : {min(ev_counts) if ev_counts else 0}\n" f" Max ev/claim : {max(ev_counts) if ev_counts else 0}\n" f"{'=' * 50}" ) return records def collate_triplets(batch): """ Custom collate for RetrievalDataset. Each item: {'anchor': str, 'positive': str, 'negatives': List[str]} """ anchors = [item["anchor"] for item in batch] positives = [item["positive"] for item in batch] negatives = [item["negatives"] for item in batch] # List[List[str]] return {"anchor": anchors, "positive": positives, "negatives": negatives} # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): if not TORCH_AVAILABLE or not SENTENCE_TRANSFORMERS_AVAILABLE: raise RuntimeError( "PyTorch and SentenceTransformers are required for training." ) parser = argparse.ArgumentParser( description="Fine-tune Retrieval Model on finfact_raw_truefalse.csv" ) parser.add_argument( "--csv", type=str, default="data/finfact_raw_truefalse.csv", help="Path to training CSV (claim/evidence columns)", ) parser.add_argument( "--model_name", type=str, default="bge-vi-base", help="Base SentenceTransformer model", ) parser.add_argument( "--batch_size", type=int, default=16, help="Training batch size" ) parser.add_argument( "--max_length", type=int, default=512, help="Maximum sequence length for tokenizer", ) parser.add_argument( "--epochs", type=int, default=5, help="Number of training epochs" ) parser.add_argument( "--learning_rate", type=float, default=2e-4, help="Learning rate" ) parser.add_argument( "--num_triplets", type=int, default=8000, help="Number of (anchor, positive, negatives) triplets to sample per epoch", ) parser.add_argument( "--num_negatives", type=int, default=3, help="Negatives per anchor" ) parser.add_argument( "--output_dir", type=str, default="models/retriever_model", help="Directory to save the trained model", ) device = "cuda" if torch.cuda.is_available() else "cpu" parser.add_argument("--device", type=str, default=device, help="Device (cuda/cpu)") args = parser.parse_args() # 1. Load data t0 = time.time() logger.info(f"Loading dataset from {args.csv}") records = load_finfact_dataset(args.csv) logger.info(f"Data loading took {time.time() - t0:.1f}s") if len(records) < 2: logger.error("Need at least 2 records to build negatives. Aborting.") return logger.info( f"Device: {args.device} | epochs={args.epochs} | batch={args.batch_size} | triplets/epoch={args.num_triplets}" ) # 2. Initialize model logger.info(f"Initializing ContrastiveEmbeddingModel ({args.model_name})...") model = ContrastiveEmbeddingModel( base_model_name=args.model_name, lambda_reg=0.001, freeze_base=True, max_length=args.max_length, encoder_device="cpu", encode_batch_size=args.batch_size, ) model = model.to(args.device) # 3. Build dataset logger.info( f"Building RetrievalDataset with {args.num_triplets} triplets " f"and {args.num_negatives} negatives each..." ) dataset = RetrievalDataset( records=records, num_triplets=args.num_triplets, num_negatives=args.num_negatives, ) dataloader = DataLoader( dataset, batch_size=args.batch_size, shuffle=True, collate_fn=collate_triplets, num_workers=0, pin_memory=False, persistent_workers=False, ) # 4. Initialize trainer trainer = CryptoEmbeddingTrainer( model=model, learning_rate=args.learning_rate, batch_size=args.batch_size, device=args.device, ) # 5. Training loop logger.info(f"\n{'=' * 50}") logger.info( f"Starting training: {args.epochs} epochs | {args.num_triplets} triplets/epoch | device={args.device}" ) logger.info(f"{'=' * 50}") train_start = time.time() history = [] for epoch in range(args.epochs): epoch_start = time.time() logger.info(f"\n--- Epoch {epoch + 1}/{args.epochs} ---") avg_loss = trainer.train_epoch(dataloader) epoch_time = time.time() - epoch_start if epoch == args.epochs - 1: logger.info("Running final evaluation...") metrics = trainer.evaluate(dataloader) logger.info( f"Epoch {epoch + 1}/{args.epochs} | Loss: {avg_loss:.4f} | " f"Time: {epoch_time:.1f}s | " f"Pos-Neg Gap: {metrics['pos_neg_gap']:.4f} | " f"Separation: {metrics['separation_rate']:.4f} | " f"Pos sim: {metrics['mean_pos_similarity']:.4f} | " f"Neg sim: {metrics['mean_neg_similarity']:.4f}" ) history.append( { **{"epoch": epoch + 1, "loss": avg_loss, "time": epoch_time}, **metrics, } ) else: logger.info( f"Epoch {epoch + 1}/{args.epochs} | Loss: {avg_loss:.4f} | Time: {epoch_time:.1f}s" ) history.append({"epoch": epoch + 1, "loss": avg_loss, "time": epoch_time}) total_time = time.time() - train_start logger.info(f"\n{'=' * 50}") logger.info( f"Training complete | Total time: {total_time:.1f}s ({total_time / 60:.1f} min)" ) logger.info("Loss history: " + " -> ".join(f"{h['loss']:.4f}" for h in history)) logger.info(f"{'=' * 50}") # 6. Save model os.makedirs(args.output_dir, exist_ok=True) if hasattr(model, "encoder") and model.encoder is not None: logger.info(f"Saving fine-tuned SentenceTransformer to {args.output_dir}") model.encoder.save(args.output_dir) torch.save( model.projection.state_dict(), os.path.join(args.output_dir, "custom_projection.pt"), ) logger.info("Training completed successfully. Model saved.") else: logger.error("Model encoder not found. Cannot save.") if __name__ == "__main__": main()