File size: 2,909 Bytes
3581a58 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | import logging
from pathlib import Path
from typing import List, Dict, Any, Optional
from evaluation.metrics import EvaluationMetrics
logger = logging.getLogger(__name__)
def _default_bertscore_model() -> str:
local_dir = Path("models/codebert-base")
if local_dir.is_dir() and (local_dir / "config.json").exists():
return str(local_dir.resolve())
return "microsoft/codebert-base"
class ModelComparator:
"""
Compares generation outputs across 4 categories:
1. Base_Model (Small Code LM)
2. LoRA_Model (Fine-tuned)
3. RAG_Pipeline (Retrieval Augmented)
4. LLM_Baseline (OpenAI/Gemini API)
"""
def __init__(self, device: str = "cpu", bertscore_model: Optional[str] = None):
self.device = device
self.bertscore_model = bertscore_model or _default_bertscore_model()
logger.info(f"ModelComparator using BERTScore/CodeBERTScore backbone: {self.bertscore_model}")
self.metrics = EvaluationMetrics()
def compare(
self,
references: List[str],
base_preds: List[str],
lora_preds: List[str],
rag_preds: Optional[List[str]] = None,
llm_preds: Optional[List[str]] = None,
test_cases: Optional[List[str]] = None,
task_type: str = "sql" # "sql" or "doc"
) -> Dict[str, Dict[str, float]]:
results = {}
categories = {
"Base_Model": base_preds,
"LoRA_Model": lora_preds
}
if rag_preds: categories["RAG_Pipeline"] = rag_preds
if llm_preds: categories["LLM_Baseline"] = llm_preds
for name, preds in categories.items():
logger.info(f"Calculating metrics for {name}...")
cat_metrics = {}
# Text-based metrics (BLEU, ROUGE, BERTScore)
cat_metrics["BLEU"] = self.metrics.compute_bleu(references, preds)
rouge_scores = self.metrics.compute_rouge(references, preds)
cat_metrics.update(rouge_scores)
cat_metrics["BERTScore"] = self.metrics.compute_bertscore(
references, preds, device=self.device, model_type=self.bertscore_model
)
# Exact & Response Accuracy
# exact_matches = [self.metrics.compute_exact_match(r, p) for r, p in zip(references, preds)]
# cat_metrics["Exact_Match_Accuracy"] = round(sum(exact_matches) / len(exact_matches), 4) if exact_matches else 0.0
resp_acc = [self.metrics.compute_response_accuracy(r, p) for r, p in zip(references, preds)]
cat_metrics["Response_Accuracy"] = round(sum(resp_acc) / len(resp_acc), 4) if resp_acc else 0.0
# SQL Execution Accuracy (if task is SQL and test cases/DBs are provided)
if task_type == "sql" and test_cases:
pass
results[name] = cat_metrics
return results
|