LLM Evaluation in Python Without Paying for SaaS: RAGAS, DeepEval, and Custom Metrics

pip install ragas datasets "langchain-community<0.4" 

from ragas import evaluate from ragas.metrics import ( context_precision, context_recall, faithfulness, answer_relevancy, ) from datasets import Dataset # Your RAG pipeline output — one row per question data = { "question": [ "What is the rate limit for the Anthropic API?", "How do I install pgvector in Python?", ], "answer": [ "The Anthropic API rate limit is 1000 requests per minute for most models.", "Install pgvector using pip: pip install pgvector", ], "contexts": [ # List of retrieved chunks for each question ["Anthropic Claude API limits: Tier 1 accounts get 50 RPM, Tier 4 get 4000 RPM..."], ["The pgvector Python package requires PostgreSQL 14+ and the vector extension..."], ], "ground_truth": [ "Anthropic API rate limits vary by tier: Tier 1 is 50 RPM, Tier 4 is 4000 RPM.", "Install pgvector with: pip install pgvector psycopg", ], } dataset = Dataset.from_dict(data) result = evaluate( dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy], ) print(result) # {'context_precision': 0.7200, 'context_recall': 0.8500, 'faithfulness': 0.9100, 'answer_relevancy': 0.8800} (example output) 

pip install deepeval 

from deepeval import evaluate from deepeval.metrics import ( AnswerRelevancyMetric, FaithfulnessMetric, HallucinationMetric, ToxicityMetric, ) from deepeval.test_case import LLMTestCase # Define your test cases test_case = LLMTestCase( input="Summarize the key risks of using RAG in production.", actual_output=""" Key production RAG risks: (1) retrieval failures when queries are ambiguous, (2) hallucination when retrieved context is outdated, (3) latency spikes from embedding + vector search in the critical path. """, expected_output=""" RAG production risks include retrieval failures, hallucination from stale context, and latency added by the embedding and search pipeline. """, context=[ "RAG systems fail when vector search returns irrelevant chunks...", "Embedding models can become stale if not updated when the domain changes...", ], ) # Run evaluation metrics = [ AnswerRelevancyMetric(threshold=0.7), FaithfulnessMetric(threshold=0.8), HallucinationMetric(threshold=0.5), ] evaluate([test_case], metrics) 

# test_llm_outputs.py import pytest from deepeval import assert_test from deepeval.metrics import AnswerRelevancyMetric, FaithfulnessMetric from deepeval.test_case import LLMTestCase from my_app import generate_answer, retrieve_context @pytest.mark.parametrize("question,expected", [ ("What is pgvector?", "pgvector is a PostgreSQL extension for vector similarity search"), ("How do I chunk documents?", "Document chunking splits text into segments for embedding"), ]) def test_answer_quality(question, expected): context = retrieve_context(question) answer = generate_answer(question, context) test_case = LLMTestCase( input=question, actual_output=answer, expected_output=expected, context=context, ) assert_test(test_case, [ AnswerRelevancyMetric(threshold=0.7), FaithfulnessMetric(threshold=0.8), ]) 

from dataclasses import dataclass from typing import Callable @dataclass class CustomMetric: name: str scorer: Callable[[str, list[str]], float] # (answer, context) -> 0.0-1.0 threshold: float = 0.8 def citation_coverage_scorer(answer: str, context: list[str]) -> float: """ Checks if the answer references information present in context. Returns: fraction of answer sentences that can be traced to context. """ import re sentences = [s.strip() for s in re.split(r'[.!?]', answer) if s.strip()] if not sentences: return 0.0 covered = 0 for sentence in sentences: # Simple keyword overlap — replace with semantic similarity for production sentence_words = set(sentence.lower().split()) for chunk in context: chunk_words = set(chunk.lower().split()) if len(sentence_words & chunk_words) / max(len(sentence_words), 1) > 0.4: covered += 1 break return covered / len(sentences) # Run against your golden dataset def evaluate_custom(test_cases: list[dict], metric: CustomMetric) -> dict: scores = [] failures = [] for tc in test_cases: score = metric.scorer(tc["answer"], tc["context"]) scores.append(score) if score < metric.threshold: failures.append({"input": tc["input"], "score": score}) avg = sum(scores) / len(scores) print(f"{metric.name}: avg={avg:.2f}, failures={len(failures)}/{len(test_cases)}") return {"avg": avg, "failures": failures} 

import json from pathlib import Path def build_golden_dataset_from_logs(log_file: str, n_samples: int = 100) -> list[dict]: """ Convert production query logs to a format suitable for RAGAS/DeepEval. """ with open(log_file) as f: logs = [json.loads(line) for line in f] # Take a representative sample sampled = logs[:n_samples] dataset = [] for entry in sampled: dataset.append({ "question": entry["user_query"], "answer": entry["llm_response"], "contexts": entry["retrieved_chunks"], "ground_truth": None, # Fill manually or with LLM-assisted labeling }) return dataset # Save for reuse dataset = build_golden_dataset_from_logs("production_queries.jsonl") Path("golden_dataset.json").write_text(json.dumps(dataset, indent=2)) 

#!/usr/bin/env python3 """ eval_pipeline.py — Run RAGAS evaluation against golden dataset. Exit code 0 = all metrics pass, 1 = failures. """ import json import sys from datasets import Dataset from ragas import evaluate from ragas.metrics import context_precision, faithfulness, answer_relevancy THRESHOLDS = { "context_precision": 0.70, "faithfulness": 0.80, "answer_relevancy": 0.75, } def main(): # Load golden dataset with open("golden_dataset.json") as f: data = json.load(f) dataset = Dataset.from_list(data) result = evaluate(dataset, metrics=[context_precision, faithfulness, answer_relevancy]) failures = [] for metric, threshold in THRESHOLDS.items(): # result[metric] is a list of per-row scores, so average it scores = result[metric] score = sum(scores) / len(scores) status = "PASS" if score >= threshold else "FAIL" print(f"{status} {metric}: {score:.2f} (threshold: {threshold})") if score < threshold: failures.append(metric) if failures: print(f"\nFailed metrics: {', '.join(failures)}") sys.exit(1) print("\nAll metrics passed.") if __name__ == "__main__": main()