Spaces:
Running
Running
| 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), | |
| } | |