Building a RAG evaluation pipeline in LlamaIndex

Most RAG demos work until they don’t. This llamaindex rag evaluation pipeline tutorial shows you how to measure retrieval precision and answer faithfulness with code you can run in CI, not just in a notebook.

Prerequisites

You need Python 3.10+ and a working LLM API key. The examples use LlamaIndex 0.10.x and the OpenAI model interface, but any OpenAI-compatible endpoint works.

pip install llama-index llama-index-llms-openai pytest

You should have a small corpus of text files in ./data and a set of evaluation questions with known relevant document IDs. If you don’t, the snippet below generates a minimal one.

Project setup

Create eval_pipeline.py. First, configure the LLM and embed model. LlamaIndex uses Settings as a global context.

from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# Point at any OpenAI-compatible endpoint. n4n.ai fronts 240+ models with
# automatic fallback on provider degradation, so a single key covers many backends.
Settings.llm = OpenAI(model="gpt-4o-mini", api_base="https://api.n4n.ai/v1", api_key="YOUR_KEY")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", api_key="YOUR_KEY")

If you run OpenAI directly, drop the api_base argument.

Build the index and retriever

Load a directory of text files and build a vector index. For evaluation we need the retriever to return nodes with stable IDs.

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

docs = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(docs)
retriever = index.as_retriever(similarity_top_k=3)

Assign explicit node IDs so expected IDs are deterministic:

from llama_index.core.schema import Document

docs = [Document(text="LlamaIndex supports evaluation modules.", id_="doc-1"),
        Document(text="RAG pipelines need retrieval metrics.", id_="doc-2")]
index = VectorStoreIndex.from_documents(docs)

Define evaluation metrics

LlamaIndex ships RetrieverEvaluator for ranking quality and FaithfulnessEvaluator / AnswerRelevancyEvaluator for generated answers.

Retrieval evaluation

Use Mean Reciprocal Rank (MRR) and hit rate. The evaluator compares returned node IDs against a list of expected IDs.

from llama_index.core.evaluation import RetrieverEvaluator

retriever_eval = RetrieverEvaluator.from_metric("mrr", retriever=retriever)
retriever_eval_hit = RetrieverEvaluator.from_metric("hit_rate", retriever=retriever)

async def eval_retrieval(query, expected_ids):
    mrr = await retriever_eval.aevaluate(query=query, expected_ids=expected_ids)
    hit = await retriever_eval_hit.aevaluate(query=query, expected_ids=expected_ids)
    return mrr, hit

Response faithfulness and relevancy

Faithfulness checks whether the answer is grounded in retrieved context. Relevancy checks whether it addresses the query.

from llama_index.core.evaluation import FaithfulnessEvaluator, AnswerRelevancyEvaluator

faithfulness_eval = FaithfulnessEvaluator(llm=Settings.llm)
relevancy_eval = AnswerRelevancyEvaluator(llm=Settings.llm)

async def eval_response(query, response):
    faith = await faithfulness_eval.aevaluate_response(query=query, response=response)
    rel = await relevancy_eval.aevaluate_response(query=query, response=response)
    return faith, rel

Assemble the evaluation pipeline

Write a function that runs a query through the full RAG flow and collects all scores.

import asyncio
from llama_index.core.query_engine import RetrieverQueryEngine

query_engine = RetrieverQueryEngine.from_args(retriever)

async def run_pipeline(query, expected_ids):
    response = await query_engine.aquery(query)
    mrr, hit = await eval_retrieval(query, expected_ids)
    faith, rel = await eval_response(query, response)
    return {
        "query": query,
        "mrr": mrr.score,
        "hit_rate": hit.score,
        "faithfulness": faith.score,
        "relevancy": rel.score,
        "answer": str(response),
    }

if __name__ == "__main__":
    cases = [
        ("What does LlamaIndex support?", ["doc-1"]),
        ("Why are metrics needed in RAG?", ["doc-2"]),
    ]
    results = asyncio.run(asyncio.gather(*[run_pipeline(q, e) for q, e in cases]))
    for r in results:
        print(r)

Run inside pytest

CI should fail on regression. Create test_eval.py:

import asyncio
from eval_pipeline import run_pipeline

def test_retrieval_quality():
    results = asyncio.run(run_pipeline("What does LlamaIndex support?", ["doc-1"]))
    assert results["mrr"] >= 0.5
    assert results["hit_rate"] == 1.0
    assert results["faithfulness"] >= 0.8

Run:

pytest test_eval.py -q

Expected output

At the checkpoint in eval_pipeline.py, you should see a dict per query:

{'query': 'What does LlamaIndex support?', 'mrr': 1.0, 'hit_rate': 1.0, 'faithfulness': 1.0, 'relevancy': 1.0, 'answer': 'LlamaIndex supports evaluation modules.'}
{'query': 'Why are metrics needed in RAG?', 'mrr': 1.0, 'hit_rate': 1.0, 'faithfulness': 1.0, 'relevancy': 1.0, 'answer': 'RAG pipelines need retrieval metrics.'}

The scores are floats between 0 and 1. MRR of 1.0 means the first retrieved node was relevant. Faithfulness of 1.0 means the LLM judge found no hallucination.

If a provider is rate-limited, the OpenAI-compatible gateway you configured will return a fallback response or error; the pipeline will raise, and pytest will catch it. That is preferable to silently shipping broken retrieval.

Extending the pipeline

Swap from_metric("mrr") for "precision" or "recall" if your corpus has multiple relevant docs per query. For batch evaluation, persist results to JSON and track deltas across commits:

import json
with open("eval_results.json", "w") as f:
    json.dump(results, f, indent=2)

The llamaindex rag evaluation pipeline tutorial above is intentionally minimal. Wire it into a pre-merge hook and you get a real signal on whether your chunk size, embed model, or prompt actually moved the needle.