import time import hashlib import numpy as np from typing import List, Dict, Any, Optional from collections import defaultdict import config as cfg from cache import LRUCache _FAITHFULNESS_PROMPT = """You are a strict judge evaluating factual correctness and faithfulness. Given the context and the answer, rate it on two axes (1-10, where 10 is perfect): 1. Faithfulness: Does the answer stay grounded in the provided context? Does it avoid introducing unsupported claims, hallucinations, or contradictions? 2. Answer Accuracy: Is the answer factually correct overall based on the context? Does it correctly address the question? Context: {context} Answer: {answer} Return ONLY two numbers on separate lines, e.g.: Faithfulness: 8 Accuracy: 7""" class Evaluator: def __init__(self, hybrid_retriever, reranker, llm_handler, context_manager, embedder): self.retriever = hybrid_retriever self.reranker = reranker self.llm = llm_handler self.context_manager = context_manager self.embedder = embedder self._relevance_map: Optional[Dict[str, List[str]]] = None self._judge_cache = LRUCache(max_size=100, ttl_seconds=3600) def build_relevance_map(self, chunks: List[Dict[str, Any]]): self._relevance_map = defaultdict(list) for c in chunks: node_id = c.get("node_id") if node_id: self._relevance_map[node_id].append(c["chunk_id"]) def create_eval_set( self, documents: List[Dict[str, Any]], num_questions: int = 20 ) -> List[Dict[str, Any]]: seen_nodes: set = set() eval_samples = [] candidates = [d for d in documents if len(d.get("text", "")) > 100] for doc in candidates: node_id = doc.get("node_id") if not node_id or node_id in seen_nodes: continue seen_nodes.add(node_id) title = doc.get("title", "") summary = doc.get("summary", "") text = doc.get("text", "") if summary and "?" in summary: question = summary.strip() elif title: question = f"Explain the key points about {title}." else: topic = (summary or text).split()[:10] question = f"What does the document cover about {' '.join(topic)}?" relevant_ids = self._relevance_map.get(node_id, [doc.get("chunk_id")]) if self._relevance_map else [doc.get("chunk_id") or node_id] eval_samples.append({ "question": question, "relevant_id": doc.get("chunk_id") or node_id, "relevant_ids": relevant_ids, "relevant_text": text, "node_id": node_id, "title": title, "summary": summary, }) if len(eval_samples) >= num_questions: break return eval_samples def evaluate_retrieval( self, eval_set: List[Dict[str, Any]] ) -> Dict[str, float]: hit_rates = {1: [], 5: [], 10: []} mrrs = {5: [], 10: []} precisions = {1: [], 5: [], 10: []} recalls = {1: [], 5: [], 10: []} latencies = [] for sample in eval_set: query = sample["question"] relevant_ids = set(sample["relevant_ids"]) start = time.perf_counter() query_emb = self.embedder.embed_query(query) results = self.retriever.search(query, query_emb, n_results=cfg.HYBRID_TOP_K) latency_ms = (time.perf_counter() - start) * 1000 latencies.append(latency_ms) retrieved_ids = [r["id"] for r in results] for k in [1, 5, 10]: rids = retrieved_ids[:k] hits = sum(1 for rid in rids if rid in relevant_ids) hit_rates[k].append(1.0 if hits > 0 else 0.0) precisions[k].append(hits / k if k > 0 else 0.0) recalls[k].append(hits / len(relevant_ids) if relevant_ids else 0.0) for k in [5, 10]: rank_list = retrieved_ids[:k] found_any = False for rank, rid in enumerate(rank_list): if rid in relevant_ids: mrrs[k].append(1.0 / (rank + 1)) found_any = True break if not found_any: mrrs[k].append(0.0) metrics = {} for k in [1, 5, 10]: metrics[f"HitRate@{k}"] = float(np.mean(hit_rates[k])) if hit_rates[k] else 0.0 metrics[f"Precision@{k}"] = float(np.mean(precisions[k])) if precisions[k] else 0.0 metrics[f"Recall@{k}"] = float(np.mean(recalls[k])) if recalls[k] else 0.0 for k in [5, 10]: metrics[f"MRR@{k}"] = float(np.mean(mrrs[k])) if mrrs[k] else 0.0 metrics["Avg Retrieval Latency (ms)"] = ( float(np.mean(latencies)) if latencies else 0.0 ) return metrics def _judge_faithfulness(self, context: str, answer: str) -> Dict[str, Optional[float]]: if not self.llm: return {"faithfulness": None, "accuracy": None} cache_key = hashlib.md5((context[:500] + answer[:500]).encode()).hexdigest() cached = self._judge_cache.get(cache_key) if cached: return cached prompt = _FAITHFULNESS_PROMPT.format(context=context[:6000], answer=answer[:1500]) response = self.llm.generate(prompt) if not response: return {"faithfulness": None, "accuracy": None} faith, acc = None, None for line in response.strip().split("\n"): line = line.strip().lower() if line.startswith("faithfulness:") or line.startswith("faithfulness :"): try: faith = float(line.split(":")[-1].strip().split("/")[0]) / 10.0 except (ValueError, IndexError): pass elif line.startswith("accuracy:") or line.startswith("accuracy :"): try: acc = float(line.split(":")[-1].strip().split("/")[0]) / 10.0 except (ValueError, IndexError): pass result = {"faithfulness": min(faith, 1.0) if faith is not None else None, "accuracy": min(acc, 1.0) if acc is not None else None} self._judge_cache.put(cache_key, result) return result def evaluate_generation( self, eval_set: List[Dict[str, Any]] ) -> Dict[str, Any]: latencies = [] faithfulness_scores = [] accuracy_scores = [] answers = [] for sample in eval_set[:10]: query = sample["question"] start = time.perf_counter() query_emb = self.embedder.embed_query(query) results = self.retriever.search(query, query_emb, n_results=cfg.HYBRID_TOP_K) reranked = self.reranker.rerank(query, results[:cfg.RERANK_CANDIDATES], top_k=cfg.FINAL_TOP_K) prompt = self.context_manager.assemble_prompt(query, reranked) if self.llm: answer = self.llm.generate(prompt) else: answer = "(LLM not configured — set GROQ_API_KEY)" latency_ms = (time.perf_counter() - start) * 1000 latencies.append(latency_ms) if not answer or answer == "(LLM not configured — set GROQ_API_KEY)": answers.append({ "question": query, "answer": answer or "(no response)", "latency_ms": round(latency_ms, 1), "faithfulness": "N/A", "accuracy": "N/A", }) continue context_for_judge = prompt.split("Context:\n")[1].split("\n\nQuestion:")[0] if "Context:\n" in prompt else "" judgment = self._judge_faithfulness(context_for_judge, answer) if judgment["faithfulness"] is not None: faithfulness_scores.append(judgment["faithfulness"]) if judgment["accuracy"] is not None: accuracy_scores.append(judgment["accuracy"]) answers.append({ "question": query, "answer": answer or "(no response)", "latency_ms": round(latency_ms, 1), "faithfulness": round(judgment["faithfulness"], 3) if judgment["faithfulness"] is not None else "N/A", "accuracy": round(judgment["accuracy"], 3) if judgment["accuracy"] is not None else "N/A", }) result = { "Avg End-to-End Latency (ms)": ( round(float(np.mean(latencies)), 1) if latencies else 0.0 ), "answers": answers, "num_samples": len(answers), } if faithfulness_scores: result["Avg Faithfulness"] = round(float(np.mean(faithfulness_scores)), 3) else: result["Avg Faithfulness"] = "N/A (set GROQ_API_KEY)" if accuracy_scores: result["Avg Answer Accuracy"] = round(float(np.mean(accuracy_scores)), 3) else: result["Avg Answer Accuracy"] = "N/A (set GROQ_API_KEY)" return result def full_evaluation( self, documents: List[Dict[str, Any]], num_questions: int = 20 ) -> Dict[str, Any]: if self._relevance_map is None: self.build_relevance_map(documents) eval_set = self.create_eval_set(documents, num_questions) retrieval_metrics = self.evaluate_retrieval(eval_set) generation_metrics = self.evaluate_generation(eval_set) return { "retrieval": retrieval_metrics, "generation": generation_metrics, "num_questions": len(eval_set), }