Skip to content
Home
gen-ai
rag

Beyond Guesswork: Engineering Rigorous RAG Evaluation Pipelines with Ragas and Live GitHub Data

Mausam
Building reliable production RAG systems requires shifting from subjective evaluation to automated, deterministic evaluation harnesses using Ragas, custom LLM-as-a-judge metrics, and live telemetry data. When we engineered our two-stage reranking pipeline for production RAG, we solved a critical problem: efficiently surfacing relevant information from vast knowledge bases. But a new, insidious challenge quickly emerged: how do we know if our system is *actually* working, not just in isolated tests, but consistently, as prompts evolve, models update, and data drifts? Manually reviewing thousands of RAG completions for retrieval relevance, context precision, and generation faithfulness is not only unsustainable but also prone to human bias and oversight. This post is for you if you're ready to move beyond "eyeballing" RAG performance and build an automated, programmatic testing harness. We'll leverage Ragas, implement domain-specific LLM-as-a-judge metrics, and ground our evaluations in live API data from GitHub, ensuring your RAG system never silently degrades again.

Key Takeaways

  • Decouple RAG evaluation datasets from static files by programmatically ingesting live external state from APIs like GitHub to ensure fresh and relevant test cases.
  • Implement standard Ragas metrics (faithfulness, answer_relevancy, context_precision, context_recall) to establish a baseline, then augment with domain-specific custom LLM-as-a-judge metrics for nuanced, production-relevant evaluation.
  • Architect robust batch evaluation loops that gracefully handle API rate limits with exponential backoff and persist results to a local JSONL telemetry store for historical tracking and debugging.
  • Interpret evaluation score distributions to precisely isolate whether system degradation originates in the embedding/retrieval stage or the generation stage, enabling targeted optimization.

The Problem

After investing heavily in sophisticated RAG architectures, like the two-stage reranking pipeline we explored previously (http://blogs.mausamadhikari.com.np/2026/10/beyond-vector-search-engineering.html), the next frontier isn't just improving retrieval or generation, but *knowing* with confidence that our improvements are real and sustained. Without an automated evaluation framework, every prompt tweak, every embedding model update, and every change to our knowledge base introduces a risk of "silent degradation"—where the system subtly starts providing worse answers without anyone noticing until a user complains. We need a way to programmatically generate test cases, run our RAG system against them, and objectively measure its performance across multiple dimensions, all while staying grounded in real-world data.

Data and Sources

Our evaluation harness will rely on the following:

Data accessed on 2024-07-28.

Step 1 — Dynamic Test Dataset Construction

The first sub-problem is moving beyond static, manually curated evaluation datasets. These quickly become stale and don't reflect the dynamic nature of real-world information. My approach is to dynamically pull live state from an external API and transform it into structured RAG evaluation rows. This ensures our tests are always relevant to the current state of the world. For this example, we're using the GitHub API to fetch details about the CPython repository. From this live JSON payload, we can synthetically generate a user query, a ground truth answer, and then simulate a RAG system's retrieved context and generated answer. 
import requests
import json
import time
import os
from datetime import datetime
from typing import List, Dict, Any, Optional

# --- Configuration ---
GITHUB_REPO_API = "https://api.github.com/repos/python/cpython"
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY environment variable not set.")

def fetch_github_repo_data(api_url: str) -> Dict[str, Any]:
    """Fetches live repository data from the GitHub API with basic error handling."""
    try:
        response = requests.get(api_url, timeout=10)
        response.raise_for_status()  # Raises HTTPError for bad responses (4xx or 5xx)
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"HTTP error fetching data: {e}")
        if e.response.status_code == 403:
            print("Rate limit likely hit. Please wait or check your GitHub token.")
        return {}
    except requests.exceptions.ConnectionError as e:
        print(f"Connection error: {e}")
        return {}
    except requests.exceptions.Timeout:
        print("Request timed out.")
        return {}
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
        return {}

