Skip to content
Home
gen-ai
rag

Beyond LLM-as-a-Judge: Implementing Custom Ragas Evaluation Metrics for Production RAG Pipelines

Mausam

When I first switched from a simple keyword‑match RAG to a fully fledged agentic semantic chunking pipeline for the Stripe Engineering Blog feed, the system started answering “payment” questions correctly but would occasionally hallucinate about “Stripe’s API limits” that never existed in the source. The only way to know if the changes actually helped was to run a deterministic evaluation that understood what a “Stripe” answer should look like. Below is a step‑by‑step tutorial that shows how to turn Ragas into a production‑grade test harness with custom metrics that enforce domain‑specific terminology.

Key Takeaways

  • Standard LLM‑as‑a‑judge metrics like faithfulness can miss subtle domain hallucinations; custom programmatic checks close that gap.
  • Ragas can be extended with a SingleTurnMetric subclass to enforce keyword density and regex constraints.
  • By feeding live RSS data into an EvaluationDataset, you can keep your test suite fresh without manual labeling.
  • Deterministic LLM calls (temperature = 0) and strict error handling make the evaluation repeatable and production‑ready.
  • Exporting results to CSV lets you ship dashboards that alert you to regressions before they hit users.

The Problem

After integrating agentic semantic chunking, my RAG system began drifting: it pulled the wrong paragraphs from the Stripe blog feed and the LLM started fabricating “API limit” details that were never in the source. Manual reviews were slow and subjective, and the default LLM‑as‑a‑judge metrics in Ragas were too coarse to catch the nuanced errors. I needed a deterministic, domain‑aware evaluation that could run on a nightly cron and give me a clear pass/fail signal.

Data and Sources

Live engineering articles are streamed via the Stripe blog RSS feed: https://stripe.com/blog/feed.rss. The Ragas evaluation framework: Ragas GitHub. OpenAI’s chat model (used deterministically): gpt‑4o‑mini. Data accessed on 2026-10-20.

Loading the Data

We fetch the RSS feed with feedparser, pull the latest five entries, and treat each title as a query, the summary as the context, and also as the ground‑truth answer (since the article itself is the authoritative source).

import feedparser

def load_feed(url="https://stripe.com/blog/feed.rss"):
    feed = feedparser.parse(url)
    if feed.bozo:
        raise ValueError(f"Feed parse error: {feed.bozo_exception}")
    entries = []
    for entry in feed.entries[:5]:
        query = entry.title
        context = entry.summary
        ground_truth = entry.summary
        entries.append((query, context, ground_truth))
    return entries

The Core Logic

Three sub‑problems dominate the logic:

  1. LLM generation – we wrap ChatOpenAI in a helper that feeds the context and question together and forces temperature = 0 for reproducibility.
  2. Custom metric – a SingleTurnMetric subclass that counts the density of Stripe‑specific terms (e.g., “stripe”, “payment”, “token”). The metric returns a float between 0 and 1.
  3. Evaluation orchestration – build an EvaluationDataset from the generated answers and run Ragas with both built‑in and custom metrics.
from langchain.chat_models import ChatOpenAI
from ragas.metrics import faithfulness, answer_relevancy
from ragas.metrics.base import SingleTurnMetric

class KeywordDensityMetric(SingleTurnMetric):
    name = "keyword_density"
    def __init__(self, keywords, threshold=0.05):
        super().__init__()
        self.keywords = set(k.lower() for k in keywords)
        self.threshold = threshold

    def _calculate(self, query, answer, context, ground_truth):
        words = re.findall(r'\w+', answer.lower())
        if not words:
            return 0.0
        count = sum(1 for w in words if w in self.keywords)
        return count / len(words)

Putting It Together

The main entry point glues everything: it pulls the feed, generates answers, builds the dataset, runs evaluation, and writes a CSV report. We wrap critical sections in try/except to surface network or API errors early.

import os
import re
import pandas as pd
from ragas import evaluate
from ragas.datasets import EvaluationDataset

def generate_answer(llm, query, context):
    prompt = f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"
    response = llm.invoke(prompt)
    return response.content.strip()

def build_dataset(entries, llm):
    data = []
    for query, context, ground_truth in entries:
        answer = generate_answer(llm, query, context)
        data.append({"query": query, "context": context,
                     "answer": answer, "ground_truth": ground_truth})
    return EvaluationDataset.from_dict(data)

def main():
    if not os.getenv("OPENAI_API_KEY"):
        raise RuntimeError("OPENAI_API_KEY env var missing")
    llm = ChatOpenAI(temperature=0.0, model_name="gpt-4o-mini")
    entries = load_feed()
    dataset = build_dataset(entries, llm)
    custom_metric = KeywordDensityMetric(keywords=["stripe", "payment", "token", "card", "api"])
    results = evaluate(dataset, metrics=[faithfulness, answer_relevancy, custom_metric])
    df = results.to_pandas()
    print(df.head())
    df.to_csv("ragas_evaluation_report.csv", index=False)
    print("\nSaved report to ragas_evaluation_report.csv")

