Update evaluator.py

evaluator.py

@@ -205,204 +205,235 @@
 ###############################################################################################################################
 
 # evaluator.py
-"""
-Evaluator for Agentic Evaluation Framework
-"""
-
+import math
+import uuid
+import json
+import re
+from typing import List, Dict, Tuple
 import numpy as np
 import pandas as pd
 import matplotlib.pyplot as plt
 import seaborn as sns
-import math, uuid, re
-
-from sentence_transformers import SentenceTransformer, util
-from transformers import AutoTokenizer, AutoModelForSequenceClassification
 
-# ------------------------
-# Models (lightweight)
-# ------------------------
-EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
+# -----------------------------
+# Lazy model loading
+# -----------------------------
 NLI_MODEL = "textattack/roberta-base-MNLI"
+EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
 
-_embed_model = SentenceTransformer(EMBED_MODEL)
-_nli_tokenizer = AutoTokenizer.from_pretrained(NLI_MODEL)
-_nli_model = AutoModelForSequenceClassification.from_pretrained(NLI_MODEL)
-_id2label = {int(k): v.upper() for k, v in _nli_model.config.id2label.items()}
-
-# ------------------------
-# Metrics
-# ------------------------
-
-def check_instruction_following(prompt, response):
+_nli_tokenizer = None
+_nli_model = None
+_embed_model = None
+_id2label = None
+
+def ensure_models_loaded():
+    global _nli_tokenizer, _nli_model, _embed_model, _id2label, util
+    if _embed_model is None:
+        from sentence_transformers import SentenceTransformer, util
+        _embed_model = SentenceTransformer(EMBED_MODEL)
+        globals()["util"] = util
+    if _nli_model is None:
+        from transformers import AutoTokenizer, AutoModelForSequenceClassification
+        _nli_tokenizer = AutoTokenizer.from_pretrained(NLI_MODEL)
+        _nli_model = AutoModelForSequenceClassification.from_pretrained(NLI_MODEL)
+        _nli_model.to("cpu")
+        _nli_model.eval()
+        _id2label = {int(k): v.upper() for k, v in _nli_model.config.id2label.items()}
+
+def get_embed_model():
+    ensure_models_loaded()
+    return _embed_model
+
+def get_nli_tokenizer_and_model():
+    ensure_models_loaded()
+    return _nli_tokenizer, _nli_model, _id2label
+
+
+# -----------------------------
+# Metric functions
+# -----------------------------
+
+def check_instruction_following(prompt: str, response: str) -> float:
     if not prompt or not response:
         return 0.0
-    p_emb = _embed_model.encode(prompt, convert_to_tensor=True)
-    r_emb = _embed_model.encode(response, convert_to_tensor=True)
+    embed_model = get_embed_model()
+    p_emb = embed_model.encode(prompt, convert_to_tensor=True)
+    r_emb = embed_model.encode(response, convert_to_tensor=True)
     sim = float(util.cos_sim(p_emb, r_emb).item())
     return round(max(0.0, min(1.0, sim)), 3)
 
-def check_hallucination(prompt, response):
+def check_hallucination(prompt: str, response: str) -> float:
     if not prompt or not response:
         return 0.0
-    inputs = _nli_tokenizer.encode_plus(prompt, response, return_tensors="pt", truncation=True)
-    outputs = _nli_model(**inputs)
+    tokenizer, model, id2label = get_nli_tokenizer_and_model()
+    inputs = tokenizer.encode_plus(prompt, response, return_tensors="pt", truncation=True)
+    outputs = model(**inputs)
     probs = outputs.logits.softmax(dim=1).detach().cpu().numpy()[0]
