LlamaIndex batch evaluation runner explained

The LlamaIndex batch evaluation runner lets you score hundreds of queries against your RAG pipeline without writing custom loops. It handles concurrency, retries, and result aggregation out of the box — but the defaults can hide latency spikes and cost overruns if you don’t configure them deliberately. This guide walks through a production-ready setup, from dataset construction to metric selection and CI integration.

Build a reusable evaluation dataset

Evaluation starts with a dataset that reflects real traffic, not hand-crafted golden examples. Pull 200–500 actual user queries from your logs, strip PII, and pair each with the expected answer or at least a relevant document ID. Store as JSONL so you can version it alongside code.

{"query": "How do I reset my API key?", "expected_doc_ids": ["doc-12", "doc-45"], "category": "auth"}
{"query": "What's the rate limit for the embeddings endpoint?", "expected_doc_ids": ["doc-7"], "category": "billing"}
{"query": "Can I use GPT-4o with n4n.ai?", "expected_doc_ids": ["doc-3", "doc-9"], "category": "models"}

Load it into a LabelledEvaluatorDataset — this is the format the batch runner expects:

from llama_index.core.evaluation import LabelledEvaluatorDataset
from llama_index.core.schema import Document

def load_dataset(path: str) -> LabelledEvaluatorDataset:
    queries = []
    with open(path) as f:
        for line in f:
            item = json.loads(line)
            queries.append({
                "query": item["query"],
                "reference": item.get("expected_answer", ""),
                "reference_doc_ids": item.get("expected_doc_ids", []),
                "metadata": {"category": item.get("category", "general")}
            })
    return LabelledEvaluatorDataset.from_list(queries)

Pitfall: using synthetic queries from an LLM. They lack the ambiguity, typos, and domain-specific phrasing of real traffic. If you must augment, use an LLM to paraphrase existing queries — not to invent new ones.

Choose metrics that match your failure modes

LlamaIndex ships with several evaluators. Pick based on what breaks in production:

Evaluator	What it measures	When to use
`FaithfulnessEvaluator`	Hallucination rate (answer grounded in context)	RAG pipelines where hallucination is costly
`RelevancyEvaluator`	Answer relevance to query	General-purpose quality gate
`CorrectnessEvaluator`	Semantic match to reference answer	When you have ground-truth answers
`HitRateEvaluator` / `MRREvaluator`	Retrieval quality	Debugging retrieval separately from generation
`PairwiseComparisonEvaluator`	A/B between two pipelines	Model or prompt migrations

For a typical RAG system, run faithfulness + hit rate + MRR. Faithfulness catches generator drift; hit rate and MRR catch retriever regressions.

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    HitRateEvaluator,
    MRREvaluator,
)
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini", temperature=0.0)

evaluators = {
    "faithfulness": FaithfulnessEvaluator(llm=llm),
    "relevancy": RelevancyEvaluator(llm=llm),
    "hit_rate": HitRateEvaluator(),
    "mrr": MRREvaluator(),
}

Tradeoff: LLM-as-judge evaluators (faithfulness, relevancy, correctness) add latency and cost. A 500-query batch with gpt-4o-mini takes 3–5 minutes and costs ~$0.50. Retrieval-only metrics are near-instant and free. Run LLM judges on a sampled subset in CI; run the full suite nightly.

Configure the batch runner for production

The BatchEvalRunner accepts concurrency, retry, and timeout settings that matter at scale. Defaults are conservative: 10 concurrent workers, 3 retries, 60-second timeout. Tune these for your provider’s rate limits and your SLA.

from llama_index.core.evaluation import BatchEvalRunner
from llama_index.core.evaluation import EvaluationResult

runner = BatchEvalRunner(
    evaluators=evaluators,
    workers=20,                    # match your provider's concurrent request limit
    max_retries=3,
    timeout=120.0,                 # seconds per query; increase for slow retrievers
    show_progress=True,
)

Critical: set workers below your provider’s rate limit. If you’re routing through a gateway that enforces per-model limits (like n4n.ai does for its 240+ models), exceed that limit and you’ll get 429s that burn retries. Start at 50% of the documented limit and scale up after observing actual throughput.

Run the evaluation:

dataset = load_dataset("eval/queries.jsonl")
results = await runner.aevaluate_dataset(dataset)

The runner returns a Dict[str, List[EvaluationResult]] keyed by evaluator name. Each EvaluationResult contains query, response, passing (bool), score (float), and feedback (str).