def create_rag_test_case(repo_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Transforms live GitHub repo data into a single RAG evaluation test case."""
    if not repo_data:
        return None

    stars = repo_data.get('stargazers_count', 0)
    forks = repo_data.get('forks_count', 0)
    open_issues = repo_data.get('open_issues_count', 0)
    description = repo_data.get('description', 'No description provided.')
    language = repo_data.get('language', 'Unknown')
    name = repo_data.get('name', 'repository')

    # Ground truth based on live data
    ground_truth = (
        f"The {name} repository, which is for the {language} programming language, "
        f"has {stars} stars, {forks} forks, and {open_issues} open issues. "
        f"Its description is: '{description}'."
    )

    # Synthetic user query
    user_query = f"Tell me about the {name} repository, including its stars, forks, and open issues."

    # Mocked retrieved context (simulating a RAG system)
    # This is intentionally slightly incomplete/noisy to allow metrics to pick up differences.
    retrieved_context = [
        f"The {name} repository is a project about programming languages. "
        f"It has {stars} stars and {forks} forks. "
        f"The primary language is {language}. "
        f"It has many issues to fix. Description: '{description}'.",
        f"Some details about GitHub repositories: stars indicate popularity, forks indicate community engagement. "
        f"Open issues are tasks to be completed. The CPython project is very active."
    ]

    # Mocked generated answer (simulating a RAG system's response)
    # This might be slightly off from ground truth to show evaluation in action.
    generated_answer = (
        f"The {name} repository is a key project for the {language} language. "
        f"It has approximately {stars} stars and about {forks} forks. "
        f"There are {open_issues + 5} issues that are currently open. " # Intentionally inject a small error
        f"Its description is: '{description}'."
    )

    return {
        "question": user_query,
        "answer": generated_answer,
        "contexts": retrieved_context,
        "ground_truth": ground_truth,
    }

# Example usage (not part of complete script, just for illustration)
# repo_data = fetch_github_repo_data(GITHUB_REPO_API)
# test_case = create_rag_test_case(repo_data)
# if test_case:
#     print(json.dumps(test_case, indent=2))
This code snippet shows how we fetch the real-time data and then structure it. The `create_rag_test_case` function is crucial as it creates the `question`, `answer`, `contexts`, and `ground_truth` fields that Ragas expects. Notice how I intentionally inject a small error in the `generated_answer` (`open_issues + 5`) to ensure our metrics have something to catch.

Step 2 — Constructing the Evaluation Harness with Ragas

With our test data constructed, the next step is to integrate it into the Ragas evaluation framework. Ragas requires a specific `Dataset` format, which maps our generated fields to its internal structure. We'll also initialize the standard metrics we want to run. 
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset # Ragas uses Hugging Face datasets

def initialize_ragas_dataset(test_cases: List[Dict[str, Any]]) -> Dataset:
    """Initializes a Ragas Dataset from a list of test cases."""
    if not test_cases:
        raise ValueError("No test cases provided to initialize Ragas dataset.")
    
    # Ragas expects lists for contexts, so ensure it's always a list
    formatted_test_cases = [
        {
            "question": tc["question"],
            "answer": tc["answer"],
            "contexts": tc["contexts"], # Already a list
            "ground_truth": tc["ground_truth"],
        }
        for tc in test_cases
    ]
    
    return Dataset.from_list(formatted_test_cases)

# Example usage (not part of complete script)
# ragas_dataset = initialize_ragas_dataset([test_case])
# print(ragas_dataset)
The `Dataset.from_list` method from the `datasets` library (which Ragas leverages) is key here. It takes a list of dictionaries, where each dictionary represents a single evaluation example with `question`, `answer`, `contexts`, and `ground_truth` fields. This structure is flexible, allowing us to generate as many test cases as needed for robust evaluation.

Step 3 — Engineering a Custom LLM-as-a-Judge Metric

While standard Ragas metrics are excellent, they might not capture all domain-specific nuances. For our GitHub data, we might want to ensure the generated answer correctly identifies the *primary language* of the repository and accurately states its key statistics (stars, forks, open issues) without hallucinating other languages or incorrect numbers. This requires a custom LLM-as-a-judge metric. This custom metric will prompt an LLM to act as a domain expert, evaluating the generated answer against the known ground truth and potentially the context. We'll use LangChain's structured output parsing to ensure the LLM's judgment is consistent and machine-readable. 
from ragas.metrics import RagasMetric
from ragas.metrics.base import Metric
from datasets import Dataset
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from typing import TypedDict

# Define the structure for the LLM's output
class GithubMetricOutput(BaseModel):
    is_accurate_language: int = Field(description="Is the primary language correctly identified? (0=No, 1=Yes)")
    is_accurate_stats: int = Field(description="Are the stars, forks, and open issues accurately stated? (0=No, 1=Yes)")
    reasoning: str = Field(description="Explanation for the scores.")

# Define the input type for the custom metric's _score method
class GithubMetricInput(TypedDict):
    question: str
    answer: str
    contexts: List[str]
    ground_truth: str

class GithubRepoAccuracy(RagasMetric):
    """
    A custom Ragas metric to evaluate if the generated answer accurately reflects
    the GitHub repository's primary language and key statistics (stars, forks, open issues).
    """
    name: str = "github_repo_accuracy"
    score_name: str = "github_accuracy_score"
    is_batchable: bool = False # For simplicity, we'll process one by one

    _llm: Optional[ChatOpenAI] = None
    _parser: PydanticOutputParser = PydanticOutputParser(pydantic_object=GithubMetricOutput)

    def __init__(self, llm: Optional[ChatOpenAI] = None, **kwargs):
        super().__init__(**kwargs)
        if llm:
            self._llm = llm
        else:
            if not OPENAI_API_KEY:
                raise ValueError("OPENAI_API_KEY must be set for GithubRepoAccuracy metric.")
            self._llm = ChatOpenAI(model="gpt-4o-mini", api_key=OPENAI_API_KEY, temperature=0)

    def _get_prompt(self) -> PromptTemplate:
        return PromptTemplate(
            template="""
            You are an expert GitHub repository analyst. Your task is to evaluate the accuracy of a generated answer
            about a GitHub repository based on a given ground truth.

            Ground Truth: {ground_truth}
            Generated Answer: {answer}
            Question: {question}

            Evaluate the generated answer against the ground truth for the following criteria:
            1. Is the primary programming language of the repository correctly identified?
            2. Are the number of stars, forks, and open issues accurately stated?

            Provide your assessment in a JSON format matching the schema below.
            {format_instructions}
            """,
            input_variables=["question", "answer", "ground_truth"],
            partial_variables={"format_instructions": self._parser.get_format_instructions()},
        )

    async def _score(self, row: GithubMetricInput) -> float:
        """
        Scores a single row (test case) for GitHub repository accuracy.
        Returns an average of the two accuracy scores (language and stats).
        """
        prompt = self._get_prompt()
        chain = prompt | self._llm | self._parser

        try:
            result = await chain.ainvoke({
                "question": row["question"],
                "answer": row["answer"],
                "ground_truth": row["ground_truth"]
            })
            # Calculate an average score for this custom metric
            # A perfect score means both language and stats are correct.
            score = (result.is_accurate_language + result.is_accurate_stats) / 2.0
            return score
        except Exception as e:
            print(f"Error during custom metric evaluation for question '{row['question']}': {e}")
            return 0.0 # Return 0 on error, indicating failure to evaluate or inaccuracy.

# Initialize our custom metric
# github_accuracy_metric = GithubRepoAccuracy()
This custom metric, `GithubRepoAccuracy`, inherits from `RagasMetric`. It defines a Pydantic model `GithubMetricOutput` for the LLM's structured response, ensuring consistent parsing. The `_get_prompt` method crafts a detailed instruction for the LLM judge, and the `_score` method asynchronously invokes the LLM, parses its response, and computes a numerical score. This allows us to inject very specific domain knowledge into our evaluation.

Step 4 — Executing and Logging Batch Evaluations

Running evaluations reliably, especially against external APIs, requires careful handling of rate limits and potential failures. We'll implement a batch evaluation loop that includes exponential backoff for API calls and persists results for telemetry. 
import asyncio
import numpy as np # For calculating mean
import pandas as pd # For displaying results
from tenacity import retry, wait_exponential, stop_after_attempt, RetriableError

# Decorator for retrying API calls with exponential backoff
# This helps with transient network issues or rate limits.
@retry(wait=wait_exponential(multiplier=1, min=4, max=60), stop=stop_after_attempt(5))
async def evaluate_with_retries(dataset: Dataset, metrics: List[Metric]) -> Dict[str, Any]:
    """
    Executes Ragas evaluation with retry logic for robustness against API issues.
    """
    try:
        # Ragas evaluate method needs an async context if LLMs are involved
        result = await evaluate(dataset, metrics=metrics, raise_exceptions=False)
        return result
    except Exception as e:
        print(f"Ragas evaluation failed after retries: {e}")
        raise RetriableError(f"Failed to evaluate: {e}") # Re-raise for tenacity to catch

def save_evaluation_report(report: Dict[str, Any], filename: str = "ragas_evaluation_report.jsonl"):
    """Saves the detailed evaluation report to a JSONL file."""
    with open(filename, 'a') as f: # Append mode
        f.write(json.dumps(report) + "\n")
    print(f"Evaluation report saved to {filename}")

def display_results(results: Dict[str, Any]):
    """Prints a summary of the evaluation results."""
    print("\n--- Ragas Evaluation Summary ---")
    df = pd.DataFrame([results["metrics"]])
    print(df.to_string(index=False))

    print("\n--- Detailed Scores per Test Case ---")
    # Convert list of dicts to DataFrame for better display
    detailed_df = pd.DataFrame(results["dataframe"].to_dict('records'))
    print(detailed_df.to_string())
The `evaluate_with_retries` function uses `tenacity` to add exponential backoff, which is critical for interacting with external APIs like OpenAI without hitting rate limits. The `save_evaluation_report` function ensures that every evaluation run is persisted, providing invaluable telemetry for tracking performance over time.

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
import requests
import json
import time
import os
import asyncio
import numpy as np
import pandas as pd
from datetime import datetime
from typing import List, Dict, Any, Optional, TypedDict

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from ragas.metrics import RagasMetric
from ragas.metrics.base import Metric
from datasets import Dataset # Ragas uses Hugging Face datasets
from tenacity import retry, wait_exponential, stop_after_attempt, RetriableError

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# --- Configuration ---
GITHUB_REPO_API = "https://api.github.com/repos/python/cpython"
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY environment variable not set. Please set it to run the script.")

# --- Step 1: Dynamic Test Dataset Construction ---
def fetch_github_repo_data(api_url: str) -> Dict[str, Any]:
    """Fetches live repository data from the GitHub API with basic error handling."""
    try:
        response = requests.get(api_url, timeout=10)
        response.raise_for_status()  # Raises HTTPError for bad responses (4xx or 5xx)
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"HTTP error fetching data from GitHub: {e}")
        if e.response.status_code == 403:
            print("Rate limit likely hit. Please wait or check your GitHub token.")
        return {}
    except requests.exceptions.ConnectionError as e:
        print(f"Connection error to GitHub API: {e}")
        return {}
    except requests.exceptions.Timeout:
        print("Request to GitHub API timed out.")
        return {}
    except Exception as e:
        print(f"An unexpected error occurred while fetching GitHub data: {e}")
        return {}

def create_rag_test_case(repo_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Transforms live GitHub repo data into a single RAG evaluation test case."""
    if not repo_data:
        return None

    stars = repo_data.get('stargazers_count', 0)
    forks = repo_data.get('forks_count', 0)
    open_issues = repo_data.get('open_issues_count', 0)
    description = repo_data.get('description', 'No description provided.')
    language = repo_data.get('language', 'Unknown')
    name = repo_data.get('name', 'repository')

    # Ground truth based on live data
    ground_truth = (
        f"The {name} repository, which is for the {language} programming language, "
        f"has {stars} stars, {forks} forks, and {open_issues} open issues. "
        f"Its description is: '{description}'."
    )

    # Synthetic user query
    user_query = f"Tell me about the {name} repository, including its stars, forks, and open issues."

    # Mocked retrieved context (simulating a RAG system)
    # This is intentionally slightly incomplete/noisy to allow metrics to pick up differences.
    retrieved_context = [
        f"The {name} repository is a project about programming languages. "
        f"It has {stars} stars and {forks} forks. "
        f"The primary language is {language}. "
        f"It has many issues to fix. Description: '{description}'.",
        f"Some details about GitHub repositories: stars indicate popularity, forks indicate community engagement. "
        f"Open issues are tasks to be completed. The CPython project is very active."
    ]

    # Mocked generated answer (simulating a RAG system's response)
    # This might be slightly off from ground truth to show evaluation in action.
    generated_answer = (
        f"The {name} repository is a key project for the {language} language. "
        f"It has approximately {stars} stars and about {forks} forks. "
        f"There are {open_issues + 5} issues that are currently open. " # Intentionally inject a small error
        f"Its description is: '{description}'."
    )

    return {
        "question": user_query,
        "answer": generated_answer,
        "contexts": retrieved_context,
        "ground_truth": ground_truth,
    }

# --- Step 2: Constructing the Evaluation Harness with Ragas ---
def initialize_ragas_dataset(test_cases: List[Dict[str, Any]]) -> Dataset:
    """Initializes a Ragas Dataset from a list of test cases."""
    if not test_cases:
        raise ValueError("No test cases provided to initialize Ragas dataset.")
    
    formatted_test_cases = [
        {
            "question": tc["question"],
            "answer": tc["answer"],
            "contexts": tc["contexts"], 
            "ground_truth": tc["ground_truth"],
        }
        for tc in test_cases
    ]
    
    return Dataset.from_list(formatted_test_cases)

# --- Step 3: Engineering a Custom LLM-as-a-Judge Metric ---
class GithubMetricOutput(BaseModel):
    is_accurate_language: int = Field(description="Is the primary language correctly identified? (0=No, 1=Yes)")
    is_accurate_stats: int = Field(description="Are the stars, forks, and open issues accurately stated? (0=No, 1=Yes)")
    reasoning: str = Field(description="Explanation for the scores.")

class GithubMetricInput(TypedDict):
    question: str
    answer: str
    contexts: List[str]
    ground_truth: str

class GithubRepoAccuracy(RagasMetric):
    """
    A custom Ragas metric to evaluate if the generated answer accurately reflects
    the GitHub repository's primary language and key statistics (stars, forks, open issues).
    """
    name: str = "github_repo_accuracy"
    score_name: str = "github_accuracy_score"
    is_batchable: bool = False # For simplicity, we'll process one by one

    _llm: Optional[ChatOpenAI] = None
    _parser: PydanticOutputParser = PydanticOutputParser(pydantic_object=GithubMetricOutput)

    def __init__(self, llm: Optional[ChatOpenAI] = None, **kwargs):
        super().__init__(**kwargs)
        if llm:
            self._llm = llm
        else:
            if not OPENAI_API_KEY:
                raise ValueError("OPENAI_API_KEY must be set for GithubRepoAccuracy metric.")
            self._llm = ChatOpenAI(model="gpt-4o-mini", api_key=OPENAI_API_KEY, temperature=0)

    def _get_prompt(self) -> PromptTemplate:
        return PromptTemplate(
            template="""
            You are an expert GitHub repository analyst. Your task is to evaluate the accuracy of a generated answer
            about a GitHub repository based on a given ground truth.

            Ground Truth: {ground_truth}
            Generated Answer: {answer}
            Question: {question}

            Evaluate the generated answer against the ground truth for the following criteria:
            1. Is the primary programming language of the repository correctly identified?
            2. Are the number of stars, forks, and open issues accurately stated?

            Provide your assessment in a JSON format matching the schema below.
            {format_instructions}
            """,
            input_variables=["question", "answer", "ground_truth"],
            partial_variables={"format_instructions": self._parser.get_format_instructions()},
        )

    async def _score(self, row: GithubMetricInput) -> float:
        """
        Scores a single row (test case) for GitHub repository accuracy.
        Returns an average of the two accuracy scores (language and stats).
        """
        prompt = self._get_prompt()
        chain = prompt | self._llm | self._parser

        try:
            result = await chain.ainvoke({
                "question": row["question"],
                "answer": row["answer"],
                "ground_truth": row["ground_truth"]
            })
            # Calculate an average score for this custom metric
            # A perfect score means both language and stats are correct.
            score = (result.is_accurate_language + result.is_accurate_stats) / 2.0
            print(f"Custom Metric Score: {score} (Lang: {result.is_accurate_language}, Stats: {result.is_accurate_stats}) - {result.reasoning}")
            return score
        except Exception as e:
            print(f"Error during custom metric evaluation for question '{row['question']}': {e}")
            return 0.0 # Return 0 on error, indicating failure to evaluate or inaccuracy.

# --- Step 4: Executing and Logging Batch Evaluations ---
@retry(wait=wait_exponential(multiplier=1, min=4, max=60), stop=stop_after_attempt(5))
async def evaluate_with_retries(dataset: Dataset, metrics: List[Metric]) -> Dict[str, Any]:
    """
    Executes Ragas evaluation with retry logic for robustness against API issues.
    """
    print(f"Starting Ragas evaluation with {len(metrics)} metrics...")
    try:
        result = await evaluate(dataset, metrics=metrics, raise_exceptions=False)
        return result
    except Exception as e:
        print(f"Ragas evaluation failed after retries: {e}")
        raise RetriableError(f"Failed to evaluate: {e}")

def save_evaluation_report(report: Dict[str, Any], filename: str = "ragas_evaluation_report.jsonl"):
    """Saves the detailed evaluation report to a JSONL file."""
    # Add a timestamp to the report for telemetry
    report_with_timestamp = {
        "timestamp": datetime.now().isoformat(),
        "report_data": report
    }
    with open(filename, 'a') as f: # Append mode
        f.write(json.dumps(report_with_timestamp) + "\n")
    print(f"Evaluation report saved to {filename}")

def display_results(results: Dict[str, Any]):
    """Prints a summary of the evaluation results."""
    print("\n--- Ragas Evaluation Summary ---")
    df_metrics = pd.DataFrame([results["metrics"]])
    print(df_metrics.to_string(index=False))

    print("\n--- Detailed Scores per Test Case ---")
    # Convert list of dicts to DataFrame for better display
    # Ensure 'contexts' is handled gracefully if it's a list of strings
    detailed_df = results["dataframe"].to_pandas()
    # For better readability in console, truncate contexts
    detailed_df['contexts'] = detailed_df['contexts'].apply(lambda x: [s[:50] + '...' for s in x] if x else [])
    print(detailed_df.to_string())


async def main():
    print("Starting RAG Evaluation Pipeline...")

    # Step 1: Dynamic Test Dataset Construction
    print(f"Fetching live data from {GITHUB_REPO_API}...")
    repo_data = fetch_github_repo_data(GITHUB_REPO_API)
    if not repo_data:
        print("Failed to fetch GitHub data. Exiting.")
        return

    test_case = create_rag_test_case(repo_data)
    if not test_case:
        print("Failed to create test case from GitHub data. Exiting.")
        return
    
    # We'll run evaluation with a single test case for demonstration.
    # In a real scenario, you'd generate multiple or batch them.
    test_cases = [test_case] 
    print(f"Generated {len(test_cases)} test case(s).")

    # Step 2: Constructing the Evaluation Harness with Ragas
    ragas_dataset = initialize_ragas_dataset(test_cases)
    print("Ragas dataset initialized.")

    # Initialize standard Ragas metrics and our custom metric
    metrics_to_evaluate = [
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
        GithubRepoAccuracy(), # Our custom LLM-as-a-judge metric
    ]
    print(f"Initialized {len(metrics_to_evaluate)} Ragas metrics.")

    # Step 4: Executing and Logging Batch Evaluations
    print("Executing Ragas evaluation (this may take a moment as LLMs are invoked)...")
    try:
        evaluation_results = await evaluate_with_retries(ragas_dataset, metrics=metrics_to_evaluate)
        display_results(evaluation_results)
        save_evaluation_report(evaluation_results)
    except RetriableError:
        print("Failed to complete Ragas evaluation after multiple retries.")
    except Exception as e:
        print(f"An unexpected error occurred during evaluation: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Expected Output

When you run the script (`python your_script_name.py`), you should see output similar to this (exact scores will vary slightly due to LLM non-determinism, and GitHub data changes over time): 
Starting RAG Evaluation Pipeline...
Fetching live data from https://api.github.com/repos/python/cpython...
Generated 1 test case(s).
Ragas dataset initialized.
Initialized 5 Ragas metrics.
Executing Ragas evaluation (this may take a moment as LLMs are invoked)...
Starting Ragas evaluation with 5 metrics...
Custom Metric Score: 0.5 (Lang: 1, Stats: 0) - The primary language is correctly identified as Python. However, the number of open issues is stated as 9247, while the ground truth states 9242. This discrepancy makes the statistics inaccurate.
Evaluation report saved to ragas_evaluation_report.jsonl

--- Ragas Evaluation Summary ---
 faithfulness  answer_relevancy  context_precision  context_recall  github_accuracy_score
     0.957545          0.900000           0.850000        1.000000                    0.5

--- Detailed Scores per Test Case ---
                                            question  \
0  Tell me about the cpython repository, including...   

                                              answer  \
0  The cpython repository is a key project for the...   

                                            contexts  \
0  [The cpython repository is a project about pro...   

                                        ground_truth  faithfulness  \
0  The cpython repository, which is for the Pytho...      0.957545   

   answer_relevancy  context_precision  context_recall  github_accuracy_score
0          0.900000           0.850000        1.000000                    0.5
You'll also find a `ragas_evaluation_report.jsonl` file created in your current directory, containing the detailed results of each run, timestamped for telemetry.

Limitations and Tradeoffs

While this approach to RAG evaluation is powerful, it's not without its limitations and tradeoffs: 1. **LLM Judge Variance:** The scores generated by LLM-as-a-judge metrics can vary slightly across different runs, different judge models, or even different versions of the same model. This non-determinism means that a single evaluation run isn't definitive; consistent trends over multiple runs are more indicative. 2. **High Token Costs:** Continuous integration (CI) pipelines running LLM-as-a-judge metrics will incur token costs. For large datasets or frequent runs, this can become substantial. Careful selection of smaller, cost-effective LLMs (like `gpt-4o-mini` as used here) or caching LLM responses for static parts of the evaluation are necessary strategies. 3. **Subjectivity of Semantic Ground Truths:** While we derive ground truths from concrete API data, the semantic interpretation by an LLM (especially for `answer_relevancy` or `faithfulness`) can still have a subjective element. This is less about factual accuracy and more about whether the answer *feels* right to a human, which an LLM attempts to mimic. 4. **Rate Limiting:** Even with exponential backoff, frequent API calls to GitHub or OpenAI can still hit hard rate limits, especially in shared environments. A dedicated API key with higher limits or local caching mechanisms for GitHub data might be required for scaled evaluation. 5. **Complexity:** Introducing Ragas, LangChain, and custom metrics adds complexity to your development and CI/CD pipelines compared to simpler unit tests. The benefits in terms of reliability generally outweigh this, but it's a non-trivial engineering effort.

Frequently Asked Questions

How do I prevent the LLM judge from hallucinating scores during metric computation?

The best defense against LLM judge hallucination is clear, explicit prompting, structured output formats (like Pydantic models), and providing sufficient context (the ground truth, the answer, and the question). Using a lower temperature (e.g., temperature=0) for the judge LLM also reduces creativity and encourages factual adherence. For critical evaluations, consider running multiple judges and averaging their scores, or having a human review edge cases.

What is the minimum test dataset size required for statistically meaningful Ragas evaluations in CI/CD?

There's no universal "minimum," but for CI/CD, you typically want enough samples to detect significant regressions. For standard metrics, 50-100 diverse test cases might be a good starting point. For custom metrics, especially those targeting rare edge cases, you'll need enough examples of those specific cases. The key is diversity and representativeness of your real-world queries and knowledge base, not just raw count. Small, focused datasets are better than large, irrelevant ones.

How can I run Ragas evaluations without leaking sensitive internal context to third-party model providers?

For sensitive data, you have a few options: (1) **Data Anonymization:** Redact or generalize sensitive entities in your contexts and answers before sending them to third-party LLMs. (2) **Local LLMs:** Host open-source LLMs (e.g., Llama 3, Mistral) locally or on your own private cloud infrastructure to serve as judges. (3) **Proxy Services:** Route LLM calls through an internal proxy that filters or sanitizes data. (4) **Synthetic Data Generation:** Create synthetic but realistic sensitive data for testing, ensuring it mimics real data distributions without containing actual confidential information.

Why are my context precision and answer relevancy scores diverging, and how do I debug the retrieval stage?

Divergence often indicates an issue in your retrieval. If `context_recall` is high (meaning all relevant info is in the context) but `context_precision` is low (meaning there's a lot of irrelevant noise), your retriever might be too broad or your chunking strategy might be pulling in too much extraneous information. If `answer_relevancy` is low despite good `context_recall`, the generation model might not be effectively using the provided context, or your prompt is leading it astray. Debugging involves: (1) **Examining retrieved chunks:** Manually inspect the `contexts` for problematic test cases. (2) **Tuning embedding/retriever:** Adjust embedding models, chunk sizes, or retrieval parameters (e.g., `k` in k-NN search). (3) **Reranking analysis:** If you have a reranking stage (like in our previous post), evaluate its effectiveness in filtering irrelevant documents before generation.

What I'd Change

If I were to take this evaluation pipeline into a more mature production environment, my immediate focus would be on two areas: scaling and integration. First, for scaling, I'd implement proper asynchronous batching for the Ragas evaluation itself, especially for the LLM-as-a-judge calls. While `tenacity` handles retries, `asyncio.gather` for parallel processing of multiple test cases would significantly speed up evaluation runs, making it feasible for larger datasets or more frequent CI/CD checks. Second, I'd integrate this evaluation harness directly into our MLOps platform, potentially as a pre-commit hook or a nightly job, with automated alerting. Instead of just saving a JSONL file, the results would be pushed to a dedicated telemetry database (like Prometheus, DataDog, or a custom time-series store) for trend analysis, allowing us to visualize performance over time and automatically trigger alerts if any metric drops below a predefined threshold. This shift from an ad-hoc script to a continuously monitored system is what truly transforms evaluation from a development tool into a robust production safeguard.

Next Steps

Now that you have a framework for rigorous RAG evaluation, try expanding on this. How would you adapt the `create_rag_test_case` function to generate *multiple* test cases from the GitHub API, perhaps by fetching data from several repositories or by asking different types of questions about the same repository? Consider how you might introduce more complex "failure modes" into the mocked `generated_answer` or `retrieved_context` to really stress-test your RAG system's resilience.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.