-    labels = [ _id2label[i] for i in range(len(probs)) ]
-    entailment_prob = float(probs[labels.index("ENTAILMENT")]) if "ENTAILMENT" in labels else float(probs.max())
-    return round(max(0.0, min(1.0, entailment_prob)), 3)
-
-def check_accuracy(reference, response):
-    if not reference or not response:
+    labels = [id2label[i] for i in range(len(probs))]
+    if "ENTAILMENT" in labels:
+        entailment_prob = float(probs[labels.index("ENTAILMENT")])
+    else:
+        entailment_prob = float(probs.max())
+    return round(entailment_prob, 3)
+
+def check_accuracy(prompt: str, response: str, metadata: str) -> float:
+    """
+    Accuracy:
+      - if metadata contains an 'answer', compare response vs answer using embeddings
+      - else fall back to entailment probability (NLI)
+    """
+    answer = None
+    if isinstance(metadata, str) and metadata.strip().startswith("{"):
+        try:
+            meta = json.loads(metadata)
+            for k in ("answer", "reference", "ground_truth"):
+                if k in meta:
+                    answer = meta[k]
+                    break
+        except Exception:
+            pass
+    elif isinstance(metadata, dict):
+        for k in ("answer", "reference", "ground_truth"):
+            if k in metadata:
+                answer = metadata[k]
+                break
+
+    if answer:
+        embed_model = get_embed_model()
+        ref_emb = embed_model.encode(str(answer), convert_to_tensor=True)
+        r_emb = embed_model.encode(response, convert_to_tensor=True)
+        sim = float(util.cos_sim(ref_emb, r_emb).item())
+        return round(max(0.0, min(1.0, sim)), 3)
+    else:
+        return check_hallucination(prompt, response)
+
+def check_coherence(response: str) -> float:
+    """
+    Coherence: average pairwise sentence similarity if >1 sentence,
+    otherwise fallback heuristic.
+    """
+    if not response or not isinstance(response, str) or response.strip() == "":
+        return 0.25
+    sents = [s.strip() for s in re.split(r"[.!?]+", response) if s.strip()]
+    embed_model = get_embed_model()
+    if len(sents) > 1:
+        try:
+            embs = embed_model.encode(sents, convert_to_tensor=True)
+            sims = []
+            for i in range(len(embs)):
+                for j in range(i + 1, len(embs)):
+                    sims.append(float(util.cos_sim(embs[i], embs[j]).item()))
+            if sims:
+                score = (np.mean(sims) + 1.0) / 2.0
+                return round(max(0.0, min(1.0, score)), 3)
+        except Exception:
+            pass
+    words = len(re.findall(r"\w+", response))
+    return round(min(0.8, max(0.3, words / 50.0)), 3)
+
+def check_fluency(response: str) -> float:
+    if not response or not isinstance(response, str):
         return 0.0
+    letters = sum(ch.isalpha() for ch in response)
+    total = len(response)
+    return round(max(0.0, min(1.0, letters / max(1, total))), 3)
 
-    # Compute BERTScore
-    try:
-        bert_results = bertscore.compute(predictions=[response], references=[reference], lang="en")
-        bert_f1 = bert_results["f1"][0]
-    except Exception:
-        bert_f1 = 0.0
-
-    # Compute BLEU
-    try:
-        bleu_results = bleu.compute(predictions=[response], references=[[reference.split()]])
-        bleu_score = bleu_results["bleu"]
-    except Exception:
-        bleu_score = 0.0
-
-    # Compute ROUGE-L
-    try:
-        rouge_results = rouge.compute(predictions=[response], references=[reference])
-        rouge_l = rouge_results["rougeL"]
-    except Exception:
-        rouge_l = 0.0
 
-    return float((bert_f1 + bleu_score + rouge_l) / 3)
+# -----------------------------
+# Visualization helpers
+# -----------------------------
 
+def spider_net_multi(labels: List[str], rows: List[Dict], title: str, fill_alpha: float = 0.12):
+    N = len(labels)
+    angles = [n / float(N) * 2 * math.pi for n in range(N)]
+    angles += angles[:1]
+    fig = plt.figure(figsize=(6.5, 6.5))
+    ax = plt.subplot(111, polar=True)
+    ax.set_xticks(angles[:-1])
+    ax.set_xticklabels(labels, fontsize=9)
+    ax.set_ylim(0, 1)
+    ax.set_yticks([0, 0.25, 0.5, 0.75, 1.0])
+    for r in rows:
+        values = r["values"]
+        values_closed = values + values[:1]
+        ax.plot(angles, values_closed, linewidth=1.5, label=r["name"])
+        ax.fill(angles, values_closed, alpha=fill_alpha)
+    ax.set_title(title, y=1.08, fontsize=12)
+    ax.legend(loc="upper right", bbox_to_anchor=(1.25, 1.1))
+    return fig
 