Aggregate and interpret results

Raw results are noisy. Aggregate by category, percentile, and trend.

import pandas as pd
from collections import defaultdict

def summarize(results: dict[str, list[EvaluationResult]]) -> pd.DataFrame:
    rows = []
    for eval_name, eval_results in results.items():
        for r in eval_results:
            rows.append({
                "evaluator": eval_name,
                "query": r.query,
                "passing": r.passing,
                "score": r.score or 0.0,
                "category": r.metadata.get("category", "unknown") if r.metadata else "unknown",
            })
    df = pd.DataFrame(rows)
    
    # Overall pass rate per evaluator
    overall = df.groupby("evaluator")["passing"].mean().reset_index()
    overall.columns = ["evaluator", "pass_rate"]
    
    # Per-category breakdown
    by_cat = df.groupby(["evaluator", "category"])["passing"].mean().reset_index()
    by_cat.columns = ["evaluator", "category", "pass_rate"]
    
    # Low-score queries for manual review
    worst = df.nsmallest(20, "score")[["evaluator", "query", "score", "category"]]
    
    return overall, by_cat, worst

overall, by_cat, worst = summarize(results)
print(overall.to_string(index=False))
print("\nBy category:")
print(by_cat.to_string(index=False))
print("\nWorst 20:")
print(worst.to_string(index=False))

What to watch for:

Faithfulness pass rate below 90%: your generator is hallucinating. Check context window truncation or prompt instructions.
Hit rate below 70%: retriever is missing relevant docs. Check embedding model, chunk size, or top-k.
MRR significantly lower than hit rate: relevant docs are ranked low. Re-rank or adjust similarity threshold.
Category-specific drops: a doc set went stale, or a new query type emerged.

Gate deployments in CI

Add a step that fails the build if core metrics regress. Use a threshold file checked into the repo so thresholds are versioned with the pipeline.

# .github/workflows/eval.yml
name: Nightly Evaluation
on:
  schedule:
    - cron: '0 2 * * *'
  workflow_dispatch:

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements-eval.txt
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python -m eval.run_batch
      - name: Check thresholds
        run: python -m eval.check_thresholds

# eval/check_thresholds.py
import json
import sys
from pathlib import Path

THRESHOLDS = {
    "faithfulness": 0.90,
    "hit_rate": 0.75,
    "mrr": 0.60,
}

def main():
    results_path = Path("eval/results/latest.json")
    with open(results_path) as f:
        data = json.load(f)
    
    failed = []
    for evaluator, threshold in THRESHOLDS.items():
        pass_rate = data[evaluator]["pass_rate"]
        if pass_rate < threshold:
            failed.append(f"{evaluator}: {pass_rate:.2%} < {threshold:.0%}")
    
    if failed:
        print("THRESHOLD FAILURES:")
        for f in failed:
            print(f"  - {f}")
        sys.exit(1)
    
    print("All thresholds passed")

if __name__ == "__main__":
    main()

Pitfall: gating on LLM-judge metrics in CI without a fixed judge model. If the judge model updates (e.g., gpt-4o-mini gets a silent refresh), your pass rates shift. Pin the judge model version or run a calibration set alongside every evaluation to detect judge drift.

Handle async pipelines and streaming responses

If your RAG pipeline uses async retrieval or streams tokens, wrap it in a synchronous callable for the runner. The batch runner expects a query_engine with a .query() method, but you can adapt any async function:

from llama_index.core.query_engine import BaseQueryEngine
from llama_index.core.response import Response
from typing import Any

class AsyncQueryEngineWrapper(BaseQueryEngine):
    def __init__(self, async_query_fn):
        self._async_query_fn = async_query_fn
    
    def _query(self, query_bundle: QueryBundle) -> Response:
        import asyncio
        return asyncio.run(self._async_query_fn(query_bundle.query_str))
    
    async def _aquery(self, query_bundle: QueryBundle) -> Response:
        return await self._async_query_fn(query_bundle.query_str)
    
    @property
    def retriever(self):
        # Required by some evaluators; return your actual retriever
        return self._retriever
    
    @retriever.setter
    def retriever(self, value):
        self._retriever = value

# Usage
async def my_rag_pipeline(query: str) -> Response:
    # Your async retrieval + generation logic
    nodes = await retriever.aretrieve(query)
    response = await generator.agenerate(query, nodes)
    return Response(response=response, source_nodes=nodes)

