| |
| """ |
| 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 |
|
|
| |
| 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) |
| |
| 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() |
|
|
| |
| 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() |
|
|
| |
| 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() |
|
|
| |
| em_total += (pred.lower() == ref.lower()) |
| f1_total += token_f1(pred, ref) |
|
|
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| hallucination_total += compute_hallucination_proxy(pred, ref, prompt) |
| coherence_total += compute_coherence_score(pred) |
|
|
| |
| 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() |
|
|