-# def check_coherence(response):
-#     if not response:
-#         return 0.0
-#     sents = [s.strip() for s in re.split(r"[.!?]+", response) if s.strip()]
-#     if len(sents) <= 1:
-#         return 1.0
-#     embs = _embed_model.encode(sents, convert_to_tensor=True)
-#     sims = []
-#     for i in range(len(embs)):
-#         for j in range(i+1, len(embs)):
-#             sims.append(float(util.cos_sim(embs[i], embs[j]).item()))
-#     avg = np.mean(sims)
-#     return round((avg + 1) / 2, 3)  # normalize to [0,1]
-def check_coherence(response, reference=None):
-    try:
-        model = get_embed_model()
-        emb = model.encode(response, convert_to_tensor=True, normalize_embeddings=True)
-        if reference:
-            ref_emb = model.encode(reference, convert_to_tensor=True, normalize_embeddings=True)
-            return float(torch.nn.functional.cosine_similarity(emb, ref_emb, dim=0))
-        else:
-            return float(torch.norm(emb))  # or another meaningful scalar
-    except Exception:
-        return 0.0
+def heatmap_plot(df: pd.DataFrame, metric_cols: List[str], title: str = "Metric Correlations"):
+    fig, ax = plt.subplots(figsize=(7, 5))
+    sns.heatmap(df[metric_cols].corr(), annot=True, fmt=".2f", cmap="coolwarm", ax=ax)
+    ax.set_title(title)
+    return fig
 
 
-def check_fluency(response):
-    if not response:
-        return 0.0
-    letters = sum(ch.isalpha() for ch in response)
-    total = len(response)
-    return round(letters / max(1, total), 3)
+# -----------------------------
+# Main evaluation entrypoint
+# -----------------------------
 
-# ------------------------
-# Evaluation
-# ------------------------
+def evaluate_dataframe(df: pd.DataFrame) -> Tuple[pd.DataFrame, List[Tuple[str,str]], pd.DataFrame]:
+    df = df.copy()
 
-def evaluate_dataframe(df: pd.DataFrame):
     scores = []
-    for _, row in df.iterrows():
+    for _, r in df.iterrows():
+        prompt = str(r.get("prompt", "") or "")
+        response = str(r.get("response", "") or "")
+        metadata = r.get("metadata", "")
+
         s = {}
-        s["instruction_following"] = check_instruction_following(str(row.get("prompt", "")), str(row.get("response", "")))
-        s["hallucination"] = check_hallucination(str(row.get("prompt", "")), str(row.get("response", "")))
-        s["accuracy"] = check_accuracy(str(row.get("reference", "")), str(row.get("response", "")))
-        s["coherence"] = check_coherence(str(row.get("response", "")))
-        s["fluency"] = check_fluency(str(row.get("response", "")))
-        # clamp
+        s["instruction_following"] = check_instruction_following(prompt, response)
+        s["hallucination"] = check_hallucination(prompt, response)
+        s["accuracy"] = check_accuracy(prompt, response, metadata)
+        s["coherence"] = check_coherence(response)
+        s["fluency"] = check_fluency(response)
+
+        # clamp and nanmean
         for k in s:
-            s[k] = max(0.0, min(1.0, s[k]))
-        s["final_score"] = round(float(np.mean(list(s.values()))), 3)
+            try:
+                s[k] = float(s[k])
+            except Exception:
+                s[k] = np.nan
+            if not np.isnan(s[k]):
+                s[k] = max(0.0, min(1.0, s[k]))
+
+        vals = [v for v in s.values() if not np.isnan(v)]
+        s["FinalScore"] = round(float(np.nanmean(vals)) if vals else 0.0, 3)
         scores.append(s)
 
     metrics_df = pd.concat([df.reset_index(drop=True), pd.DataFrame(scores)], axis=1)