wrapped_engine = AsyncQueryEngineWrapper(my_rag_pipeline)
wrapped_engine.retriever = retriever  # for hit_rate/MRR evaluators

results = await runner.aevaluate_dataset(dataset, query_engine=wrapped_engine)

Debug failures without re-running the full batch

When a threshold fails, you need to inspect the failing cases without re-evaluating everything. Persist full results with metadata:

import json
from datetime import datetime
from pathlib import Path

def persist_results(results: dict[str, list[EvaluationResult]], run_id: str):
    out_dir = Path(f"eval/results/{run_id}")
    out_dir.mkdir(parents=True, exist_ok=True)
    
    for eval_name, eval_results in results.items():
        serializable = []
        for r in eval_results:
            serializable.append({
                "query": r.query,
                "response": r.response,
                "passing": r.passing,
                "score": r.score,
                "feedback": r.feedback,
                "metadata": r.metadata,
            })
        with open(out_dir / f"{eval_name}.json", "w") as f:
            json.dump(serializable, f, indent=2)
    
    # Write a manifest
    with open(out_dir / "manifest.json", "w") as f:
        json.dump({
            "run_id": run_id,
            "timestamp": datetime.utcnow().isoformat(),
            "evaluators": list(results.keys()),
            "total_queries": len(next(iter(results.values()))),
        }, f, indent=2)

Then query failures locally:

# Show all faithfulness failures with feedback
jq '.[] | select(.passing == false) | {query, score, feedback}' eval/results/2024-01-15/faithfulness.json

Common pitfalls and how to avoid them

1. Evaluating the wrong thing. The batch runner evaluates the query engine, not the retriever in isolation. If you swap retrievers but keep the same generator, faithfulness may stay flat while hit rate drops. Run retrieval-only metrics separately when debugging retriever changes.

2. Ignoring latency variance. The runner reports pass/fail, not p99 latency. A pipeline that passes evaluation but takes 30s/query is broken in production. Add a latency budget check:

import time
from functools import wraps

def timed_query(engine):
    @wraps(engine.query)
    def wrapper(query_bundle):
        start = time.perf_counter()
        response = engine.query(query_bundle)
        response.metadata = response.metadata or {}
        response.metadata["latency_ms"] = (time.perf_counter() - start) * 1000
        return response
    engine.query = wrapper
    return engine

3. Using the same LLM for generation and judging. If your generator is gpt-4o-mini and your judge is gpt-4o-mini, they share failure modes. Use a stronger model (gpt-4o) for judging, or at minimum a different temperature (judge at 0.0, generator at your production setting).

4. Dataset drift. A dataset from six months ago doesn’t reflect current traffic. Automate dataset refresh: sample 10% of daily queries, deduplicate against existing set, add to the eval dataset weekly. Keep a rolling 90-day window.

Scaling beyond the single-machine runner

The built-in runner is single-process. For 10k+ queries, you’ll hit memory and time limits. Options:

Ray: Distribute aevaluate_dataset shards across a cluster. Each worker loads the query engine independently.
Batch API: If your provider supports batch inference (OpenAI, Anthropic), submit all prompts at once and poll for results. Cuts cost by 50% but adds hours of latency.
Streaming evaluation: Evaluate incrementally as queries arrive in production. Log query + response + retrieved docs, then run evaluators asynchronously. This is how you get continuous eval without a separate batch job.

# Conceptual: production logging for continuous eval
async def log_for_eval(query: str, response: Response, latency_ms: float):
    await eval_queue.put({
        "query": query,
        "response": response.response,
        "source_nodes": [n.node_id for n in response.source_nodes],
        "latency_ms": latency_ms,
        "timestamp": datetime.utcnow().isoformat(),
    })

# Separate consumer reads from eval_queue and runs evaluators

This approach lets you catch regressions within minutes of deployment, not hours or days later.

Final checklist before you ship

Dataset reflects real traffic (200+ queries, categorized, versioned)
Metrics match your failure modes (faithfulness + retrieval metrics minimum)
Runner concurrency respects provider rate limits
Thresholds checked into repo, gated in CI
Full results persisted for debugging
Latency budget enforced alongside quality thresholds
Judge model pinned or calibrated
Dataset refresh automated

The batch runner is a solid foundation. The engineering work is wiring around it — dataset hygiene, threshold discipline, observability — is what makes evaluation a reliable gate instead of a checkbox.