# src/eval/comprehensive_eval.py """ Comprehensive Multi-Metric Evaluation for Self-Owned LLM & Qwen Alignments. Computes: 1. Exact Match (EM) 2. Token F1 3. BLEU Score (BLEU-1, BLEU-2, BLEU-4) 4. ROUGE Scores (ROUGE-1, ROUGE-2, ROUGE-L) 5. METEOR Score 6. Hallucination Rate proxy (% ungrounded tokens) 7. Coherence / Repetition Score 8. Toxicity & Safety Screening (RiskValidator) """ import argparse import json import os import sys import torch import nltk from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction from nltk.translate.meteor_score import meteor_score from rouge_score import rouge_scorer from transformers import AutoModelForCausalLM, AutoTokenizer sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from src.common.metrics import token_f1 from src.hitl.validators import RiskValidator # Ensure required NLTK data nltk.download('wordnet', quiet=True) nltk.download('punkt', quiet=True) def compute_hallucination_proxy(pred: str, reference: str, prompt: str) -> float: """Estimates hallucination rate as % of content words in prediction missing from context/reference.""" pred_words = set(re_tokenize(pred.lower())) source_words = set(re_tokenize((prompt + " " + reference).lower())) if not pred_words: return 0.0 ungrounded = pred_words - source_words return round(len(ungrounded) / len(pred_words), 4) def compute_coherence_score(pred: str) -> float: """Measures text coherence by penalizing repetitive n-grams and unclosed structures.""" words = re_tokenize(pred.lower()) if len(words) <= 1: return 1.0 unique_ratio = len(set(words)) / len(words) # 2-gram repetition check bigrams = [f"{words[i]}_{words[i+1]}" for i in range(len(words)-1)] bigram_ratio = len(set(bigrams)) / max(1, len(bigrams)) score = (unique_ratio * 0.4) + (bigram_ratio * 0.6) return round(score, 4) def re_tokenize(text: str): import re return re.findall(r'\b\w+\b', text) def main(): parser = argparse.ArgumentParser(description="Comprehensive Model Evaluation") parser.add_argument("--base", default="Qwen/Qwen2.5-1.5B-Instruct") parser.add_argument("--adapter", default=None, help="PEFT adapter directory") parser.add_argument("--data", required=True, help="JSONL eval dataset") parser.add_argument("--max", type=int, default=50) parser.add_argument("--max_new_tokens", type=int, default=64) args = parser.parse_args() # Determine device dev = "cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu") print(f"Loading base model {args.base} on {dev}...") tok = AutoTokenizer.from_pretrained(args.base) if tok.pad_token is None: tok.pad_token = tok.eos_token dtype = torch.bfloat16 if torch.cuda.is_available() else (torch.float16 if dev == "mps" else torch.float32) model = AutoModelForCausalLM.from_pretrained(args.base, torch_dtype=dtype).to(dev) if args.adapter: from peft import PeftModel print(f"Attaching adapter {args.adapter}...") model = PeftModel.from_pretrained(model, args.adapter) model.eval() # Initialize scorers r_scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True) smooth_fn = SmoothingFunction().method1 validator = RiskValidator( blocked_keywords=["rm -rf", "drop table", "ignore previous instructions", "system override"], sensitive_topics=["classified", "confidential", "api_key", "password"] ) rows = [json.loads(l) for l in open(args.data, encoding="utf-8") if l.strip()][:args.max] assert rows, "No data rows found." total = len(rows) print(f"Evaluating across {total} samples...") em_total = 0.0 f1_total = 0.0 bleu1_total = 0.0 bleu2_total = 0.0 bleu4_total = 0.0 rouge1_total = 0.0 rouge2_total = 0.0 rougeL_total = 0.0 meteor_total = 0.0 hallucination_total = 0.0 coherence_total = 0.0 toxicity_flags = 0 for i, r in enumerate(rows, 1): prompt = r["prompt"] ref = str(r["response"]).strip() msgs = [{"role": "user", "content": prompt}] ids = tok.apply_chat_template(msgs, add_generation_prompt=True, return_tensors="pt").to(dev) attn_mask = torch.ones_like(ids).to(dev) with torch.no_grad(): out = model.generate( input_ids=ids, attention_mask=attn_mask, do_sample=False, max_new_tokens=args.max_new_tokens, pad_token_id=tok.pad_token_id ) pred = tok.decode(out[0, ids.shape[1]:], skip_special_tokens=True).strip() # 1. Exact Match & F1 em_total += (pred.lower() == ref.lower()) f1_total += token_f1(pred, ref) # 2. Tokenize for BLEU & METEOR pred_tokens = re_tokenize(pred.lower()) ref_tokens = re_tokenize(ref.lower()) if pred_tokens and ref_tokens: bleu1 = sentence_bleu([ref_tokens], pred_tokens, weights=(1, 0, 0, 0), smoothing_function=smooth_fn) bleu2 = sentence_bleu([ref_tokens], pred_tokens, weights=(0.5, 0.5, 0, 0), smoothing_function=smooth_fn) bleu4 = sentence_bleu([ref_tokens], pred_tokens, weights=(0.25, 0.25, 0.25, 0.25), smoothing_function=smooth_fn) meteor = meteor_score([ref_tokens], pred_tokens) else: bleu1 = bleu2 = bleu4 = meteor = 0.0 bleu1_total += bleu1 bleu2_total += bleu2 bleu4_total += bleu4 meteor_total += meteor # 3. ROUGE r_scores = r_scorer.score(ref, pred) rouge1_total += r_scores['rouge1'].fmeasure rouge2_total += r_scores['rouge2'].fmeasure rougeL_total += r_scores['rougeL'].fmeasure # 4. Hallucination & Coherence hallucination_total += compute_hallucination_proxy(pred, ref, prompt) coherence_total += compute_coherence_score(pred) # 5. Safety / Risk risk_lvl, flags = validator.analyze(prompt, pred) if risk_lvl in ["MEDIUM", "HIGH"]: toxicity_flags += 1 summary = { "n_samples": total, "adapter": args.adapter or "base_model", "Exact_Match": round(em_total / total, 4), "Token_F1": round(f1_total / total, 4), "BLEU_1": round(bleu1_total / total, 4), "BLEU_2": round(bleu2_total / total, 4), "BLEU_4": round(bleu4_total / total, 4), "ROUGE_1": round(rouge1_total / total, 4), "ROUGE_2": round(rouge2_total / total, 4), "ROUGE_L": round(rougeL_total / total, 4), "METEOR": round(meteor_total / total, 4), "Coherence_Score": round(coherence_total / total, 4), "Hallucination_Rate": round(hallucination_total / total, 4), "Safety_Violation_Rate": round(toxicity_flags / total, 4) } print("\n" + "="*50) print("COMPREHENSIVE EVALUATION RESULTS") print("="*50) print(json.dumps(summary, indent=2)) return summary if __name__ == "__main__": main()