From 08e54e6df16bb32c0b3bff3ffef4f3c1b12447e9 Mon Sep 17 00:00:00 2001 From: Rounak Bende <153109208+rounakbende10@users.noreply.github.com> Date: Mon, 21 Apr 2025 00:24:37 -0400 Subject: [PATCH 1/2] rag evaluator updated --- rag_model/RAGEvaluator.py | 774 ++++++++++++++++++++++++++----------- rag_model/rag_evaluator.py | 220 ++++++----- 2 files changed, 664 insertions(+), 330 deletions(-) diff --git a/rag_model/RAGEvaluator.py b/rag_model/RAGEvaluator.py index 5302010..e845373 100644 --- a/rag_model/RAGEvaluator.py +++ b/rag_model/RAGEvaluator.py @@ -1,121 +1,275 @@ import json import os from typing import Any, Callable, Dict, List, Optional, Tuple +from collections import defaultdict +import warnings import mlflow import pandas as pd from openai import OpenAI -from sklearn.metrics.pairwise import cosine_similarity +# Note: sklearn cosine_similarity is not used in the provided snippet +# from sklearn.metrics.pairwise import cosine_similarity +# Suppress NLTK download messages if already downloaded +os.environ['NLTK_DATA'] = os.path.join(os.path.expanduser('~'), 'nltk_data') class RAGEvaluator: - """Evaluator for RAG pipelines""" - + """ + Evaluator for RAG pipelines, including bias analysis across topics and industries. + + Assumes the test dataset includes 'topic' and 'industry' keys for each entry. + """ + def __init__(self, test_dataset_path: str, rag_pipeline): """ - Initialize evaluator with test dataset - + Initialize evaluator with test dataset. + Args: test_dataset_path: Path to test dataset with format: [ { "query": "...", "ground_truth": "...", - "relevant_docs": ["...", ...] (optional) + "relevant_docs": ["...", ...], # Optional for some evals + "topic": "...", # Required for bias analysis + "industry": "..." # Required for bias analysis }, ... ] + rag_pipeline: An instance of the RAG pipeline to be evaluated. + Expected to have methods like `semantic_search(query, k=...)`, + `generate_response(query, docs)`, `query(query)`, + and a `config` attribute with relevant parameters. """ self.test_dataset = self._load_test_dataset(test_dataset_path) self.rag_pipeline = rag_pipeline + # Ensure API key is accessed correctly from the pipeline's config + if not hasattr(rag_pipeline, 'config') or not hasattr(rag_pipeline.config, 'llm_api_key'): + raise ValueError("RAG pipeline object must have a 'config' attribute with an 'llm_api_key'.") + # Consider adding error handling for missing API key self.client = OpenAI(base_url="https://api.groq.com/openai/v1", api_key=self.rag_pipeline.config.llm_api_key) - + self._validate_dataset_keys() + def _load_test_dataset(self, path: str) -> List[Dict[str, Any]]: - """Load test dataset from file""" - with open(path, 'r') as f: - return json.load(f) - - def evaluate_embedding_model(self) -> Dict[str, float]: - """Evaluate embedding model quality""" - results = {} + """Load test dataset from JSON file.""" + try: + with open(path, 'r', encoding='utf-8') as f: + return json.load(f) + except FileNotFoundError: + print(f"Error: Test dataset file not found at {path}") + raise + except json.JSONDecodeError: + print(f"Error: Could not decode JSON from file at {path}") + raise + + def _validate_dataset_keys(self): + """Check if 'topic' and 'industry' keys are present in the dataset.""" + if not self.test_dataset: + warnings.warn("Test dataset is empty.") + return + + sample_item = self.test_dataset[0] + missing_keys = [] + if 'topic' not in sample_item: + missing_keys.append('topic') + if 'industry' not in sample_item: + missing_keys.append('industry') + + if missing_keys: + raise ValueError(f"Test dataset items are missing required keys for bias analysis: {', '.join(missing_keys)}") + # Optional: Check all items, though this can be slow for large datasets + # for i, item in enumerate(self.test_dataset): + # if 'topic' not in item or 'industry' not in item: + # warnings.warn(f"Dataset item at index {i} is missing 'topic' or 'industry'. Bias analysis might be incomplete.") + + + def _safe_divide(self, numerator: float, denominator: int) -> float: + """Helper function for safe division, returns 0 if denominator is 0.""" + return numerator / denominator if denominator > 0 else 0.0 + + def evaluate_embedding_model(self) -> Dict[str, Any]: + """ + Evaluate embedding model quality based on retrieval metrics, + stratified by topic and industry. + + Returns: + A dictionary containing overall and stratified retrieval metrics. + """ + # Accumulators for overall metrics retrieval_precision_sum = 0 retrieval_recall_sum = 0 retrieval_f1_sum = 0 - + num_cases_with_relevant_docs = 0 + + # Accumulators for stratified metrics + # Structure: { 'topic': { 'topic_name': {'precision_sum': 0, 'recall_sum': 0, 'f1_sum': 0, 'count': 0}, ... } } + topic_metrics = defaultdict(lambda: defaultdict(float)) + industry_metrics = defaultdict(lambda: defaultdict(float)) + for i, test_case in enumerate(self.test_dataset): - query = test_case["query"] - - # Skip if no ground truth relevant docs - if "relevant_docs" not in test_case: + query = test_case.get("query") + ground_truth_docs_list = test_case.get("relevant_docs") + topic = test_case.get("topic") + industry = test_case.get("industry") + + if not query or not topic or not industry: + warnings.warn(f"Skipping test case {i} due to missing query, topic, or industry.") + continue + + # Skip if no ground truth relevant docs for this test case + if not ground_truth_docs_list: continue - - ground_truth_docs = set(test_case["relevant_docs"]) - - # Get retrieved docs + + num_cases_with_relevant_docs += 1 + ground_truth_docs = set(ground_truth_docs_list) + + # Get retrieved docs using the default top_k from the pipeline config + # Ensure semantic_search can handle default k retrieved_docs = set(self.rag_pipeline.semantic_search(query)) + + # Calculate precision, recall, F1 + true_positives = len(ground_truth_docs.intersection(retrieved_docs)) - # Calculate precision and recall - if len(retrieved_docs) > 0: - precision = len(ground_truth_docs.intersection(retrieved_docs)) / len(retrieved_docs) - else: - precision = 0 - - if len(ground_truth_docs) > 0: - recall = len(ground_truth_docs.intersection(retrieved_docs)) / len(ground_truth_docs) - else: - recall = 0 - - # Calculate F1 - if precision + recall > 0: - f1 = 2 * precision * recall / (precision + recall) - else: - f1 = 0 - + precision = self._safe_divide(true_positives, len(retrieved_docs)) + recall = self._safe_divide(true_positives, len(ground_truth_docs)) + f1 = self._safe_divide(2 * precision * recall, precision + recall) + + # Accumulate overall scores retrieval_precision_sum += precision retrieval_recall_sum += recall retrieval_f1_sum += f1 - - num_cases = len(self.test_dataset) - results["retrieval_precision"] = retrieval_precision_sum / num_cases - results["retrieval_recall"] = retrieval_recall_sum / num_cases - results["retrieval_f1"] = retrieval_f1_sum / num_cases - + + # Accumulate scores per topic + topic_metrics[topic]['precision_sum'] += precision + topic_metrics[topic]['recall_sum'] += recall + topic_metrics[topic]['f1_sum'] += f1 + topic_metrics[topic]['count'] += 1 + + # Accumulate scores per industry + industry_metrics[industry]['precision_sum'] += precision + industry_metrics[industry]['recall_sum'] += recall + industry_metrics[industry]['f1_sum'] += f1 + industry_metrics[industry]['count'] += 1 + + # Calculate final average metrics + results = { + "overall": { + "retrieval_precision": self._safe_divide(retrieval_precision_sum, num_cases_with_relevant_docs), + "retrieval_recall": self._safe_divide(retrieval_recall_sum, num_cases_with_relevant_docs), + "retrieval_f1": self._safe_divide(retrieval_f1_sum, num_cases_with_relevant_docs), + "evaluated_cases_count": num_cases_with_relevant_docs + }, + "by_topic": {}, + "by_industry": {} + } + + for topic, metrics in topic_metrics.items(): + count = metrics['count'] + results["by_topic"][topic] = { + "retrieval_precision": self._safe_divide(metrics['precision_sum'], count), + "retrieval_recall": self._safe_divide(metrics['recall_sum'], count), + "retrieval_f1": self._safe_divide(metrics['f1_sum'], count), + "evaluated_cases_count": count + } + + for industry, metrics in industry_metrics.items(): + count = metrics['count'] + results["by_industry"][industry] = { + "retrieval_precision": self._safe_divide(metrics['precision_sum'], count), + "retrieval_recall": self._safe_divide(metrics['recall_sum'], count), + "retrieval_f1": self._safe_divide(metrics['f1_sum'], count), + "evaluated_cases_count": count + } + return results - - def evaluate_top_k_strategy(self, k_values: List[int]) -> Dict[str, Dict[str, float]]: - """Evaluate different top-k retrieval strategies""" + + def evaluate_top_k_strategy(self, k_values: List[int]) -> Dict[str, Dict[str, Any]]: + """ + Evaluate different top-k retrieval strategies based on answer accuracy, + stratified by topic and industry. + + Args: + k_values: A list of integers representing the top-k values to test. + + Returns: + A nested dictionary with results for each k, including stratified metrics. + Example: {"top_k_3": {"overall": {...}, "by_topic": {...}, "by_industry": {...}}} + """ results = {} - + for k in k_values: - with mlflow.start_run(nested=True, run_name=f"top_k_{k}"): + run_name = f"top_k_{k}" + with mlflow.start_run(nested=True, run_name=run_name): mlflow.log_param("top_k", k) - + + # Overall accumulators for this k accuracy_sum = 0 - - for test_case in self.test_dataset: - query = test_case["query"] - ground_truth = test_case["ground_truth"] + num_cases = 0 + + # Stratified accumulators for this k + topic_metrics = defaultdict(lambda: defaultdict(float)) + industry_metrics = defaultdict(lambda: defaultdict(float)) + + for i, test_case in enumerate(self.test_dataset): + query = test_case.get("query") + ground_truth = test_case.get("ground_truth") + topic = test_case.get("topic") + industry = test_case.get("industry") + + if not query or not ground_truth or not topic or not industry: + warnings.warn(f"Skipping test case {i} for top_k={k} due to missing fields.") + continue + num_cases += 1 + # Get retrieved docs with specific k retrieved_docs = self.rag_pipeline.semantic_search(query, k=k) - + # Generate response with these docs response = self.rag_pipeline.generate_response(query, retrieved_docs) - + # Evaluate answer quality using LLM as judge accuracy = self._llm_judge_answer_quality(response, ground_truth) - accuracy_sum += accuracy - avg_accuracy = accuracy_sum / len(self.test_dataset) - mlflow.log_metric("answer_accuracy", avg_accuracy) - - results[f"top_k_{k}"] = {"answer_accuracy": avg_accuracy} - + # Accumulate overall + accuracy_sum += accuracy + + # Accumulate stratified + topic_metrics[topic]['accuracy_sum'] += accuracy + topic_metrics[topic]['count'] += 1 + industry_metrics[industry]['accuracy_sum'] += accuracy + industry_metrics[industry]['count'] += 1 + + # Calculate and log overall average accuracy for this k + avg_accuracy = self._safe_divide(accuracy_sum, num_cases) + mlflow.log_metric("answer_accuracy_overall", avg_accuracy) + + k_results = { + "overall": {"answer_accuracy": avg_accuracy, "evaluated_cases_count": num_cases}, + "by_topic": {}, + "by_industry": {} + } + + # Calculate, log, and store stratified results for this k + for topic, metrics in topic_metrics.items(): + count = metrics['count'] + avg_topic_accuracy = self._safe_divide(metrics['accuracy_sum'], count) + mlflow.log_metric(f"answer_accuracy_topic_{topic}", avg_topic_accuracy) + k_results["by_topic"][topic] = {"answer_accuracy": avg_topic_accuracy, "evaluated_cases_count": count} + + for industry, metrics in industry_metrics.items(): + count = metrics['count'] + avg_industry_accuracy = self._safe_divide(metrics['accuracy_sum'], count) + mlflow.log_metric(f"answer_accuracy_industry_{industry}", avg_industry_accuracy) + k_results["by_industry"][industry] = {"answer_accuracy": avg_industry_accuracy, "evaluated_cases_count": count} + + results[run_name] = k_results + return results - + def _llm_judge_answer_quality(self, response: str, ground_truth: str) -> float: - """Use LLM as judge to evaluate answer quality""" + """Use LLM as judge to evaluate answer quality (factual accuracy/relevance).""" prompt = f""" On a scale of 0 to 1, how accurately does the following answer address the question compared to the ground truth? @@ -123,188 +277,370 @@ def _llm_judge_answer_quality(self, response: str, ground_truth: str) -> float: Generated Answer: {response} - Score only the factual accuracy and relevance, not writing style or verbosity. - Return only a single number between 0 and 1, where: - 0 = completely incorrect or irrelevant - 1 = perfectly accurate and relevant + Score only the factual accuracy and relevance, not writing style or verbosity. + Return only a single floating-point number between 0.0 and 1.0. + Examples: 0.0, 0.25, 0.5, 0.75, 1.0 """ - - judge_response = self.client.chat.completions.create( - model="llama3-8b-8192", - messages=[{"role": "user", "content": prompt}], - temperature=0.1 - ) - - # Extract score from response try: - score = float(judge_response.choices[0].message.content.strip()) - return min(max(score, 0.0), 1.0) # Ensure score is in [0,1] - except ValueError: - # If can't parse as float, use a default + judge_response = self.client.chat.completions.create( + model="llama3-8b-8192", # Ensure this model is available via Groq + messages=[{"role": "user", "content": prompt}], + temperature=0.1, + max_tokens=10 # Restrict output length + ) + score_str = judge_response.choices[0].message.content.strip() + # Try to extract a number even if there's extra text + import re + match = re.search(r"(\d\.?\d*)", score_str) + if match: + score = float(match.group(1)) + return min(max(score, 0.0), 1.0) # Clamp score to [0, 1] + else: + warnings.warn(f"LLM Judge (Quality) failed to return a parseable number. Response: '{score_str}'. Defaulting to 0.5.") + return 0.5 + except Exception as e: + warnings.warn(f"LLM Judge (Quality) API call failed: {e}. Defaulting to 0.5.") return 0.5 - - def evaluate_rag_system(self) -> Dict[str, float]: - """Evaluate full RAG system using mlflow.evaluate()""" - results = {} - - # Run full pipeline on test dataset + + + def evaluate_rag_system(self) -> Dict[str, Any]: + """ + Evaluate the full RAG system using BLEU and LLM-judged metrics, + stratifying LLM metrics by topic and industry. + + Returns: + A dictionary containing overall and stratified evaluation results. + Note: BLEU score is calculated overall, not stratified per category easily. + """ predictions = [] ground_truths = [] - - for test_case in self.test_dataset: - query = test_case["query"] - ground_truth = test_case["ground_truth"] - - rag_result = self.rag_pipeline.query(query) - - predictions.append(rag_result["response"]) - ground_truths.append(ground_truth) - - # Convert to dataframe for mlflow - eval_df = pd.DataFrame({ - "prediction": predictions, - "target": ground_truths - }) - - # Basic metrics - with mlflow.start_run(nested=True, run_name="rag_evaluation"): - # Calculate metric: BLEU score - import nltk - from nltk.translate.bleu_score import SmoothingFunction, corpus_bleu - nltk.download('punkt', quiet=True) - nltk.download('punkt_tab', quiet=True) - - # Tokenize predictions and references - tokenized_predictions = [nltk.word_tokenize(pred.lower()) for pred in predictions] - tokenized_references = [[nltk.word_tokenize(ref.lower())] for ref in ground_truths] - - # Calculate BLEU score - smoothing = SmoothingFunction().method1 - bleu_score = corpus_bleu(tokenized_references, tokenized_predictions, smoothing_function=smoothing) - - mlflow.log_metric("bleu_score", bleu_score) - results["bleu_score"] = bleu_score - - # LLM as judge metrics - accuracy_sum = 0 - relevance_sum = 0 - completeness_sum = 0 - - for pred, truth in zip(predictions, ground_truths): + topics = [] + industries = [] + + print("Generating predictions for full RAG system evaluation...") + for i, test_case in enumerate(self.test_dataset): + query = test_case.get("query") + ground_truth = test_case.get("ground_truth") + topic = test_case.get("topic") + industry = test_case.get("industry") + + if not query or not ground_truth or not topic or not industry: + warnings.warn(f"Skipping test case {i} for full RAG eval due to missing fields.") + continue + + try: + # Assuming rag_pipeline.query returns a dict like {"response": "..."} + rag_result = self.rag_pipeline.query(query) + predictions.append(rag_result.get("response", "")) # Handle missing response + ground_truths.append(ground_truth) + topics.append(topic) + industries.append(industry) + except Exception as e: + warnings.warn(f"Error processing query for test case {i}: {e}. Skipping.") + continue # Skip this test case if pipeline fails + + if not predictions: + print("No predictions were generated. Cannot evaluate RAG system.") + return {"error": "No predictions generated."} + + results = { + "overall": {}, + "by_topic": defaultdict(lambda: defaultdict(float)), + "by_industry": defaultdict(lambda: defaultdict(float)) + } + num_examples = len(predictions) + + # --- Overall Metrics --- + with mlflow.start_run(nested=True, run_name="rag_full_evaluation"): + # 1. BLEU Score (Overall) + try: + import nltk + from nltk.translate.bleu_score import SmoothingFunction, corpus_bleu + # Attempt to download punkt quietly, handle potential errors + try: + nltk.download('punkt', quiet=True, download_dir=os.environ['NLTK_DATA']) + except Exception as e: + warnings.warn(f"NLTK punkt download failed: {e}. BLEU score calculation might fail.") + + tokenized_predictions = [nltk.word_tokenize(pred.lower()) for pred in predictions] + tokenized_references = [[nltk.word_tokenize(ref.lower())] for ref in ground_truths] + + smoothing = SmoothingFunction().method1 + bleu_score = corpus_bleu(tokenized_references, tokenized_predictions, smoothing_function=smoothing) + + mlflow.log_metric("bleu_score_overall", bleu_score) + results["overall"]["bleu_score"] = bleu_score + except ImportError: + warnings.warn("NLTK not installed. Skipping BLEU score calculation. Install with: pip install nltk") + results["overall"]["bleu_score"] = None + except Exception as e: + warnings.warn(f"BLEU score calculation failed: {e}") + results["overall"]["bleu_score"] = None + + + # 2. LLM as Judge Metrics (Overall and Stratified) + print("Evaluating generated responses using LLM judge...") + accuracy_sum, relevance_sum, completeness_sum = 0, 0, 0 + # Use temporary dicts for sums before averaging + topic_sums = defaultdict(lambda: defaultdict(float)) + industry_sums = defaultdict(lambda: defaultdict(float)) + # Counts needed for averaging stratified results + topic_counts = defaultdict(int) + industry_counts = defaultdict(int) + + + for i, (pred, truth, topic, industry) in enumerate(zip(predictions, ground_truths, topics, industries)): + if i % 10 == 0: # Print progress + print(f" Judging response {i+1}/{num_examples}...") # Accuracy accuracy = self._llm_judge_answer_quality(pred, truth) accuracy_sum += accuracy - + topic_sums[topic]['accuracy_sum'] += accuracy + industry_sums[industry]['accuracy_sum'] += accuracy + # Relevance - relevance = self._llm_judge_answer_relevance(pred, truth) + relevance = self._llm_judge_answer_relevance(pred, truth) # Assuming query implied by truth relevance_sum += relevance - + topic_sums[topic]['relevance_sum'] += relevance + industry_sums[industry]['relevance_sum'] += relevance + # Completeness completeness = self._llm_judge_answer_completeness(pred, truth) completeness_sum += completeness + topic_sums[topic]['completeness_sum'] += completeness + industry_sums[industry]['completeness_sum'] += completeness + + # Increment counts for averaging + topic_counts[topic] += 1 + industry_counts[industry] += 1 + + # Calculate and log overall averages + avg_accuracy = self._safe_divide(accuracy_sum, num_examples) + avg_relevance = self._safe_divide(relevance_sum, num_examples) + avg_completeness = self._safe_divide(completeness_sum, num_examples) + + mlflow.log_metric("llm_judge_accuracy_overall", avg_accuracy) + mlflow.log_metric("llm_judge_relevance_overall", avg_relevance) + mlflow.log_metric("llm_judge_completeness_overall", avg_completeness) + + results["overall"]["llm_judge_accuracy"] = avg_accuracy + results["overall"]["llm_judge_relevance"] = avg_relevance + results["overall"]["llm_judge_completeness"] = avg_completeness + results["overall"]["evaluated_cases_count"] = num_examples + + + # Calculate, log, and store stratified results + print("Calculating stratified LLM judge metrics...") + for topic, t_sums in topic_sums.items(): + count = topic_counts[topic] + avg_topic_accuracy = self._safe_divide(t_sums['accuracy_sum'], count) + avg_topic_relevance = self._safe_divide(t_sums['relevance_sum'], count) + avg_topic_completeness = self._safe_divide(t_sums['completeness_sum'], count) + + mlflow.log_metric(f"llm_judge_accuracy_topic_{topic}", avg_topic_accuracy) + mlflow.log_metric(f"llm_judge_relevance_topic_{topic}", avg_topic_relevance) + mlflow.log_metric(f"llm_judge_completeness_topic_{topic}", avg_topic_completeness) + + results["by_topic"][topic] = { + "llm_judge_accuracy": avg_topic_accuracy, + "llm_judge_relevance": avg_topic_relevance, + "llm_judge_completeness": avg_topic_completeness, + "evaluated_cases_count": count + } + + for industry, i_sums in industry_sums.items(): + count = industry_counts[industry] + avg_industry_accuracy = self._safe_divide(i_sums['accuracy_sum'], count) + avg_industry_relevance = self._safe_divide(i_sums['relevance_sum'], count) + avg_industry_completeness = self._safe_divide(i_sums['completeness_sum'], count) + + mlflow.log_metric(f"llm_judge_accuracy_industry_{industry}", avg_industry_accuracy) + mlflow.log_metric(f"llm_judge_relevance_industry_{industry}", avg_industry_relevance) + mlflow.log_metric(f"llm_judge_completeness_industry_{industry}", avg_industry_completeness) + + results["by_industry"][industry] = { + "llm_judge_accuracy": avg_industry_accuracy, + "llm_judge_relevance": avg_industry_relevance, + "llm_judge_completeness": avg_industry_completeness, + "evaluated_cases_count": count + } - num_examples = len(predictions) - avg_accuracy = accuracy_sum / num_examples - avg_relevance = relevance_sum / num_examples - avg_completeness = completeness_sum / num_examples - - mlflow.log_metric("llm_judge_accuracy", avg_accuracy) - mlflow.log_metric("llm_judge_relevance", avg_relevance) - mlflow.log_metric("llm_judge_completeness", avg_completeness) - - results["llm_judge_accuracy"] = avg_accuracy - results["llm_judge_relevance"] = avg_relevance - results["llm_judge_completeness"] = avg_completeness - + # Log results dictionary as artifact + try: + results_path = "rag_evaluation_results.json" + with open(results_path, 'w', encoding='utf-8') as f: + json.dump(results, f, indent=4) + mlflow.log_artifact(results_path) + print(f"Stratified evaluation results saved and logged to MLflow: {results_path}") + except Exception as e: + warnings.warn(f"Failed to log evaluation results as MLflow artifact: {e}") + + return results - + def _llm_judge_answer_relevance(self, response: str, ground_truth: str) -> float: - """Judge how relevant the answer is to the implied question""" + """Judge how relevant the answer is to the implied question (using ground_truth to infer).""" + # This prompt assumes the ground truth helps define the question's scope. prompt = f""" - On a scale of 0 to 1, how relevant is the following answer to the question implied by the ground truth? + Evaluate the relevance of the 'Generated Answer' to the core question or topic implied by the 'Ground Truth Answer'. + Score on a scale of 0.0 to 1.0, where 0.0 means completely irrelevant and 1.0 means perfectly relevant to the implied subject. - Ground Truth Answer: {ground_truth} + Ground Truth Answer (provides context for the implied question): + {ground_truth} - Generated Answer: {response} + Generated Answer: + {response} - Return only a single number between 0 and 1, where: - 0 = completely irrelevant - 1 = perfectly relevant + Return only a single floating-point number between 0.0 and 1.0. + Examples: 0.0, 0.25, 0.5, 0.75, 1.0 """ - - judge_response = self.client.chat.completions.create( - model="llama3-8b-8192", - messages=[{"role": "user", "content": prompt}], - temperature=0.1 - ) - try: - score = float(judge_response.choices[0].message.content.strip()) - return min(max(score, 0.0), 1.0) - except ValueError: + judge_response = self.client.chat.completions.create( + model="llama3-8b-8192", + messages=[{"role": "user", "content": prompt}], + temperature=0.1, + max_tokens=10 + ) + score_str = judge_response.choices[0].message.content.strip() + import re + match = re.search(r"(\d\.?\d*)", score_str) + if match: + score = float(match.group(1)) + return min(max(score, 0.0), 1.0) + else: + warnings.warn(f"LLM Judge (Relevance) failed to return a parseable number. Response: '{score_str}'. Defaulting to 0.5.") + return 0.5 + except Exception as e: + warnings.warn(f"LLM Judge (Relevance) API call failed: {e}. Defaulting to 0.5.") return 0.5 - + + def _llm_judge_answer_completeness(self, response: str, ground_truth: str) -> float: - """Judge how complete the answer is compared to ground truth""" + """Judge how complete the answer is compared to the ground truth.""" prompt = f""" - On a scale of 0 to 1, how complete is the following answer compared to the ground truth? + Evaluate how completely the 'Generated Answer' covers the key information present in the 'Ground Truth Answer'. + Score on a scale of 0.0 to 1.0, where 0.0 means it misses almost all key information, and 1.0 means it includes all essential information from the ground truth. - Ground Truth Answer: {ground_truth} + Ground Truth Answer: + {ground_truth} - Generated Answer: {response} + Generated Answer: + {response} - Return only a single number between 0 and 1, where: - 0 = completely incomplete (missing critical information) - 1 = perfectly complete (covers all important information) + Return only a single floating-point number between 0.0 and 1.0. + Examples: 0.0, 0.25, 0.5, 0.75, 1.0 """ - - judge_response = self.client.chat.completions.create( - model="llama3-8b-8192", - messages=[{"role": "user", "content": prompt}], - temperature=0.1 - ) - try: - score = float(judge_response.choices[0].message.content.strip()) - return min(max(score, 0.0), 1.0) - except ValueError: + judge_response = self.client.chat.completions.create( + model="llama3-8b-8192", + messages=[{"role": "user", "content": prompt}], + temperature=0.1, + max_tokens=10 + ) + score_str = judge_response.choices[0].message.content.strip() + import re + match = re.search(r"(\d\.?\d*)", score_str) + if match: + score = float(match.group(1)) + return min(max(score, 0.0), 1.0) + else: + warnings.warn(f"LLM Judge (Completeness) failed to return a parseable number. Response: '{score_str}'. Defaulting to 0.5.") + return 0.5 + except Exception as e: + warnings.warn(f"LLM Judge (Completeness) API call failed: {e}. Defaulting to 0.5.") return 0.5 - + def run_full_evaluation(self, experiment_name: str) -> Dict[str, Any]: - """Run all evaluations""" - mlflow.set_experiment(experiment_name) - - with mlflow.start_run(run_name="full_evaluation"): - # Log RAG configuration - mlflow.log_params({ - "embedding_model": self.rag_pipeline.config.embedding_model, - "llm_model": self.rag_pipeline.config.llm_model, - "top_k": self.rag_pipeline.config.top_k, - "temperature": self.rag_pipeline.config.temperature - }) - - # Evaluate embedding model - print("Evaluating embedding model...") - embedding_results = self.evaluate_embedding_model() - for metric, value in embedding_results.items(): - mlflow.log_metric(metric, value) - - # Evaluate top-k strategies - print("Evaluating top-k strategies...") - top_k_results = self.evaluate_top_k_strategy( - k_values=[1, 3, 5] - ) - - # Evaluate full RAG system - print("Evaluating full RAG system...") - rag_results = self.evaluate_rag_system() - for metric, value in rag_results.items(): - mlflow.log_metric(metric, value) - - all_results = { - "embedding_evaluation": embedding_results, - "top_k_evaluation": top_k_results, - "rag_evaluation": rag_results - } - - return all_results + """ + Run all evaluations (embedding, top-k, full RAG) and log results + to MLflow, including stratified metrics for bias analysis. + + Args: + experiment_name: The name for the MLflow experiment. + + Returns: + A dictionary containing the results from all evaluation steps. + """ + try: + mlflow.set_experiment(experiment_name) + except Exception as e: + warnings.warn(f"Could not set MLflow experiment '{experiment_name}': {e}. Check MLflow setup.") + # Decide if you want to proceed without MLflow or raise error + # For now, we proceed but MLflow logging might fail silently later if setup is wrong. + + all_results = {} + + # Use a single parent run for the entire evaluation process + with mlflow.start_run(run_name="full_rag_pipeline_evaluation"): + print("Starting full RAG pipeline evaluation...") + + # Log RAG configuration parameters + try: + mlflow.log_params({ + "embedding_model": getattr(self.rag_pipeline.config, 'embedding_model', 'N/A'), + "llm_model": getattr(self.rag_pipeline.config, 'llm_model', 'N/A'), + "configured_top_k": getattr(self.rag_pipeline.config, 'top_k', 'N/A'), + "llm_temperature": getattr(self.rag_pipeline.config, 'temperature', 'N/A') + }) + except AttributeError: + warnings.warn("Could not log all RAG pipeline config parameters. Ensure rag_pipeline.config has expected attributes.") + except Exception as e: + warnings.warn(f"MLflow log_params failed: {e}") + + + # --- Evaluate Embedding Model --- + print("\nEvaluating embedding model (Retrieval Metrics)...") + try: + embedding_results = self.evaluate_embedding_model() + all_results["embedding_evaluation"] = embedding_results + print(" Embedding evaluation complete.") + # Log overall embedding metrics + if "overall" in embedding_results: + for metric, value in embedding_results["overall"].items(): + if isinstance(value, (int, float)): # Log only numeric metrics + mlflow.log_metric(f"embedding_{metric}_overall", value) + # Log stratified embedding metrics (optional, can make MLflow UI busy) + # Consider logging these as artifacts instead if too numerous + # Example: log_stratified_metrics_to_mlflow(embedding_results, "embedding") + + except Exception as e: + print(f"Error during embedding evaluation: {e}") + all_results["embedding_evaluation"] = {"error": str(e)} + + # --- Evaluate Top-K Strategies --- + print("\nEvaluating top-k strategies (Answer Accuracy)...") + try: + # Define which k values to test + k_values_to_test = [1, 3, 5] # Example values + top_k_results = self.evaluate_top_k_strategy(k_values=k_values_to_test) + all_results["top_k_evaluation"] = top_k_results + print(f" Top-k evaluation complete for k={k_values_to_test}.") + # MLflow logging happens *inside* evaluate_top_k_strategy + except Exception as e: + print(f"Error during top-k evaluation: {e}") + all_results["top_k_evaluation"] = {"error": str(e)} + + # --- Evaluate Full RAG System --- + print("\nEvaluating full RAG system (BLEU, LLM Judged Metrics)...") + try: + rag_results = self.evaluate_rag_system() + all_results["rag_system_evaluation"] = rag_results + print(" Full RAG system evaluation complete.") + # MLflow logging happens *inside* evaluate_rag_system + except Exception as e: + print(f"Error during full RAG system evaluation: {e}") + all_results["rag_system_evaluation"] = {"error": str(e)} + + # --- Log aggregated results as artifact --- + try: + all_results_path = "full_evaluation_summary.json" + with open(all_results_path, 'w', encoding='utf-8') as f: + # Use default=str to handle potential non-serializable types like defaultdict + json.dump(all_results, f, indent=4, default=str) + mlflow.log_artifact(all_results_path) + print(f"\nFull evaluation summary saved and logged to MLflow: {all_results_path}") + except Exception as e: + warnings.warn(f"Failed to log full evaluation summary artifact: {e}") + + print("\nFull evaluation finished.") + return all_results \ No newline at end of file diff --git a/rag_model/rag_evaluator.py b/rag_model/rag_evaluator.py index 7a2eebe..9a7385b 100644 --- a/rag_model/rag_evaluator.py +++ b/rag_model/rag_evaluator.py @@ -70,39 +70,51 @@ def check_llm_results(results, thresholds): Returns: dict: Boolean flags indicating if each category passes and overall result. """ - def check_metrics(metrics, threshold_metrics): + def check_overall(metrics, threshold_metrics): return all(metrics.get(key, 0) >= threshold_metrics.get(key, 0) for key in threshold_metrics) - embedding_pass = check_metrics(results["embedding_evaluation"], thresholds["embedding_evaluation"]) - - top_k_pass = all( - results["top_k_evaluation"][k]["answer_accuracy"] >= thresholds["top_k_evaluation"].get(k, 0) - for k in thresholds["top_k_evaluation"] - ) + # Embedding Evaluation + embedding_metrics = results.get("embedding_evaluation", {}).get("overall", {}) + embedding_pass = check_overall(embedding_metrics, thresholds.get("embedding_evaluation", {})) + + # Top-K Evaluation (skip if error) + top_k_eval = results.get("top_k_evaluation", {}) + top_k_pass = False + if "error" not in top_k_eval: + top_k_pass = all( + top_k_eval.get(k, {}).get("answer_accuracy", 0) >= thresholds["top_k_evaluation"].get(k, 0) + for k in thresholds["top_k_evaluation"] + ) - rag_pass = check_metrics(results["rag_evaluation"], thresholds["rag_evaluation"]) + # RAG System Evaluation + rag_metrics = results.get("rag_system_evaluation", {}).get("overall", {}) + rag_pass = check_overall(rag_metrics, thresholds.get("rag_system_evaluation", {})) return { "embedding_pass": embedding_pass, "top_k_pass": top_k_pass, "rag_pass": rag_pass, - "overall_pass": embedding_pass or top_k_pass or rag_pass + "overall_pass": embedding_pass and top_k_pass and rag_pass } def main(): - print(sys.argv) + print("Starting RAG Evaluator...") + print(f"Arguments provided: {sys.argv}") # Get test dataset path from environment test_dataset_path = os.getenv("TEST_DATASET_PATH") + if not test_dataset_path: + print("Error: TEST_DATASET_PATH environment variable not set") + sys.exit(1) print(f"Test dataset path: {test_dataset_path}") # Define RAG configuration config = RAGConfig( embedding_model=os.getenv("EMBEDDING_MODEL"), llm_model=os.getenv("LLM_MODEL"), - top_k=int(os.getenv("TOP_K")), - temperature=float(os.getenv("TEMPERATURE")), + top_k=int(os.getenv("TOP_K", "5")), + temperature=float(os.getenv("TEMPERATURE", "0.7")), collection_name=os.getenv("CHROMA_COLLECTION"), host=os.getenv("CHROMA_HOST"), port=os.getenv("CHROMA_PORT"), @@ -110,142 +122,128 @@ def main(): embedding_api_key=os.getenv("OPENAI_API_KEY"), ) + # Check for required arguments + if len(sys.argv) < 3: + print("Usage: python rag_evaluator.py ") + print("Example: python rag_evaluator.py CRAGPipeline rag_eval_CRAGPipeline") + sys.exit(1) + + # Get pipeline name and experiment name from arguments + pipeline_name = sys.argv[1] + experiment_name = sys.argv[2] + + print(f"Using pipeline: {pipeline_name}") + print(f"Using experiment name: {experiment_name}") + rag_config_path = os.path.abspath( os.path.join( - os.path.dirname(__file__), "..", "backend", "app", "rag", sys.argv[1] + ".py" + os.path.dirname(__file__), "..", "backend", "app", "rag", pipeline_name + ".py" ) ) - # Handle different argument scenarios - if len(sys.argv) == 3: - - # Original 2-argument scenario (pipeline name and run name) - Pipeline = get_class_from_input(rag_config_path, sys.argv[1]) - print(Pipeline) - - if Pipeline: - # Initialize RAG pipeline - rag_pipeline = Pipeline(config) - - # Initialize evaluator - evaluator = RAGEvaluator(test_dataset_path, rag_pipeline) - - # Run evaluation - results = evaluator.run_full_evaluation(sys.argv[2]) + try: + # Load the pipeline class + Pipeline = get_class_from_input(rag_config_path, pipeline_name) + print(f"Successfully loaded pipeline: {pipeline_name}") + + # Initialize RAG pipeline + rag_pipeline = Pipeline(config) + + # Initialize evaluator + evaluator = RAGEvaluator(test_dataset_path, rag_pipeline) + + # Always run full evaluation with provided experiment name + print(f"Running full evaluation with experiment name: {experiment_name}") + results = evaluator.run_full_evaluation(experiment_name) + + print("\nEvaluation Results:") + print(json.dumps(results, indent=2)) + + # Run bias analysis + print("\nRunning bias analysis...") + bias_analyzer = RAGBiasAnalyzer(evaluator) + bias_report = bias_analyzer.generate_bias_report(experiment_name=experiment_name) + print("\nBias Analysis Report:") + print(json.dumps(bias_report, indent=2)) + + # Save results to database if performance meets threshold + try: + thresholds = { + 'embedding_evaluation': { + 'retrieval_precision': 0.2, + 'retrieval_recall': 0.2, + 'retrieval_f1': 0.2 + }, + 'top_k_evaluation': { + 'top_k_1': 0.2, + 'top_k_3': 0.2, + 'top_k_5': 0.2 + }, + 'rag_system_evaluation': { + 'bleu_score': 0.2, + 'llm_judge_accuracy': 0.2, + 'llm_judge_relevance': 0.2, + 'llm_judge_completeness': 0.2 + } + } + llms = check_llm_results(results, thresholds) - print(results) + # Create database session + DB_USER = os.getenv("DB_USER") + DB_PASSWORD = os.getenv("DB_PASSWORD") + DB_HOST = os.getenv("DB_HOST") + DB_PORT = os.getenv("DB_PORT") + DB_NAME = os.getenv("DB_NAME") - # Save results to database if performance meets threshold - try: - thresholds = { - 'embedding_evaluation': { - 'retrieval_precision': 0.2, - 'retrieval_recall': 0.2, - 'retrieval_f1': 0.2 - }, - 'top_k_evaluation': { - 'top_k_1': 0.2, - 'top_k_3': 0.2, - 'top_k_5': 0.2 - }, - 'rag_evaluation': { - 'bleu_score': 0.2, - 'llm_judge_accuracy': 0.2, - 'llm_judge_relevance': 0.2, - 'llm_judge_completeness': 0.2 - } - } - llms = check_llm_results(results, thresholds) - # Create database session - DB_USER = os.getenv("DB_USER") - DB_PASSWORD = os.getenv("DB_PASSWORD") - DB_HOST = os.getenv("DB_HOST") - DB_PORT = os.getenv("DB_PORT") - DB_NAME = os.getenv("DB_NAME") + if all([DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME]): engine = create_engine(f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}") Session = sessionmaker(bind=engine) session = Session() # Check if RAG model already exists in database - existing_rag = session.query(RAG).filter_by(rag_name=sys.argv[1]).first() + existing_rag = session.query(RAG).filter_by(rag_name=pipeline_name).first() if llms["overall_pass"]: if existing_rag: # Update existing RAG model existing_rag.is_available = True session.commit() - print(f"Updated existing RAG model '{sys.argv[1]}' in database with True availability") + print(f"Updated existing RAG model '{pipeline_name}' in database with True availability") else: # Add new RAG model to database new_rag = RAG( rag_id=uuid.uuid4(), - rag_name=sys.argv[1], + rag_name=pipeline_name, is_available=True ) session.add(new_rag) - print(f"RAG model '{sys.argv[1]}' added to database with True availability") + print(f"RAG model '{pipeline_name}' added to database with True availability") else: if existing_rag: # Update existing RAG model existing_rag.is_available = False session.commit() - print(f"Updated existing RAG model '{sys.argv[1]}' in database with False availability") + print(f"Updated existing RAG model '{pipeline_name}' in database with False availability") else: # Add new RAG model to database new_rag = RAG( rag_id=uuid.uuid4(), - rag_name=sys.argv[1], + rag_name=pipeline_name, is_available=False ) session.add(new_rag) - print(f"RAG model '{sys.argv[1]}' added to database with False availability") - session.commit() + print(f"RAG model '{pipeline_name}' added to database with False availability") session.close() - print("Evaluation Complete!") - except Exception as e: - print(f"Failed to save to database: {e}") - traceback.print_exc() - - elif len(sys.argv) == 4: - # New 3-argument scenario (pipeline name, eval run name, bias run name) - Pipeline = get_class_from_input(rag_config_path, sys.argv[1]) - print(Pipeline) - - if Pipeline: - # Initialize RAG pipeline - rag_pipeline = Pipeline(config) - - # Initialize evaluator - evaluator = RAGEvaluator(test_dataset_path, rag_pipeline) - - # Run full evaluation - eval_results = evaluator.run_full_evaluation(sys.argv[2]) - - # Initialize bias analyzer - bias_analyzer = RAGBiasAnalyzer(evaluator) - - # Run bias analysis - bias_report = bias_analyzer.generate_bias_report( - experiment_name=sys.argv[3] - ) - - # Optional: Save bias report to a file - with open("bias_analysis_report.json", "w") as f: - json.dump(bias_report, f, indent=4) - - print("Evaluation and Bias Analysis Complete!") - else: - print( - "Please provide the RAG pipeline name, evaluation run name, and bias analysis run name as command line arguments." - ) - else: - print("Usage:") - print( - "- For standard evaluation: python script.py " - ) - print( - "- For evaluation with bias analysis: python script.py " - ) - + else: + print("Warning: Database credentials not fully configured. Skipping database update.") + + except Exception as e: + print(f"Warning: Failed to update database: {str(e)}") + traceback.print_exc() + + except Exception as e: + print(f"Error: {str(e)}") + traceback.print_exc() + sys.exit(1) if __name__ == "__main__": main() From 204eb7ad633aa3e0812a140ae31a1e10c892c2a7 Mon Sep 17 00:00:00 2001 From: Rounak Bende <153109208+rounakbende10@users.noreply.github.com> Date: Mon, 21 Apr 2025 00:43:32 -0400 Subject: [PATCH 2/2] Create README.md --- rag_model/README.md | 173 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 173 insertions(+) create mode 100644 rag_model/README.md diff --git a/rag_model/README.md b/rag_model/README.md new file mode 100644 index 0000000..a638cef --- /dev/null +++ b/rag_model/README.md @@ -0,0 +1,173 @@ +# RAG Model Evaluation Framework + +This folder contains tools for evaluating Retrieval-Augmented Generation (RAG) models, including performance metrics, bias analysis, and database integration. + +## Overview + +The RAG Model Evaluation Framework provides a comprehensive suite of tools to evaluate the performance and fairness of RAG systems. It includes: + +- **RAGEvaluator**: Core evaluation engine for measuring RAG performance +- **RAGBiasAnalyzer**: Tool for detecting and analyzing biases in RAG responses +- **Database Integration**: Automatic storage of evaluation results in PostgreSQL + +## Components + +### RAGEvaluator + +The `RAGEvaluator` class is the main evaluation engine that measures the performance of RAG systems across multiple dimensions: + +- **Embedding Model Evaluation**: Measures the quality of document embeddings +- **Top-K Strategy Evaluation**: Evaluates the effectiveness of different retrieval strategies +- **RAG System Evaluation**: Assesses the overall performance of the RAG pipeline + +### RAGBiasAnalyzer + +The `RAGBiasAnalyzer` class analyzes potential biases in RAG responses by: + +- Detecting demographic biases +- Identifying topic biases +- Measuring response diversity +- Analyzing sentiment distribution + +### Database Integration + +Evaluation results are automatically stored in a PostgreSQL database when performance meets predefined thresholds. The database tracks: + +- RAG model availability +- Performance metrics +- Bias analysis results + +## Usage + +### Running the Evaluator + +To run a full evaluation of a RAG model: + +```bash +python rag_evaluator.py +``` + +Example: +```bash +python rag_evaluator.py CRAGPipeline rag_eval_CRAGPipeline +``` + +### Environment Variables + +The following environment variables must be set in your `.env` file: + +``` +# OpenAI API Keys +OPENAI_API_KEY=your_openai_api_key +GROQ_API_KEY=your_groq_api_key + +# Model Configuration +EMBEDDING_MODEL=text-embedding-ada-002 +LLM_MODEL=llama3-70b-8192 +TOP_K=5 +TEMPERATURE=0.7 + +# ChromaDB Configuration +CHROMA_COLLECTION=your_collection_name +CHROMA_HOST=your_chroma_host +CHROMA_PORT=your_chroma_port + +# Test Dataset +TEST_DATASET_PATH=path/to/your/test_dataset.json + +# Database Configuration +DB_USER=your_db_user +DB_PASSWORD=your_db_password +DB_HOST=your_db_host +DB_PORT=your_db_port +DB_NAME=your_db_name + +# MLflow Configuration +MLFLOW_TRACKING_URI=your_mlflow_tracking_uri +``` + +### Test Dataset Format + +The test dataset should be a JSON file with the following structure: + +```json +[ + { + "query": "What is the capital of France?", + "ground_truth": "The capital of France is Paris.", + "relevant_docs": ["doc1", "doc2"], + "topic": "geography", + "industry": "education" + }, + ... +] +``` + +## Evaluation Metrics + +### Embedding Model Evaluation + +- **Retrieval Precision**: Proportion of retrieved documents that are relevant +- **Retrieval Recall**: Proportion of relevant documents that are retrieved +- **Retrieval F1**: Harmonic mean of precision and recall + +### Top-K Strategy Evaluation + +- **Answer Accuracy**: Measures how well the retrieved documents support the correct answer +- **Relevance Score**: Measures the relevance of retrieved documents to the query + +### RAG System Evaluation + +- **BLEU Score**: Measures the similarity between generated and ground truth responses +- **LLM Judge Accuracy**: Human-like evaluation of answer correctness +- **LLM Judge Relevance**: Human-like evaluation of answer relevance +- **LLM Judge Completeness**: Human-like evaluation of answer completeness + +### Bias Analysis + +- **Demographic Bias**: Measures bias across demographic groups +- **Topic Bias**: Measures bias across different topics +- **Response Diversity**: Measures the diversity of generated responses +- **Sentiment Distribution**: Analyzes the sentiment distribution of responses + +## Performance Thresholds + +The following thresholds are used to determine if a RAG model meets performance requirements: + +```python +thresholds = { + 'embedding_evaluation': { + 'retrieval_precision': 0.2, + 'retrieval_recall': 0.2, + 'retrieval_f1': 0.2 + }, + 'top_k_evaluation': { + 'top_k_1': 0.2, + 'top_k_3': 0.2, + 'top_k_5': 0.2 + }, + 'rag_system_evaluation': { + 'bleu_score': 0.2, + 'llm_judge_accuracy': 0.2, + 'llm_judge_relevance': 0.2, + 'llm_judge_completeness': 0.2 + } +} +``` + +## Dependencies + +- Python 3.8+ +- OpenAI API +- Groq API +- ChromaDB +- MLflow +- PostgreSQL +- SQLAlchemy +- NumPy +- Pandas +- scikit-learn + +## License + +This project is licensed under the MIT License - see the LICENSE file for details. \ No newline at end of file