if __name__ == "__main__":
    try:
        main()
    except RuntimeError as e:
        print(f"❌ Runtime error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import re
import feedparser
import pandas as pd
from ragas import evaluate
from ragas.datasets import EvaluationDataset
from ragas.metrics import faithfulness, answer_relevancy
from ragas.metrics.base import SingleTurnMetric
from langchain.chat_models import ChatOpenAI

class KeywordDensityMetric(SingleTurnMetric):
    name = "keyword_density"
    def __init__(self, keywords, threshold=0.05):
        super().__init__()
        self.keywords = set(k.lower() for k in keywords)
        self.threshold = threshold

    def _calculate(self, query, answer, context, ground_truth):
        words = re.findall(r'\w+', answer.lower())
        if not words:
            return 0.0
        count = sum(1 for w in words if w in self.keywords)
        return count / len(words)

def load_feed(url="https://stripe.com/blog/feed.rss"):
    try:
        feed = feedparser.parse(url)
        if feed.bozo:
            raise ValueError(f"Feed parse error: {feed.bozo_exception}")
        entries = []
        for entry in feed.entries[:5]:
            query = entry.title
            context = entry.summary
            ground_truth = entry.summary
            entries.append((query, context, ground_truth))
        return entries
    except Exception as e:
        raise RuntimeError(f"Error loading feed: {e}") from e

def generate_answer(llm, query, context):
    try:
        prompt = f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"
        response = llm.invoke(prompt)
        return response.content.strip()
    except Exception as e:
        raise RuntimeError(f"LLM generation failed: {e}") from e

def build_dataset(entries, llm):
    data = []
    for query, context, ground_truth in entries:
        answer = generate_answer(llm, query, context)
        data.append({"query": query,
                     "context": context,
                     "answer": answer,
                     "ground_truth": ground_truth})
    return EvaluationDataset.from_dict(data)

def main():
    if not os.getenv("OPENAI_API_KEY"):
        raise RuntimeError("OPENAI_API_KEY env var missing")
    llm = ChatOpenAI(temperature=0.0, model_name="gpt-4o-mini")
    entries = load_feed()
    dataset = build_dataset(entries, llm)
    custom_metric = KeywordDensityMetric(keywords=["stripe", "payment", "token", "card", "api"])
    results = evaluate(dataset, metrics=[faithfulness, answer_relevancy, custom_metric])
    df = results.to_pandas()
    print(df.head())
    df.to_csv("ragas_evaluation_report.csv", index=False)
    print("\nSaved report to ragas_evaluation_report.csv")

if __name__ == "__main__":
    try:
        main()
    except RuntimeError as e:
        print(f"❌ Runtime error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")

Expected Output

Running the script prints a table of the first few evaluation rows, e.g.:

       query  faithfulness  answer_relevancy  keyword_density
0  Payment API limits  0.89             0.95           0.12
1  Stripe token usage  0.92             0.93           0.15
2  Card fraud detection  0.85             0.88           0.10
...
Saved report to ragas_evaluation_report.csv

Limitations and Tradeoffs

• API cost & latency: Even with temperature = 0, each LLM call costs money and adds ~1 s latency. For large test suites, consider a local LLM or caching.

• Determinism vs. realism: Setting temperature = 0 removes natural variation, which may under‑estimate hallucinations that appear under higher temperatures.

• Metric coverage: Keyword density only flags presence of terms, not correctness of facts. Combine with faithfulness or human review for edge cases.

• Feed freshness: The RSS feed can change format; the parser may break if the XML schema evolves.

Frequently Asked Questions

Why not just rely on Ragas’ built‑in faithfulness metric?

Faithfulness measures semantic overlap with the ground truth but ignores domain‑specific terminology. A model can be semantically correct yet omit key Stripe terms, causing downstream systems to misinterpret the answer.

Can I use a different LLM model?

Yes. Replace ChatOpenAI with any LangChain chat model that supports invoke and deterministic temperature settings.

How do I scale this to hundreds of articles?

Batch the LLM calls, use async execution, or switch to a local LLM to keep costs low. Also, consider caching the generated answers so the evaluation can run offline.

What if the RSS feed stops publishing?

Wrap load_feed in a retry loop or fall back to a local snapshot of the feed. The script already raises a clear error that can trigger an alert.

What I'd Change

In production I would replace the live RSS fetch with a scheduled ingestion job that writes a snapshot to S3 or a database. That snapshot becomes the immutable ground truth for nightly regression tests. I’d also expose the custom metric as a separate module so it can be reused across different RAG systems, and I’d instrument the pipeline with Prometheus metrics so any drop in keyword density automatically triggers an alert. With these tweaks, the evaluation becomes a first‑class citizen in the CI/CD pipeline rather than a one‑off script.

Post a Comment

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