-    metric_cols = ["instruction_following", "hallucination", "accuracy", "coherence", "fluency", "final_score"]
 
+    metric_cols = ["instruction_following", "hallucination", "accuracy", "coherence", "fluency", "FinalScore"]
     leaderboard = (
         metrics_df.groupby(["agent", "task_type"])[metric_cols]
         .mean()
         .reset_index()
     )
-    return metrics_df, [], leaderboard
-
-# ------------------------
-# Visualizations
-# ------------------------
-
-def plot_radar_chart(leaderboard, metric_cols):
-    categories = metric_cols
-    angles = np.linspace(0, 2*np.pi, len(categories), endpoint=False).tolist()
-    angles += angles[:1]
-    fig = plt.figure(figsize=(6,6))
-    ax = plt.subplot(111, polar=True)
-    for agent in leaderboard["agent"].unique():
-        vals = leaderboard[leaderboard["agent"]==agent][metric_cols].mean().tolist()
-        vals += vals[:1]
-        ax.plot(angles, vals, label=agent)
-        ax.fill(angles, vals, alpha=0.1)
-    ax.set_xticks(angles[:-1])
-    ax.set_xticklabels(categories)
-    ax.set_ylim(0,1)
-    ax.legend(loc="upper right")
-    return fig
-
-def plot_heatmap(metrics_df, metric_cols):
-    fig, ax = plt.subplots(figsize=(7,5))
-    sns.heatmap(metrics_df[metric_cols].corr(), annot=True, fmt=".2f", cmap="coolwarm", ax=ax)
-    return fig
 
-# def plot_boxplot(metrics_df, metric_cols):
-#     fig, ax = plt.subplots(figsize=(7,5))
-#     sns.boxplot(data=metrics_df[metric_cols], ax=ax)
-#     return fig
-
-def plot_bar(leaderboard, metric_cols):
-    fig, ax = plt.subplots(figsize=(8,5))
-    leaderboard.plot(x="agent", y="final_score", kind="bar", ax=ax, legend=False)
-    ax.set_ylabel("Final Score")
-    return fig
+    images = []
 
-def generate_visualizations(metrics_df, leaderboard):
-    metric_cols = ["instruction_following", "hallucination", "accuracy", "coherence", "fluency", "final_score"]
-    figs = []
     try:
-        figs.append(plot_radar_chart(leaderboard, metric_cols))
+        rows = []
+        for agent in leaderboard["agent"].unique():
+            vals = leaderboard[leaderboard["agent"] == agent][metric_cols[:-1]].mean().tolist()
+            rows.append({"name": agent, "values": vals})
+        fig1 = spider_net_multi(metric_cols[:-1], rows, "Agent Performance Radar")
+        path1 = f"/tmp/radar_{uuid.uuid4().hex}.png"
+        fig1.savefig(path1, bbox_inches="tight")
+        plt.close(fig1)
+        images.append((path1, "Radar Plot"))
     except Exception as e:
-        print("Radar failed:", e)
+        print("Radar plot failed:", e)
+
     try:
-        figs.append(plot_heatmap(metrics_df, metric_cols))
+        fig2 = heatmap_plot(metrics_df, metric_cols[:-1], title="Metric Correlation Heatmap")
+        path2 = f"/tmp/heatmap_{uuid.uuid4().hex}.png"
+        fig2.savefig(path2, bbox_inches="tight")
+        plt.close(fig2)
+        images.append((path2, "Correlation Heatmap"))
     except Exception as e:
         print("Heatmap failed:", e)
-    # try:
-    #     figs.append(plot_boxplot(metrics_df, metric_cols))
-    # except Exception as e:
-    #     print("Boxplot failed:", e)
-    try:
-        figs.append(plot_bar(leaderboard, metric_cols))
-    except Exception as e:
-        print("Bar failed:", e)
 
-    # Save to temp and return as gallery list
-    images = []
-    for fig in figs:
-        path = f"/tmp/viz_{uuid.uuid4().hex}.png"
-        fig.savefig(path, bbox_inches="tight")
-        plt.close(fig)
-        images.append(path)
-    return images
+    return metrics_df, images, leaderboard
+