Update evaluator.py

evaluator.py

@@ -206,174 +206,85 @@
 
 # evaluator.py
 """
-Upgraded Evaluation logic for the Agentic Evaluation Framework.
-Provides scoring functions, visualization generation, and summary outputs.
+Evaluator for Agentic Evaluation Framework
 """
 
-import math
-import uuid
-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
 
-# -----------------------------
-# Lazy model loading
-# -----------------------------
-NLI_MODEL = "textattack/roberta-base-MNLI"
+from sentence_transformers import SentenceTransformer, util
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+
+# ------------------------
+# Models (lightweight)
+# ------------------------
 EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
+NLI_MODEL = "textattack/roberta-base-MNLI"
 
-_nli_tokenizer = None
-_nli_model = None
-_embed_model = None
-_id2label = None
-
-def ensure_models_loaded():
-    global _nli_tokenizer, _nli_model, _embed_model, _id2label
-    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:
+_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):
     if not prompt or not response:
         return 0.0
-    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)
+    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: str, response: str) -> float:
+def check_hallucination(prompt, response):
     if not prompt or not response:
         return 0.0
-    tokenizer, model, id2label = get_nli_tokenizer_and_model()
-    inputs = tokenizer.encode_plus(prompt, response, return_tensors="pt", truncation=True)
-    outputs = model(**inputs)
+    inputs = _nli_tokenizer.encode_plus(prompt, response, return_tensors="pt", truncation=True)
+    outputs = _nli_model(**inputs)
     probs = outputs.logits.softmax(dim=1).detach().cpu().numpy()[0]
-    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(reference: str, response: str) -> float:
+    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:
         return 0.0
-    embed_model = get_embed_model()
-    ref_emb = embed_model.encode(reference, convert_to_tensor=True)
-    r_emb = embed_model.encode(response, convert_to_tensor=True)
+    ref_emb = _embed_model.encode(reference, 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)
 
-def check_coherence(response: str) -> float:
-    if not response or not isinstance(response, str):
+def check_coherence(response):
+    if not response:
         return 0.0
-    sentences = [s.strip() for s in response.split(".") if s.strip()]
-    if not sentences:
-        return 0.0
-    lengths = [len(s.split()) for s in sentences]
-    avg_len = np.mean(lengths)
-    std = np.std(lengths)
-    score = max(0.0, min(1.0, 1.0 - (std / (avg_len + 1e-6))))
-    return round(score, 3)
-
-def check_fluency(response: str) -> float:
-    if not response or not isinstance(response, str):
+    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_fluency(response):
+    if not response:
         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)
-
-# -----------------------------
-# 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 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 task_agent_heatmap(leaderboard: pd.DataFrame, metric: str):
-    pivot = leaderboard.pivot(index="task", columns="agent", values=metric)
-    fig, ax = plt.subplots(figsize=(7, 5))
-    sns.heatmap(pivot, annot=True, fmt=".2f", cmap="YlGnBu", ax=ax)
-    ax.set_title(f"Task-Agent Performance ({metric})")
-    return fig
-
-def leaderboard_barplot(leaderboard: pd.DataFrame, metric_cols: List[str]):
-    melted = leaderboard.melt(id_vars=["agent"], value_vars=metric_cols, var_name="metric", value_name="score")
-    fig, ax = plt.subplots(figsize=(8, 5))
-    sns.barplot(x="metric", y="score", hue="agent", data=melted, ax=ax)
-    ax.set_title("Leaderboard Bar Chart")
-    ax.set_ylim(0, 1)
-    return fig
-
-def distribution_plot(metrics_df: pd.DataFrame, metric: str):
-    fig, ax = plt.subplots(figsize=(7, 5))
-    sns.boxplot(x="agent", y=metric, data=metrics_df, ax=ax)
-    sns.stripplot(x="agent", y=metric, data=metrics_df, ax=ax, color="black", alpha=0.4, jitter=True)
-    ax.set_title(f"Distribution of {metric} Scores per Agent")
-    ax.set_ylim(0, 1)
-    return fig
-
-def scatter_two_metrics(metrics_df: pd.DataFrame, metric_x: str, metric_y: str):
-    fig, ax = plt.subplots(figsize=(6, 6))
-    sns.scatterplot(x=metric_x, y=metric_y, hue="agent", data=metrics_df, ax=ax, alpha=0.7)
-    ax.set_title(f"{metric_x} vs {metric_y}")
-    ax.set_xlim(0, 1)
-    ax.set_ylim(0, 1)
-    return fig
+    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()
-
-    # compute scores per row
+def evaluate_dataframe(df: pd.DataFrame):
     scores = []
     for _, row in df.iterrows():
         s = {}
