Upload .\src\evaluation\metrics.py with huggingface_hub

.//src//evaluation//metrics.py

@@ -0,0 +1,289 @@
+"""Evaluation metrics for BwengeAi."""
+
+import json
+import logging
+import math
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+class BwengeEvaluator:
+    """Evaluation manager for BwengeAi."""
+
+    def __init__(self, config: dict[str, Any]):
+        self.config = config
+        self.eval_config = config.get("evaluation", {})
+        self.metrics = self.eval_config.get("metrics", ["perplexity", "rouge", "bleu", "accuracy"])
+
+    def compute_perplexity(
+        self,
+        model: Any,
+        tokenizer: Any,
+        texts: list[str],
+        max_length: int = 512,
+    ) -> dict[str, float]:
+        """Compute perplexity on a set of texts."""
+        import torch
+        
+        model.eval()
+        total_loss = 0.0
+        total_tokens = 0
+        
+        for text in texts:
+            inputs = tokenizer(
+                text,
+                return_tensors="pt",
+                truncation=True,
+                max_length=max_length,
+                padding=True,
+            ).to(model.device)
+            
+            with torch.no_grad():
+                outputs = model(**inputs, labels=inputs["input_ids"])
+                loss = outputs.loss
+            
+            num_tokens = inputs["input_ids"].numel()
+            total_loss += loss.item() * num_tokens
+            total_tokens += num_tokens
+        
+        avg_loss = total_loss / total_tokens if total_tokens > 0 else float("inf")
+        perplexity = np.exp(avg_loss)
+        
+        return {"perplexity": float(perplexity), "avg_loss": float(avg_loss)}
+
+    def compute_rouge(
+        self,
+        predictions: list[str],
+        references: list[str],
+    ) -> dict[str, float]:
+        """Compute ROUGE scores (simplified implementation)."""
+        
+        def get_ngrams(text: str, n: int) -> list[tuple[str, ...]]:
+            words = text.lower().split()
+            return [tuple(words[i:i + n]) for i in range(len(words) - n + 1)]
+        
+        def compute_rouge_n(pred: str, ref: str, n: int) -> dict[str, float]:
+            pred_ngrams = get_ngrams(pred, n)
+            ref_ngrams = get_ngrams(ref, n)
+            
+            if not pred_ngrams or not ref_ngrams:
+                return {"precision": 0.0, "recall": 0.0, "fmeasure": 0.0}
+            
+            pred_counts = {}
+            for ng in pred_ngrams:
+                pred_counts[ng] = pred_counts.get(ng, 0) + 1
+            
+            ref_counts = {}
+            for ng in ref_ngrams:
+                ref_counts[ng] = ref_counts.get(ng, 0) + 1
+            
+            overlap = 0
+            for ng, count in pred_counts.items():
+                overlap += min(count, ref_counts.get(ng, 0))
+            
+            precision = overlap / len(pred_ngrams) if pred_ngrams else 0.0
+            recall = overlap / len(ref_ngrams) if ref_ngrams else 0.0
+            fmeasure = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0
+            
+            return {"precision": precision, "recall": recall, "fmeasure": fmeasure}
+        
+        scores = {
+            "rouge1": {"precision": [], "recall": [], "fmeasure": []},
+            "rouge2": {"precision": [], "recall": [], "fmeasure": []},
+            "rougeL": {"precision": [], "recall": [], "fmeasure": []},
+        }
+        
+        for pred, ref in zip(predictions, references):
+            result1 = compute_rouge_n(pred, ref, 1)
+            result2 = compute_rouge_n(pred, ref, 2)
+            
+            words_pred = pred.lower().split()
+            words_ref = ref.lower().split()
+            lcs_len = self._lcs_length(words_pred, words_ref)
+            precision_lcs = lcs_len / len(words_pred) if words_pred else 0.0
+            recall_lcs = lcs_len / len(words_ref) if words_ref else 0.0
+            fmeasure_lcs = 2 * precision_lcs * recall_lcs / (precision_lcs + recall_lcs) if (precision_lcs + recall_lcs) > 0 else 0.0
+            resultL = {"precision": precision_lcs, "recall": recall_lcs, "fmeasure": fmeasure_lcs}
+            
+            for metric, result in [("rouge1", result1), ("rouge2", result2), ("rougeL", resultL)]:
+                scores[metric]["precision"].append(result["precision"])
+                scores[metric]["recall"].append(result["recall"])
+                scores[metric]["fmeasure"].append(result["fmeasure"])
+        
+        averaged = {}
+        for metric in scores:
+            averaged[f"{metric}_precision"] = float(np.mean(scores[metric]["precision"]))
+            averaged[f"{metric}_recall"] = float(np.mean(scores[metric]["recall"]))
+            averaged[f"{metric}_fmeasure"] = float(np.mean(scores[metric]["fmeasure"]))
+        
+        return averaged
+    
+    def _lcs_length(self, x: list[str], y: list[str]) -> int:
+        """Compute length of longest common subsequence."""
+        m, n = len(x), len(y)
+        dp = [[0] * (n + 1) for _ in range(m + 1)]
+        for i in range(1, m + 1):
+            for j in range(1, n + 1):
+                if x[i - 1] == y[j - 1]:
+                    dp[i][j] = dp[i - 1][j - 1] + 1
+                else:
+                    dp[i][j] = max(dp[i - 1][j], dp[i][j - 1])
+        return dp[m][n]
+
+    def compute_bleu(
+        self,
+        predictions: list[str],
+        references: list[str],
+    ) -> dict[str, float]:
+        """Compute BLEU scores (simplified implementation)."""
+        
+        def compute_bleu_single(pred: list[str], refs: list[list[str]], max_order: int = 4) -> float:
+            clipped_counts = [0] * max_order
+            total_counts = [0] * max_order
+            
+            for order in range(1, max_order + 1):
+                pred_ngrams = {}
+                for i in range(len(pred) - order + 1):
+                    ngram = tuple(pred[i:i + order])
+                    pred_ngrams[ngram] = pred_ngrams.get(ngram, 0) + 1
+                
+                max_ref_count = 0
+                for ref in refs:
+                    ref_ngrams = {}
+                    for i in range(len(ref) - order + 1):
+                        ngram = tuple(ref[i:i + order])
+                        ref_ngrams[ngram] = ref_ngrams.get(ngram, 0) + 1
+                    
+                    for ngram, count in pred_ngrams.items():
+                        max_ref_count = max(max_ref_count, min(count, ref_ngrams.get(ngram, 0)))
+                
+                clipped_counts[order - 1] += max_ref_count
+                total_counts[order - 1] += max(1, len(pred) - order + 1)
+            
+            p_scores = []
+            for i in range(max_order):
+                if total_counts[i] > 0:
+                    p_scores.append(clipped_counts[i] / total_counts[i])
+                else:
+                    p_scores.append(0.0)
+            
+            if min(p_scores) > 0:
+                p_avg = sum(p_scores) / len(p_scores)
+                penalty = math.exp(1 - max_order / max(1, len(pred))) if len(pred) < max_order else 1.0
+                return penalty * p_avg
+            else:
+                return 0.0
+        
+        scores = {"bleu_1": [], "bleu_2": [], "bleu_3": [], "bleu_4": []}
+        
+        for pred, ref in zip(predictions, references):
+            pred_tokens = pred.split()
+            ref_tokens = [ref.split()]
+            
+            for order in range(1, 5):
+                bleu = compute_bleu_single(pred_tokens, ref_tokens, max_order=order)
+                scores[f"bleu_{order}"].append(bleu)
+        
+        averaged = {}
+        for metric in scores:
+            averaged[metric] = float(np.mean(scores[metric])) if scores[metric] else 0.0
+        
+        return averaged
+
+    def compute_accuracy(
+        self,
+        predictions: list[str],
+        references: list[str],
+    ) -> dict[str, float]:
+        """Compute accuracy metrics."""
+        exact_matches = sum(1 for p, r in zip(predictions, references) if p.strip() == r.strip())
+        accuracy = exact_matches / len(predictions) if predictions else 0.0
+        
+        return {"accuracy": accuracy, "exact_matches": exact_matches}
+
+    def evaluate_generation(
+        self,
+        model: Any,
+        tokenizer: Any,
+        test_data: list[dict[str, str]],
+        max_new_tokens: int = 256,
+    ) -> dict[str, Any]:
+        """Evaluate model generation quality."""
+        predictions = []
+        references = []
+        
+        for item in test_data:
+            prompt = item.get("instruction", item.get("prompt", item.get("question", "")))
+            reference = item.get("output", item.get("response", item.get("answer", "")))
+            
+            if not prompt or not reference:
+                continue
+            
+            inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
+            
+            import torch
+            with torch.no_grad():
+                outputs = model.generate(
+                    **inputs,
+                    max_new_tokens=max_new_tokens,
+                    temperature=0.7,
+                    top_p=0.9,
+                    do_sample=True,
+                    pad_token_id=tokenizer.eos_token_id,
+                )
+            
+            generated = tokenizer.decode(outputs[0], skip_special_tokens=True)
+            prediction = generated[len(prompt):]
+            
+            predictions.append(prediction)
+            references.append(reference)
+        
+        results = {}
+        
+        if "rouge" in self.metrics:
+            results.update(self.compute_rouge(predictions, references))
+        
+        if "bleu" in self.metrics:
+            results.update(self.compute_bleu(predictions, references))
+        
+        if "accuracy" in self.metrics:
+            results.update(self.compute_accuracy(predictions, references))
+        
+        results["num_samples"] = len(predictions)
+        
+        return results
+
+    def evaluate_model(
+        self,
+        model: Any,
+        tokenizer: Any,
+        eval_data_path: str,
+        output_dir: str | None = None,
+    ) -> dict[str, Any]:
+        """Run full evaluation."""
+        logger.info("Starting evaluation...")
+        
+        eval_data = []
+        with open(eval_data_path, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line:
+                    eval_data.append(json.loads(line))
+        
+        max_samples = self.eval_config.get("max_eval_samples", 1000)
+        eval_data = eval_data[:max_samples]
+        
+        results = self.evaluate_generation(model, tokenizer, eval_data)
+        
+        if output_dir:
+            output_path = Path(output_dir) / "eval_results.json"
+            with open(output_path, "w", encoding="utf-8") as f:
+                json.dump(results, f, indent=2, ensure_ascii=False)
+            logger.info(f"Results saved to {output_path}")
+        
+        logger.info(f"Evaluation results: {results}")
+        return results