Upload evaluate.py

evaluate.py

@@ -0,0 +1,438 @@
+# evaluate.py
+
+import os
+import json
+import time
+import re  # <-- ADD THIS IMPORT
+import pandas as pd
+from typing import List, Dict, Any
+from pathlib import Path
+
+# --- Imports from the main application ---
+# In evaluate.py
+
+try:
+    from alz_companion.agent import (
+        make_rag_chain, route_query_type, detect_tags_from_query,
+        answer_query, call_llm, build_or_load_vectorstore
+    )
+    from alz_companion.prompts import FAITHFULNESS_JUDGE_PROMPT
+    from langchain_community.vectorstores import FAISS
+    # --- Also move this import inside the try block for consistency ---
+    from langchain.schema import Document
+
+except ImportError:
+    # --- START: FALLBACK DEFINITIONS ---
+    class FAISS:
+        def __init__(self): self.docstore = type('obj', (object,), {'_dict': {}})()
+        def add_documents(self, docs): pass
+        def save_local(self, path): pass
+        @classmethod
+        def from_documents(cls, docs, embeddings=None): return cls()
+
+    class Document:
+        def __init__(self, page_content, metadata=None):
+            self.page_content = page_content
+            self.metadata = metadata or {}
+
+    def make_rag_chain(*args, **kwargs): return lambda q, **k: {"answer": f"(Eval Fallback) You asked: {q}", "sources": []}
+    def route_query_type(q, **kwargs): return "general_conversation"
+    def detect_tags_from_query(*args, **kwargs): return {}
+    def answer_query(chain, q, **kwargs): return chain(q, **kwargs)
+    def call_llm(*args, **kwargs): return "{}"
+    
+    # --- ADD FALLBACK DEFINITION FOR THE MISSING FUNCTION ---
+    def build_or_load_vectorstore(docs, index_path, is_personal=False):
+        return FAISS()
+    # --- END OF ADDITION ---
+
+    FAITHFULNESS_JUDGE_PROMPT = ""
+    print("WARNING: Could not import from alz_companion. Evaluation functions will use fallbacks.")
+    # --- END: FALLBACK DEFINITIONS ---
+    
+
+# --- LLM-as-a-Judge Prompt for Answer Correctness ---
+ANSWER_CORRECTNESS_JUDGE_PROMPT = """You are an expert evaluator. Your task is to assess the factual correctness of a generated answer against a ground truth answer.
+
+- GROUND_TRUTH_ANSWER: This is the gold-standard, correct answer.
+- GENERATED_ANSWER: This is the answer produced by the AI model.
+
+Evaluate if the GENERATED_ANSWER is factually aligned with the GROUND_TRUTH_ANSWER. Ignore minor differences in phrasing, tone, or structure. The key is factual accuracy.
+
+Respond with a single JSON object containing a float score from 0.0 to 1.0.
+- 1.0: The generated answer is factually correct and aligns perfectly with the ground truth.
+- 0.5: The generated answer is partially correct but misses key information or contains minor inaccuracies.
+- 0.0: The generated answer is factually incorrect or contradicts the ground truth.
+
+--- DATA TO EVALUATE ---
+GROUND_TRUTH_ANSWER:
+{ground_truth_answer}
+
+GENERATED_ANSWER:
+{generated_answer}
+---
+
+Return a single JSON object with your score:
+{{
+  "correctness_score": <float>
+}}
+"""
+
+test_fixtures = []
+
+def load_test_fixtures():
+    """Loads fixtures into the test_fixtures list."""
+    global test_fixtures
+    test_fixtures = []
+    env_path = os.environ.get("TEST_FIXTURES_PATH", "").strip()
+
+    # --- START: DEFINITIVE FIX ---
+    # The old code used a relative path, which is unreliable.
+    # This new code builds an absolute path to the fixture file based on
+    # the location of this (evaluate.py) script.
+    script_dir = Path(__file__).parent
+    default_fixture_file = script_dir / "small_test_cases_v10.jsonl"
+
+    candidates = [env_path] if env_path else [str(default_fixture_file)]
+    # --- END: DEFINITIVE FIX ---
+    # candidates = [env_path] if env_path else ["conversation_test_fixtures_v10.jsonl"]
+    # candidates = [env_path] if env_path else ["small_test_cases_v10.jsonl"]
+    
+    path = next((p for p in candidates if p and os.path.exists(p)), None)
+    if not path:
+        print("Warning: No test fixtures file found for evaluation.")
+        return
+    
+    # Use the corrected v10 file if available
+    # if "conversation_test_fixtures_v10.jsonl" in path:
+    if "small_test_cases_v10.jsonl" in path:
+        print(f"Using corrected test fixtures: {path}")
+
+    with open(path, "r", encoding="utf-8") as f:
+        for line in f:
+            try:
+                test_fixtures.append(json.loads(line))
+            except json.JSONDecodeError:
+                print(f"Skipping malformed JSON line in {path}")
+    print(f"Loaded {len(test_fixtures)} fixtures for evaluation from {path}")
+
+    
+def evaluate_nlu_tags(expected: Dict[str, Any], actual: Dict[str, Any], tag_key: str, expected_key_override: str = None) -> Dict[str, float]:
+    lookup_key = expected_key_override or tag_key
+    expected_raw = expected.get(lookup_key, [])
+    expected_set = set(expected_raw if isinstance(expected_raw, list) else [expected_raw]) if expected_raw and expected_raw != "None" else set()
+    actual_raw = actual.get(tag_key, [])
+    actual_set = set(actual_raw if isinstance(actual_raw, list) else [actual_raw]) if actual_raw and actual_raw != "None" else set()
+    if not expected_set and not actual_set:
+        return {"precision": 1.0, "recall": 1.0, "f1_score": 1.0}
+    true_positives = len(expected_set.intersection(actual_set))
+    precision = true_positives / len(actual_set) if actual_set else 0.0
+    recall = true_positives / len(expected_set) if expected_set else 0.0
+    f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0
+    return {"precision": precision, "recall": recall, "f1_score": f1_score}
+
+def _parse_judge_json(raw_str: str) -> dict | None:
+    try:
+        start_brace = raw_str.find('{')
+        end_brace = raw_str.rfind('}')
+        if start_brace != -1 and end_brace > start_brace:
+            json_str = raw_str[start_brace : end_brace + 1]
+            return json.loads(json_str)
+        return None
+    except (json.JSONDecodeError, AttributeError):
+        return None
+
+# --- NEW: helpers for categorisation and error-class labelling ---
+def _categorize_test(test_id: str) -> str:
+    tid = (test_id or "").lower()
+    if "synonym" in tid: return "synonym"
+    if "multi_fact" in tid or "multi-hop" in tid or "multihop" in tid: return "multi_fact"
+    if "omission" in tid: return "omission"
+    if "hallucination" in tid: return "hallucination"
+    if "time" in tid or "temporal" in tid: return "temporal"
+    if "context" in tid: return "context_disambig"
+    return "baseline"
+
+def _classify_error(gt: str, gen: str) -> str:
+    import re
+    gt = (gt or "").strip().lower()
+    gen = (gen or "").strip().lower()
+    if not gen:
+        return "empty"
+    if not gt:
+        return "hallucination" if gen else "empty"
+    if gt in gen:
+        return "paraphrase"
+    gt_tokens = set([t for t in re.split(r'\W+', gt) if t])
+    gen_tokens = set([t for t in re.split(r'\W+', gen) if t])
+    overlap = len(gt_tokens & gen_tokens) / max(1, len(gt_tokens))
+    if overlap >= 0.3:
+        return "omission"
+    return "contradiction"
+
+## NEW
+# In evaluate.py
+def run_comprehensive_evaluation(
+    vs_general: FAISS,
+    vs_personal: FAISS,
+    nlu_vectorstore: FAISS,
+    config: Dict[str, Any],
+    storage_path: Path  # <-- ADD THIS PARAMETER
+):
+    global test_fixtures
+    if not test_fixtures:
+        # The return signature is now back to 3 items.
+        return "No test fixtures loaded.", [], []
+
+    vs_personal_test = None
+    personal_context_docs = []
+    personal_context_file = "sample_data/1 Complaints of a Dutiful Daughter.txt"
+
+    if os.path.exists(personal_context_file):
+        print(f"Found personal context file for evaluation: '{personal_context_file}'")
+        with open(personal_context_file, "r", encoding="utf-8") as f:
+            content = f.read()
+            doc = Document(page_content=content, metadata={"source": os.path.basename(personal_context_file)})
+            personal_context_docs.append(doc)
+    else:
+        print(f"WARNING: Personal context file not found at '{personal_context_file}'. Factual tests will likely fail.")
+
+    vs_personal_test = build_or_load_vectorstore(
+        personal_context_docs,
+        index_path="tmp/eval_personal_index",
+        is_personal=True
+    )
+    print(f"Successfully created temporary personal vectorstore with {len(personal_context_docs)} document(s) for this evaluation run.")
+    
+    def _norm(label: str) -> str:
+        label = (label or "").strip().lower()
+        return "factual_question" if "factual" in label else label
+    
+    print("Starting comprehensive evaluation...")
+    results: List[Dict[str, Any]] = []
+    total_fixtures = len(test_fixtures)
+    print(f"\n🚀 STARTING EVALUATION on {total_fixtures} test cases...")
+
+    for i, fx in enumerate(test_fixtures):
+        test_id = fx.get("test_id", "N/A")
+        print(f"--- Processing Test Case {i+1}/{total_fixtures}: ID = {test_id} ---")
+        
+        turns = fx.get("turns") or []
+        api_chat_history = [{"role": t.get("role"), "content": t.get("text")} for t in turns]
+        query = next((t["content"] for t in reversed(api_chat_history) if (t.get("role") or "user").lower() == "user"), "")
+        if not query: continue
+
+        print(f'Query: "{query}"')
+    
+        ground_truth = fx.get("ground_truth", {})
+        expected_route = _norm(ground_truth.get("expected_route", "caregiving_scenario"))
+        expected_tags = ground_truth.get("expected_tags", {})
+        actual_route = _norm(route_query_type(query))
+        route_correct = (actual_route == expected_route)
+        
+        actual_tags: Dict[str, Any] = {}
+        if "caregiving_scenario" in actual_route:
+            actual_tags = detect_tags_from_query(
+                query, nlu_vectorstore=nlu_vectorstore,
+                behavior_options=config["behavior_tags"], emotion_options=config["emotion_tags"],
+                topic_options=config["topic_tags"], context_options=config["context_tags"],
+            )
+
+        behavior_metrics = evaluate_nlu_tags(expected_tags, actual_tags, "detected_behaviors")
+        emotion_metrics  = evaluate_nlu_tags(expected_tags, actual_tags, "detected_emotion")
+        topic_metrics    = evaluate_nlu_tags(expected_tags, actual_tags, "detected_topics")
+        context_metrics  = evaluate_nlu_tags(expected_tags, actual_tags, "detected_contexts")
+
+        final_tags = {}
+        if "caregiving_scenario" in actual_route:
+            final_tags = {
+                "scenario_tag": (actual_tags.get("detected_behaviors") or [None])[0],
+                "emotion_tag":  actual_tags.get("detected_emotion"),
+                "topic_tag": (actual_tags.get("detected_topics") or [None])[0],
+                "context_tags": actual_tags.get("detected_contexts", [])
+            }
+        
+        current_test_role = fx.get("test_role", "patient")
+        rag_chain = make_rag_chain(
+            vs_general, 
+            vs_personal, 
+            role=current_test_role, 
+            for_evaluation=True
+        )
+        
+        t0 = time.time()
+        response = answer_query(rag_chain, query, query_type=actual_route, chat_history=api_chat_history, **final_tags)
+        latency_ms = round((time.time() - t0) * 1000.0, 1)
+        answer_text = response.get("answer", "ERROR")
+        ground_truth_answer = ground_truth.get("ground_truth_answer")
+
+        category = _categorize_test(test_id)
+        error_class = _classify_error(ground_truth_answer, answer_text)
+        
+        expected_sources_set = set(map(str, ground_truth.get("expected_sources", [])))
+        raw_sources = response.get("sources", [])
+        actual_sources_set = set(map(str, raw_sources if isinstance(raw_sources, (list, tuple)) else [raw_sources]))
+
+        print("\n" + "-"*20 + " SOURCE EVALUATION " + "-"*20)
+        print(f"  - Expected: {sorted(list(expected_sources_set))}")
+        print(f"  - Actual:   {sorted(list(actual_sources_set))}")
+        
+        true_positives = expected_sources_set.intersection(actual_sources_set)
+        false_positives = actual_sources_set - expected_sources_set
+        false_negatives = expected_sources_set - actual_sources_set    
+
+        if not false_positives and not false_negatives:
+            print("  - Result: ✅ Perfect Match!")
+        else:
+            if false_positives:
+                print(f"  - 🔻 False Positives (hurts precision): {sorted(list(false_positives))}")
+            if false_negatives:
+                print(f"  - 🔻 False Negatives (hurts recall):    {sorted(list(false_negatives))}")
+        print("-"*59 + "\n")
+
+        context_precision, context_recall = 0.0, 0.0
+        if expected_sources_set or actual_sources_set:
+            tp = len(expected_sources_set.intersection(actual_sources_set))
+            if len(actual_sources_set) > 0: context_precision = tp / len(actual_sources_set)
+            if len(expected_sources_set) > 0: context_recall = tp / len(expected_sources_set)
+        elif not expected_sources_set and not actual_sources_set:
+            context_precision, context_recall = 1.0, 1.0
+
+        # TURN DEBUG on Answer Correctness
+        # print("\n" + "-"*20 + " ANSWER & CORRECTNESS EVALUATION " + "-"*20)
+        # print(f"  - Ground Truth Answer: {ground_truth_answer}")
+        # print(f"  - Generated Answer:    {answer_text}")
+        # print("-" * 59)
+        
+        answer_correctness_score = None
+        if ground_truth_answer and "ERROR" not in answer_text:
+            try:
+                judge_msg = ANSWER_CORRECTNESS_JUDGE_PROMPT.format(ground_truth_answer=ground_truth_answer, generated_answer=answer_text)
+                print(f"  - Judge Prompt Sent:\n{judge_msg}")
+                raw_correctness = call_llm([{"role": "user", "content": judge_msg}], temperature=0.0)
+                print(f"  - Judge Raw Response: {raw_correctness}")
+                correctness_data = _parse_judge_json(raw_correctness)
+                if correctness_data and "correctness_score" in correctness_data:
+                    answer_correctness_score = float(correctness_data["correctness_score"])
+                    print(f"  - Final Score: {answer_correctness_score}")
+            except Exception as e:
+                print(f"ERROR during answer correctness judging: {e}")
+
+        faithfulness = None
+        hallucination_rate = None
+        source_docs = response.get("source_documents", [])
+        if source_docs and "ERROR" not in answer_text:
+            context_blob = "\n---\n".join([doc.page_content for doc in source_docs])
+            judge_msg = FAITHFULNESS_JUDGE_PROMPT.format(query=query, answer=answer_text, sources=context_blob)
+            try:
+                if context_blob.strip():
+                    raw = call_llm([{"role": "user", "content": judge_msg}], temperature=0.0)
+                    data = _parse_judge_json(raw)
+                    if data:
+                        denom = data.get("supported", 0) + data.get("contradicted", 0) + data.get("not_enough_info", 0)
+                        if denom > 0: 
+                            faithfulness = round(data.get("supported", 0) / denom, 3)
+                            hallucination_rate = 1.0 - faithfulness
+                        elif data.get("ignored", 0) > 0: 
+                            faithfulness = 1.0
+                            hallucination_rate = 0.0
+                            
+            except Exception as e:
+                print(f"ERROR during faithfulness judging: {e}")
+
+        sources_pretty = ", ".join(sorted(s)) if (s:=actual_sources_set) else ""
+        results.append({
+            "test_id": fx.get("test_id", "N/A"), "title": fx.get("title", "N/A"),
+            "route_correct": "✅" if route_correct else "❌", "expected_route": expected_route, "actual_route": actual_route,
+            "behavior_f1": f"{behavior_metrics['f1_score']:.2f}", "emotion_f1": f"{emotion_metrics['f1_score']:.2f}",
+            "topic_f1": f"{topic_metrics['f1_score']:.2f}", "context_f1": f"{context_metrics['f1_score']:.2f}",
+            "generated_answer": answer_text, "sources": sources_pretty, "source_count": len(actual_sources_set),
+            "context_precision": context_precision, "context_recall": context_recall,
+            "faithfulness": faithfulness, "hallucination_rate": hallucination_rate,
+            "answer_correctness": answer_correctness_score,
+            "category": category, "error_class": error_class,
+            "latency_ms": latency_ms
+        })
+        
+    df = pd.DataFrame(results)
+    summary_text, table_rows, headers = "No valid test fixtures found to evaluate.", [], []
+    
+    if not df.empty:
+        # Add "hallucination_rate" to this list of columns to ensure it is not dropped.
+        cols = [
+            "test_id", "title", "route_correct", "expected_route", "actual_route",
+            "behavior_f1", "emotion_f1", "topic_f1", "context_f1",
+            "generated_answer", "sources", "source_count", 
+            "context_precision", "context_recall", 
+            "faithfulness", "hallucination_rate",
+            "answer_correctness", 
+            "category", "error_class", "latency_ms", 
+        ]   
+        df = df[[c for c in cols if c in df.columns]]     
+
+        # --- START OF MODIFICATION ---
+        pct = df["route_correct"].value_counts(normalize=True).get("✅", 0) * 100
+        to_f = lambda s: pd.to_numeric(s, errors="coerce")
+        
+        # Calculate the mean for the NLU F1 scores
+        bf1_mean = to_f(df["behavior_f1"]).mean() * 100
+        ef1_mean = to_f(df["emotion_f1"]).mean() * 100
+        tf1_mean = to_f(df["topic_f1"]).mean() * 100
+        cf1_mean = to_f(df["context_f1"]).mean() * 100
+
+        # Calculate the mean for Faithfulness
+        faith_mean = to_f(df["faithfulness"]).mean() * 100
+        # --- CHANGE 6: Calculate the mean for the new metric ---
+        halluc_mean = to_f(df["hallucination_rate"]).mean() * 100
+        
+        rag_with_sources_pct = (df["source_count"] > 0).mean() * 100 if "source_count" in df else 0
+
+        # Add the NLU metrics to the summary f-string
+        # Choose to use Hallucination - **RAG: Faithfulness**: {faith_mean:.1f}%  
+        summary_text = f"""## Evaluation Summary
+- **Routing Accuracy**: {pct:.2f}%
+- **Behaviour F1 (avg)**: {bf1_mean:.2f}%
+- **Emotion F1 (avg)**: {ef1_mean:.2f}%
+- **Topic F1 (avg)**: {tf1_mean:.2f}%
+- **Context F1 (avg)**: {cf1_mean:.2f}%
+- **RAG: Context Precision**: {(to_f(df["context_precision"]).mean() * 100):.1f}%
+- **RAG: Context Recall**: {(to_f(df["context_recall"]).mean() * 100):.1f}%
+- **RAG  Answers w/ Sources**: {rag_with_sources_pct:.1f}%
+- **RAG: Hallucination Rate**: {halluc_mean:.1f}% (Lower is better)
+- **RAG: Answer Correctness (LLM-judge)**: {(to_f(df["answer_correctness"]).mean() * 100):.1f}%
+- **RAG: Avg Latency (ms)**: {to_f(df["latency_ms"]).mean():.1f}
+"""
+        # --- END OF MODIFICATION ---
+        print(summary_text)
+        
+        df_display = df.rename(columns={"context_precision": "Ctx. Precision", "context_recall": "Ctx. Recall"})
+        table_rows = df_display.values.tolist()
+        headers = df_display.columns.tolist()
+
+        # --- NEW: per-category averages ---
+        try:
+            cat_means = df.groupby("category")["answer_correctness"].mean().reset_index()
+            print("\n📊 Correctness by Category:")
+            print(cat_means.to_string(index=False))
+        except Exception as e:
+            print(f"WARNING: Could not compute category breakdown: {e}")
+
+        # --- NEW: confusion-style matrix ---
+        try:
+            confusion = pd.crosstab(df.get("category", []), df.get("error_class", []),
+                                    rownames=["Category"], colnames=["Error Class"], dropna=False)
+            print("\n📊 Error Class Distribution by Category:")
+            print(confusion.to_string())
+        except Exception as e:
+            print(f"WARNING: Could not build confusion matrix: {e}")
+        # END
+        
+    else:
+        summary_text = "No valid test fixtures found to evaluate."
+        table_rows, headers = [], []
+        
+
+    return summary_text, table_rows, headers
+    # return summary_text, table_rows
+
+## END