@@ -382,79 +293,85 @@ def evaluate_dataframe(df: pd.DataFrame) -> Tuple[pd.DataFrame, List[Tuple[str,s
         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
+        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)
         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"]
 
-    # leaderboard: average per agent & task
-    metric_cols = ["instruction_following", "hallucination", "accuracy", "coherence", "fluency"]
     leaderboard = (
-        metrics_df.groupby(["agent", "task"])[metric_cols]
+        metrics_df.groupby(["agent", "task_type"])[metric_cols]
         .mean()
         .reset_index()
     )
+    return metrics_df, [], leaderboard
 
-    # -------------------
-    # Visualization images
-    # -------------------
-    images = []
+# ------------------------
+# Visualizations
+# ------------------------
 
-    try:
-        rows = []
-        for agent in leaderboard["agent"].unique():
-            vals = leaderboard[leaderboard["agent"] == agent][metric_cols].mean().tolist()
-            rows.append({"name": agent, "values": vals})
-        fig1 = spider_net_multi(metric_cols, 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 plot failed:", e)
+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
 
-    try:
-        fig2 = heatmap_plot(metrics_df, metric_cols, 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)
+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
+
+def generate_visualizations(metrics_df, leaderboard):
+    metric_cols = ["instruction_following", "hallucination", "accuracy", "coherence", "fluency", "final_score"]
+    figs = []
     try:
-        fig3 = task_agent_heatmap(leaderboard, "accuracy")
-        path3 = f"/tmp/task_agent_{uuid.uuid4().hex}.png"
-        fig3.savefig(path3, bbox_inches="tight")
-        plt.close(fig3)
-        images.append((path3, "Task-Agent Heatmap (Accuracy)"))
+        figs.append(plot_radar_chart(leaderboard, metric_cols))
     except Exception as e:
-        print("Task-agent heatmap failed:", e)
-
+        print("Radar failed:", e)
     try:
-        fig4 = leaderboard_barplot(leaderboard, metric_cols)
-        path4 = f"/tmp/barplot_{uuid.uuid4().hex}.png"
-        fig4.savefig(path4, bbox_inches="tight")
-        plt.close(fig4)
-        images.append((path4, "Leaderboard Bar Chart"))
+        figs.append(plot_heatmap(metrics_df, metric_cols))
     except Exception as e:
-        print("Barplot failed:", e)
-
+        print("Heatmap failed:", e)
     try:
-        fig5 = distribution_plot(metrics_df, "accuracy")
-        path5 = f"/tmp/distribution_{uuid.uuid4().hex}.png"
-        fig5.savefig(path5, bbox_inches="tight")
-        plt.close(fig5)
-        images.append((path5, "Accuracy Distribution"))
+        figs.append(plot_boxplot(metrics_df, metric_cols))
     except Exception as e:
-        print("Distribution plot failed:", e)
-
+        print("Boxplot failed:", e)
     try:
-        fig6 = scatter_two_metrics(metrics_df, "instruction_following", "accuracy")
-        path6 = f"/tmp/scatter_{uuid.uuid4().hex}.png"
-        fig6.savefig(path6, bbox_inches="tight")
-        plt.close(fig6)
-        images.append((path6, "Instruction Following vs Accuracy"))
+        figs.append(plot_bar(leaderboard, metric_cols))
     except Exception as e:
-        print("Scatter plot failed:", 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