Building a RAG evaluation dashboard is the only way to catch retrieval regressions before they poison your generated answers. This tutorial builds a minimal, runnable RAG evaluation dashboard that records retrieved chunks, compares them against ground-truth relevance labels, and serves precision/recall/MRR trends over a simple web UI. The patterns here map directly to warehouse-backed observability if you outgrow SQLite.
Prerequisites
- Python 3.10+ in a clean virtual environment
pip install flask sqlalchemy pandas numpy openai- A retrieval function. We stub
query_corpusto simulate a vector search so the demo runs without a vector DB. - A small labeled query set: each query maps to a list of known relevant document IDs. Without labels, no retrieval metric is meaningful.
- For the generation step we’ll call an OpenAI-compatible endpoint. n4n.ai exposes one endpoint for 240+ models with automatic fallback when a provider is degraded, which keeps the demo resilient if a single vendor rate-limits you.
Step 1: Persist retrieval events
A useful RAG evaluation dashboard starts with an immutable log of what was retrieved versus what was relevant. SQLite is enough for local dev; the schema translates cleanly to Postgres.
from sqlalchemy import create_engine, Column, String, Integer, Float, JSON
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class RetrievalEvent(Base):
__tablename__ = "retrieval_events"
id = Column(Integer, primary_key=True)
query = Column(String, nullable=False)
retrieved_ids = Column(JSON, nullable=False) # ordered list of doc ids
relevant_ids = Column(JSON, nullable=False) # ground truth
latency_ms = Column(Float, nullable=False)
ts = Column(Float, nullable=False)
engine = create_engine("sqlite:///rag_eval.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)Storing retrieved_ids as an ordered JSON array preserves rank, which matters for MRR. Keep relevant_ids as the source of truth from your annotation process.
Step 2: Instrument your retrieval path
Wrap your existing retriever so every call writes to the log. In production you’d emit to a queue; for this tutorial, a synchronous commit is fine.
import time
from datetime import datetime
def instrumented_retrieve(query, relevant_ids, k=5):
start = time.monotonic()
retrieved = query_corpus(query, k) # your real vector search goes here
latency = (time.monotonic() - start) * 1000
with Session() as s:
s.add(RetrievalEvent(
query=query,
retrieved_ids=retrieved,
relevant_ids=relevant_ids,
latency_ms=latency,
ts=datetime.utcnow().timestamp()
))
s.commit()
return retrieved
def query_corpus(query, k):
# deterministic mock: returns first k doc ids
return [f"doc_{i}" for i in range(k)]The wrapper adds zero changes to your downstream generation code. You call instrumented_retrieve exactly where you previously called query_corpus.
Step 3: Compute retrieval metrics
Precision@k and recall@k are non-negotiable. MRR captures whether the first relevant hit appears early.
import numpy as np
def precision_at_k(retrieved, relevant, k):
ret = retrieved[:k]
hits = len(set(ret) & set(relevant))
return hits / k
def recall_at_k(retrieved, relevant, k):
if not relevant:
return 0.0
ret = retrieved[:k]
hits = len(set(ret) & set(relevant))
return hits / len(relevant)
def mrr(retrieved, relevant):
for i, doc in enumerate(retrieved, 1):
if doc in relevant:
return 1 / i
return 0.0
def compute_metrics(events):
rows = []
for e in events:
rows.append({
"query": e.query,
"p@5": precision_at_k(e.retrieved_ids, e.relevant_ids, 5),
"r@5": recall_at_k(e.retrieved_ids, e.relevant_ids, 5),
"mrr": mrr(e.retrieved_ids, e.relevant_ids),
"latency_ms": e.latency_ms,
"ts": e.ts
})
return rowsCheckpoint: seed and inspect
labels = {
"what is rag": ["doc_0", "doc_1"],
"eval metrics": ["doc_2", "doc_3"],
}
for q, rel in labels.items():
instrumented_retrieve(q, rel, k=5)
with Session() as s:
metrics = compute_metrics(s.query(RetrievalEvent).all())
print(metrics[0])Expected output:
{"query": "what is rag", "p@5": 0.4, "r@5": 1.0, "mrr": 1.0, "latency_ms": 0.11, "ts": 1718220000.0}Your RAG evaluation dashboard will plot these rows over time.
Step 4: Serve aggregates from Flask
The API computes rolling averages on read. For scale, pre-aggregate in a cron job; for a demo, on-read is simpler.
from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/api/metrics")
def api_metrics():
with Session() as s:
events = s.query(RetrievalEvent).all()
rows = compute_metrics(events)
if not rows:
return jsonify({"avg_p": 0, "avg_r": 0, "avg_mrr": 0, "events": []})
avg_p = float(np.mean([r["p@5"] for r in rows]))
avg_r = float(np.mean([r["r@5"] for r in rows]))
avg_mrr = float(np.mean([r["mrr"] for r in rows]))
return jsonify({
"avg_p": avg_p, "avg_r": avg_r, "avg_mrr": avg_mrr,
"events": rows[-20:]
})
if __name__ == "__main__":
app.run(port=5000)Run python app.py and curl the endpoint:
curl localhost:5000/api/metricsYou should see the JSON averages and the last 20 events.
Step 5: Render the RAG evaluation dashboard
A single HTML file polled every 5 seconds is enough. Drop it in templates/dash.html.
<!doctype html>
<html><head><style>body{font:14px monospace}table{border-collapse:collapse}td,th{border:1px solid #ccc;padding:4px}</style></head>
<body>
<h2>RAG evaluation dashboard</h2>
<div id="kpis"></div>
<table id="tbl"><tr><th>Query</th><th>P@5</th><th>R@5</th><th>MRR</th><th>ms</th></tr></table>
<script>
async function load(){
const r = await fetch('/api/metrics');
const d = await r.json();
document.getElementById('kpis').innerHTML =
`avg P@5 ${d.avg_p.toFixed(2)} | avg R@5 ${d.avg_r.toFixed(2)} | avg MRR ${d.avg_mrr.toFixed(2)}`;
const tbl = document.getElementById('tbl');
tbl.innerHTML = '<tr><th>Query</th><th>P@5</th><th>R@5</th><th>MRR</th><th>ms</th></tr>';
d.events.forEach(e=>{
tbl.innerHTML += `<tr><td>${e.query}</td><td>${e.p@5.toFixed(2)}</td><td>${e.r@5.toFixed(2)}</td><td>${e.mrr.toFixed(2)}</td><td>${e.latency_ms.toFixed(1)}</td></tr>`;
});
}
setInterval(load, 5000); load();
</script></body></html>Add a route to serve it:
from flask import render_template
@app.route("/")
def dash():
return render_template("dash.html")Open localhost:5000 and watch the table populate as you issue more instrumented queries.
Step 6: Close the loop with generation
Retrieval metrics alone don’t prove the answer is good. Log the generated text and a cheap faithfulness heuristic. Call your LLM through any OpenAI-compatible client:
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")
def generate(question, ctx):
resp = client.chat.completions.create(
model="anthropic/claude-3-haiku",
messages=[{"role":"user","content": f"Context: {ctx}\nQ: {question}"}]
)
return resp.choices[0].message.contentn4n.ai forwards provider cache-control hints and meters per-token usage, so you can attribute cost per retrieved query inside the same dashboard.
Ground truth labeling strategy
A RAG evaluation dashboard is only as trustworthy as its labels. Generate candidate relevant docs with a strong retriever, then have a human confirm. Store labels separately and join at metric time so you can revise judgments without rewriting history.
What not to measure (yet)
Skip end-to-end answer correctness scoring until retrieval is solid. If P@5 is below 0.6, no prompt engineering will save you. Latency per query belongs on the same view—a retriever that gets smarter but doubles tail latency is a regression.
Extending beyond the demo
Swap SQLite for DuckDB or Postgres, add a ?window=24h param to /api/metrics for time-series, and track context_utilization (fraction of retrieved tokens cited by the model). The core of a useful RAG evaluation dashboard is the labeled event log; everything else is plumbing.