| 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" |
| ) -> 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 = {} |
| |
| |
| 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 |
| ) |
| |
| |
| |
| |
| |
| 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 |
|
|
| |
| if task_type == "sql" and test_cases: |
| pass |
|
|
| results[name] = cat_metrics |
|
|
| return results |
|
|
