Skip to main content

RAG Retrieval-Quality Eval Harness — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build an offline, reproducible harness that scores each KB retrieval engine in isolation on a synthetic query set and reports Hit@k + MRR side-by-side.

Architecture: Python package under rag/evals/retrieval/. A fixed seed corpus is ingested into an ephemeral pgvector container (Engine A) and held in memory (Engine B). An LLM generates one paraphrased question per chunk (cached). Each engine runs the same queries in isolation; pure metric functions score the rankings; a reporter emits a comparison table.

Tech Stack: Python 3.11, Haystack 2.28 (real production components), pgvector via testcontainers, OpenRouter (embeddings/rerank/generation via httpx), pytest.

Design spec: docs/superpowers/specs/2026-06-23-rag-retrieval-eval-harness-design.md

Conventions​

  • All commands run from the rag/ directory unless stated otherwise.
  • Install dev + eval deps once: pip install -e ".[dev,eval]" from rag/.
  • New code lives under rag/evals/retrieval/. Tests under rag/evals/retrieval/tests/.
  • Commit after every green step. Commit messages use feat(rag-eval): / test(rag-eval): / chore(rag-eval):.

File structure (locked)​

rag/evals/retrieval/
  __init__.py
  types.py                  # Doc, LabeledQuery, EngineResult dataclasses
  metrics.py                # pure hit_at_k / mrr / aggregate
  corpus/
    __init__.py
    provider.py             # CorpusProvider ABC + SeedCorpusProvider
    seed_docs/              # committed *.json corpus docs
  engines/
    __init__.py
    base.py                 # RetrievalEngine ABC
    engine_b_keyword.py     # Python port of KbRetrievalService scoring
    engine_a_haystack.py    # configurable hybrid+rerank adapter
  ingest.py                 # EphemeralPgvector: stand up + load + teardown
  generate_questions.py     # QuestionGenerator + JSON cache
  report.py                 # results.md + per_query.jsonl
  run.py                    # CLI orchestrator
  questions.cache.json      # committed labeled set (generated once)
  tests/
    __init__.py
    test_metrics.py
    test_engine_b_keyword.py
    test_corpus_provider.py
    test_engine_a_haystack.py   # integration; skipped if no docker
    fixtures/
      tiny_corpus/          # 3-4 docs for wiring self-test

Task 1: Package scaffold + dependency extra​

Files:

  • Create: rag/evals/retrieval/__init__.py (empty)

  • Create: rag/evals/retrieval/tests/__init__.py (empty)

  • Create: rag/evals/retrieval/corpus/__init__.py (empty)

  • Create: rag/evals/retrieval/engines/__init__.py (empty)

  • Modify: rag/pyproject.toml

  • Step 1: Create the empty package files

Create each __init__.py listed above with no content.

  • Step 2: Add a retrieval-eval optional-dependency extra

In rag/pyproject.toml, under [project.optional-dependencies], add a new extra after the existing eval block:

retrieval-eval = [
  "testcontainers[postgres]==4.*",   # also in dev; pinned here so the harness is installable standalone
]

(LLM generation, embeddings, and rerank reuse the already-present httpx runtime dep and the production components.openrouter_ranker / Haystack stack — no new deps needed for those.)

  • Step 3: Install and verify import

Run (from rag/): pip install -e ".[dev,eval,retrieval-eval]" && python -c "import evals.retrieval" Expected: no output, exit 0.

  • Step 4: Commit
git add rag/evals/retrieval rag/pyproject.toml
git commit -m "chore(rag-eval): scaffold retrieval eval package"

Task 2: Core types​

Files:

  • Create: rag/evals/retrieval/types.py

  • Step 1: Write the dataclasses

"""Shared dataclasses for the retrieval eval harness."""

from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class Doc:
    """A corpus document. Engine A chunks+embeds `content`; Engine B scores
    only `title`/`description`/`filename` (it never reads `content`)."""

    doc_id: str
    title: str
    description: str
    filename: str
    content: str
    org_id: str = "eval-org"
    dataset: str = "default-kb"
    specialist_id: str | None = None


@dataclass(frozen=True)
class LabeledQuery:
    """A synthetic question with its single known-relevant (gold) document."""

    question: str
    gold_doc_id: str
    source_chunk_id: str


@dataclass
class EngineResult:
    """One engine's ranked output for one query, normalized to doc granularity."""

    engine: str
    question: str
    gold_doc_id: str
    ranked_doc_ids: list[str] = field(default_factory=list)
    errored: bool = False
    error_message: str | None = None
  • Step 2: Verify import

Run (from rag/): python -c "from evals.retrieval.types import Doc, LabeledQuery, EngineResult" Expected: exit 0.

  • Step 3: Commit
git add rag/evals/retrieval/types.py
git commit -m "feat(rag-eval): core dataclasses (Doc, LabeledQuery, EngineResult)"

Task 3: Metric functions (TDD — the must-be-correct core)​

Files:

  • Create: rag/evals/retrieval/metrics.py

  • Test: rag/evals/retrieval/tests/test_metrics.py

  • Step 1: Write the failing tests

from evals.retrieval.metrics import aggregate, hit_at_k, reciprocal_rank
from evals.retrieval.types import EngineResult

K_VALUES = (1, 3, 5, 10, 20)


def _result(ranked, gold="g", errored=False):
    return EngineResult(
        engine="x", question="q", gold_doc_id=gold,
        ranked_doc_ids=ranked, errored=errored,
    )


def test_hit_at_k_gold_at_rank_1():
    r = _result(["g", "a", "b"])
    assert hit_at_k(r, 1) == 1.0
    assert hit_at_k(r, 5) == 1.0


def test_hit_at_k_gold_at_rank_3_misses_k1():
    r = _result(["a", "b", "g"])
    assert hit_at_k(r, 1) == 0.0
    assert hit_at_k(r, 3) == 1.0


def test_hit_at_k_gold_absent():
    r = _result(["a", "b", "c"])
    assert hit_at_k(r, 20) == 0.0


def test_hit_at_k_errored_is_miss():
    r = _result([], errored=True)
    assert hit_at_k(r, 20) == 0.0


def test_reciprocal_rank_positions():
    assert reciprocal_rank(_result(["g", "a"])) == 1.0
    assert reciprocal_rank(_result(["a", "b", "g"])) == 1.0 / 3.0
    assert reciprocal_rank(_result(["a", "b"])) == 0.0
    assert reciprocal_rank(_result([], errored=True)) == 0.0


def test_aggregate_means_over_queries():
    results = [_result(["g"]), _result(["a", "g"]), _result(["a"])]
    agg = aggregate("engine-x", results, K_VALUES)
    assert agg["engine"] == "engine-x"
    assert agg["n_queries"] == 3
    assert agg["n_errored"] == 0
    # hit@1: 1,0,0 -> 1/3 ; hit@3: 1,1,0 -> 2/3
    assert abs(agg["hit_at_k"][1] - 1 / 3) < 1e-9
    assert abs(agg["hit_at_k"][3] - 2 / 3) < 1e-9
    # mrr: 1 + 1/2 + 0 = 1.5 / 3 = 0.5
    assert abs(agg["mrr"] - 0.5) < 1e-9


def test_aggregate_counts_errors():
    results = [_result(["g"]), _result([], errored=True)]
    agg = aggregate("e", results, K_VALUES)
    assert agg["n_errored"] == 1
    assert abs(agg["mrr"] - 0.5) < 1e-9  # errored counts as 0
  • Step 2: Run tests to verify they fail

Run (from rag/): pytest evals/retrieval/tests/test_metrics.py -v Expected: FAIL with ModuleNotFoundError: No module named 'evals.retrieval.metrics'.

  • Step 3: Implement metrics.py
"""Pure retrieval metric functions.

With exactly one gold document per query, Recall@k is identical to Hit@k
(relevant-found / total-relevant = found / 1), so only Hit@k and MRR are
reported. An errored engine result counts as a miss (0.0) everywhere.
"""

from __future__ import annotations

from collections.abc import Iterable, Sequence

from evals.retrieval.types import EngineResult


def _gold_rank(result: EngineResult) -> int | None:
    """1-based rank of the gold doc, or None if absent/errored."""
    if result.errored:
        return None
    for index, doc_id in enumerate(result.ranked_doc_ids):
        if doc_id == result.gold_doc_id:
            return index + 1
    return None


def hit_at_k(result: EngineResult, k: int) -> float:
    rank = _gold_rank(result)
    return 1.0 if rank is not None and rank <= k else 0.0


def reciprocal_rank(result: EngineResult) -> float:
    rank = _gold_rank(result)
    return 1.0 / rank if rank is not None else 0.0


def aggregate(
    engine: str,
    results: Sequence[EngineResult],
    k_values: Iterable[int],
) -> dict:
    n = len(results)
    k_list = list(k_values)
    if n == 0:
        return {
            "engine": engine, "n_queries": 0, "n_errored": 0,
            "hit_at_k": {k: 0.0 for k in k_list}, "mrr": 0.0,
        }
    return {
        "engine": engine,
        "n_queries": n,
        "n_errored": sum(1 for r in results if r.errored),
        "hit_at_k": {
            k: sum(hit_at_k(r, k) for r in results) / n for k in k_list
        },
        "mrr": sum(reciprocal_rank(r) for r in results) / n,
    }
  • Step 4: Run tests to verify they pass

Run (from rag/): pytest evals/retrieval/tests/test_metrics.py -v Expected: 7 passed.

  • Step 5: Commit
git add rag/evals/retrieval/metrics.py rag/evals/retrieval/tests/test_metrics.py
git commit -m "feat(rag-eval): pure Hit@k + MRR metric functions (TDD)"

Task 4: Engine B — Python port of keyword scoring (TDD)​

Files:

  • Create: rag/evals/retrieval/engines/base.py
  • Create: rag/evals/retrieval/engines/engine_b_keyword.py
  • Test: rag/evals/retrieval/tests/test_engine_b_keyword.py

Source of truth being ported: api/src/kb/kb-retrieval.service.ts functions normalizeText, tokenize, countOccurrences, scoreField, scoreDocument, compareScoredDocuments (as of commit on dev 2026-06-23). Keep this comment pin in the file.

  • Step 1: Write the engine interface

engines/base.py:

"""Common retrieval-engine interface for the eval harness."""

from __future__ import annotations

from abc import ABC, abstractmethod


class RetrievalEngine(ABC):
    """Each engine retrieves for a query and returns ranked DOCUMENT ids.

    Engines that operate on chunks must normalize to document granularity
    (dedupe to first occurrence of each parent doc_id, order preserved)."""

    name: str

    @abstractmethod
    def retrieve(self, query: str, k: int) -> list[str]:
        """Return up to `k` ranked doc_ids, best first."""
        raise NotImplementedError
  • Step 2: Write the failing tests
from evals.retrieval.engines.engine_b_keyword import (
    KeywordEngine, normalize_text, score_document, tokenize,
)
from evals.retrieval.types import Doc


def _doc(doc_id, title="", description="", filename="", content=""):
    return Doc(doc_id=doc_id, title=title, description=description,
               filename=filename, content=content)


def test_normalize_text_lowercases_and_collapses():
    assert normalize_text("Foo_Bar-Baz   Qux") == "foo bar baz qux"


def test_tokenize_dedupes_and_drops_short_tokens():
    assert tokenize("refund a Refund policy") == ["refund", "policy"]


def test_score_document_weights_title_highest():
    # full-query substring in title: +8*4 = 32, plus per-term title hits
    # "refund policy" -> query substring present in title
    doc = _doc("d1", title="Refund policy", content="ignored body text")
    score = score_document(doc, "refund policy", ["refund", "policy"])
    assert score > 0


def test_score_document_ignores_content():
    # gold term ONLY in body content -> Engine B must score 0 (it never reads content)
    doc = _doc("d1", title="Shipping", description="", filename="ship.md",
               content="our refund policy is 30 days")
    assert score_document(doc, "refund policy", ["refund", "policy"]) == 0.0


def test_retrieve_ranks_by_score_then_returns_doc_ids():
    docs = [
        _doc("d1", title="Refund policy details"),
        _doc("d2", title="Shipping info", description="refund mentioned once"),
        _doc("d3", title="Totally unrelated"),
    ]
    engine = KeywordEngine(docs)
    ranked = engine.retrieve("refund policy", k=10)
    assert ranked[0] == "d1"          # strongest title match first
    assert "d3" not in ranked         # zero score filtered out
    assert ranked == ["d1", "d2"]


def test_retrieve_respects_k():
    docs = [_doc(f"d{i}", title="refund") for i in range(5)]
    assert len(KeywordEngine(docs).retrieve("refund", k=3)) == 3
  • Step 3: Run tests to verify they fail

Run (from rag/): pytest evals/retrieval/tests/test_engine_b_keyword.py -v Expected: FAIL with import error.

  • Step 4: Implement engine_b_keyword.py
"""Engine B: faithful Python port of NestJS KbRetrievalService keyword scoring.

PORTED FROM: api/src/kb/kb-retrieval.service.ts (dev @ 2026-06-23).
Mirrors normalizeText / tokenize / countOccurrences / scoreField /
scoreDocument / compareScoredDocuments. ACCEPTED RISK: this is a COPY and can
drift if the NestJS logic changes (see design spec "Open questions / risks").
Engine B scores ONLY title/description/filename metadata — never chunk body.
"""

from __future__ import annotations

import re

from evals.retrieval.engines.base import RetrievalEngine
from evals.retrieval.types import Doc

_NON_ALNUM = re.compile(r"[^a-z0-9]+")
_SEP = re.compile(r"[_-]+")
_WS = re.compile(r"\s+")


def normalize_text(value: str) -> str:
    value = value.lower()
    value = _SEP.sub(" ", value)
    value = _WS.sub(" ", value)
    return value.strip()


def tokenize(value: str) -> list[str]:
    seen: dict[str, None] = {}
    for term in _NON_ALNUM.split(normalize_text(value)):
        if len(term) >= 2 and term not in seen:
            seen[term] = None
    return list(seen)


def _count_occurrences(haystack: str, needle: str) -> int:
    if not needle:
        return 0
    count = 0
    index = haystack.find(needle)
    while index != -1:
        count += 1
        index = haystack.find(needle, index + len(needle))
    return count


def _score_field(field: str, terms: list[str], query: str, weight: float) -> float:
    normalized = normalize_text(field)
    if not normalized:
        return 0.0
    score = 0.0
    if query and query in normalized:
        score += 8 * weight
    for term in terms:
        if term in normalized:
            score += weight
            score += _count_occurrences(normalized, term) * 0.25 * weight
    return score


def score_document(doc: Doc, query: str, terms: list[str]) -> float:
    return (
        _score_field(doc.title or "", terms, query, 4)
        + _score_field(doc.description or "", terms, query, 2)
        + _score_field(doc.filename or "", terms, query, 3)
    )


class KeywordEngine(RetrievalEngine):
    name = "engine_b_keyword"

    def __init__(self, docs: list[Doc]) -> None:
        self._docs = docs

    def retrieve(self, query: str, k: int) -> list[str]:
        normalized_query = normalize_text(query)
        terms = tokenize(query)
        scored = [
            (doc, score_document(doc, normalized_query, terms))
            for doc in self._docs
        ]
        scored = [item for item in scored if not normalized_query or item[1] > 0]
        # Tie-break mirrors compareScoredDocuments: score desc, then doc_id asc
        # (createdAt is not modeled in the eval corpus, so it is omitted).
        scored.sort(key=lambda item: (-item[1], item[0].doc_id))
        return [doc.doc_id for doc, _ in scored[:k]]
  • Step 5: Run tests to verify they pass

Run (from rag/): pytest evals/retrieval/tests/test_engine_b_keyword.py -v Expected: 6 passed.

  • Step 6: Commit
git add rag/evals/retrieval/engines/base.py rag/evals/retrieval/engines/engine_b_keyword.py rag/evals/retrieval/tests/test_engine_b_keyword.py
git commit -m "feat(rag-eval): Engine B keyword port + interface (TDD)"

Task 5: Corpus provider + seed docs (TDD)​

Files:

  • Create: rag/evals/retrieval/corpus/provider.py

  • Create: rag/evals/retrieval/corpus/seed_docs/*.json (the committed corpus)

  • Test: rag/evals/retrieval/tests/test_corpus_provider.py

  • Step 1: Write the failing test

from evals.retrieval.corpus.provider import SeedCorpusProvider
from evals.retrieval.types import Doc


def test_seed_provider_loads_committed_docs():
    docs = SeedCorpusProvider().get_documents()
    assert len(docs) >= 15
    assert all(isinstance(d, Doc) for d in docs)
    ids = [d.doc_id for d in docs]
    assert len(ids) == len(set(ids)), "doc_ids must be unique"
    assert all(d.content.strip() for d in docs), "every doc needs body content"
    assert all(d.title.strip() for d in docs), "every doc needs a title"
  • Step 2: Run test to verify it fails

Run (from rag/): pytest evals/retrieval/tests/test_corpus_provider.py -v Expected: FAIL with import error.

  • Step 3: Implement provider.py
"""Corpus providers. v1 ships a committed seed corpus; a future
DevKbCorpusProvider (real dev KB) implements the same interface (design C-seam).
"""

from __future__ import annotations

import json
from abc import ABC, abstractmethod
from pathlib import Path

from evals.retrieval.types import Doc

_SEED_DIR = Path(__file__).with_name("seed_docs")


class CorpusProvider(ABC):
    @abstractmethod
    def get_documents(self) -> list[Doc]:
        raise NotImplementedError


class SeedCorpusProvider(CorpusProvider):
    def __init__(self, seed_dir: Path = _SEED_DIR) -> None:
        self._seed_dir = seed_dir

    def get_documents(self) -> list[Doc]:
        docs: list[Doc] = []
        for path in sorted(self._seed_dir.glob("*.json")):
            raw = json.loads(path.read_text(encoding="utf-8"))
            docs.append(
                Doc(
                    doc_id=raw["doc_id"],
                    title=raw["title"],
                    description=raw.get("description", ""),
                    filename=raw.get("filename", f"{raw['doc_id']}.md"),
                    content=raw["content"],
                    specialist_id=raw.get("specialist_id"),
                )
            )
        return docs
  • Step 4: Author the seed corpus (15-30 docs)

Create one JSON file per document under seed_docs/, each shaped:

{
  "doc_id": "refund-policy",
  "title": "Refund and Returns Policy",
  "description": "How refunds and returns work for orders.",
  "filename": "refund-policy.md",
  "content": "Full multi-paragraph body text long enough to chunk (300+ words). Cover a clear topic so a question can be generated from each chunk..."
}

Requirements for the corpus (so the eval is meaningful):

  • 15-30 docs spanning varied shapes: short FAQ-style, long-form policy (multi-chunk), and at least 3 docs whose distinctive content lives only in the body, not the title (these expose Engine B's metadata-only blind spot).

  • Topics should be distinct enough that a question from one doc is not ambiguously answerable by another (keeps the single-gold assumption honest).

  • Write original prose — do not paste real client data (per project secret-handling rules).

  • Step 5: Run test to verify it passes

Run (from rag/): pytest evals/retrieval/tests/test_corpus_provider.py -v Expected: 1 passed.

  • Step 6: Commit
git add rag/evals/retrieval/corpus rag/evals/retrieval/tests/test_corpus_provider.py
git commit -m "feat(rag-eval): seed corpus + CorpusProvider (TDD)"

Task 6: Ephemeral pgvector ingestor​

Files:

  • Create: rag/evals/retrieval/ingest.py

This stands up a throwaway pgvector container, runs the real production ingestion components (cleaner → splitter → embedder → writer) over the corpus, and exposes the connection string for Engine A. It also returns the chunk→doc map so question generation knows each chunk's parent.

  • Step 1: Implement ingest.py
"""Ephemeral pgvector store + corpus ingestion for Engine A.

Reuses the production Haystack ingestion components (same splitter/embedder/
store config as rag/pipelines/ingestion/pipeline_wrapper.py) so the eval
measures the real chunking + embedding path, not a reimplementation.
"""

from __future__ import annotations

import os
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any, Iterator

from evals.retrieval.types import Doc

PG_IMAGE = "pgvector/pgvector:pg16"


@dataclass
class IngestedChunk:
    chunk_id: str
    doc_id: str
    content: str


class EphemeralPgvector:
    """Owns the throwaway Postgres container and the loaded Haystack store."""

    def __init__(self, conn_str: str) -> None:
        self.conn_str = conn_str
        self._store: Any = None
        self.chunks: list[IngestedChunk] = []

    def _build_store(self) -> Any:
        from haystack.utils import Secret
        from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore

        os.environ["PG_CONN_STR"] = self.conn_str
        return PgvectorDocumentStore(
            connection_string=Secret.from_env_var("PG_CONN_STR"),
            schema_name="haystack",
            table_name="documents",
            embedding_dimension=1536,
            vector_function="cosine_similarity",
            search_strategy="hnsw",
            hnsw_index_creation_kwargs={"m": 16, "ef_construction": 64},
            hnsw_ef_search=40,
            keyword_index_name="documents_bm25_idx",
            recreate_table=True,
        )

    def load(self, docs: list[Doc]) -> None:
        from haystack.components.embedders import OpenAIDocumentEmbedder
        from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
        from haystack.components.writers import DocumentWriter
        from haystack.dataclasses import Document
        from haystack.document_stores.types import DuplicatePolicy
        from haystack.utils import Secret

        self._store = self._build_store()
        cleaner = DocumentCleaner(
            remove_empty_lines=True, remove_extra_whitespaces=True,
            remove_repeated_substrings=False,
        )
        splitter = DocumentSplitter(
            split_by="word", split_length=400, split_overlap=80,
            respect_sentence_boundary=True,
        )
        embedder = OpenAIDocumentEmbedder(
            api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
            api_base_url="https://openrouter.ai/api/v1",
            model="openai/text-embedding-3-small",
            dimensions=1536, batch_size=64,
        )
        writer = DocumentWriter(document_store=self._store, policy=DuplicatePolicy.OVERWRITE)

        haystack_docs: list[Document] = []
        for doc in docs:
            base_meta = {
                "org_id": doc.org_id, "dataset": doc.dataset,
                "specialist_id": doc.specialist_id, "doc_id": doc.doc_id,
                "filename": doc.filename,
            }
            haystack_docs.append(Document(content=doc.content, meta=dict(base_meta)))

        cleaned = cleaner.run(documents=haystack_docs)["documents"]
        chunks = splitter.run(documents=cleaned)["documents"]
        for index, chunk in enumerate(chunks):
            parent = chunk.meta["doc_id"]
            chunk.id = f"{parent}:{index}"
            chunk.meta = {**chunk.meta, "chunk_index": index}
            self.chunks.append(
                IngestedChunk(chunk_id=chunk.id, doc_id=parent, content=chunk.content)
            )
        embedded = embedder.run(documents=chunks)["documents"]
        writer.run(documents=embedded)


@contextmanager
def ephemeral_pgvector(docs: list[Doc]) -> Iterator[EphemeralPgvector]:
    """Context manager: start container (or use EVAL_PG_CONN_STR), ingest, teardown.

    If EVAL_PG_CONN_STR is set, use that existing pgvector DB (no container).
    Otherwise spin up a throwaway pgvector container via testcontainers.
    """
    existing = os.environ.get("EVAL_PG_CONN_STR")
    if existing:
        store = EphemeralPgvector(existing)
        store.load(docs)
        try:
            yield store
        finally:
            pass  # caller-owned DB; do not drop
        return

    from testcontainers.postgres import PostgresContainer

    with PostgresContainer(PG_IMAGE, driver=None) as pg:
        conn_str = (
            f"postgresql://{pg.username}:{pg.password}"
            f"@{pg.get_container_host_ip()}:{pg.get_exposed_port(5432)}/{pg.dbname}"
        )
        store = EphemeralPgvector(conn_str)
        store.load(docs)
        yield store  # container auto-stops on context exit (teardown)
  • Step 2: Smoke-check the module imports

Run (from rag/): python -c "from evals.retrieval.ingest import ephemeral_pgvector, EphemeralPgvector" Expected: exit 0. (Live container behavior is exercised in Task 8's integration test.)

  • Step 3: Commit
git add rag/evals/retrieval/ingest.py
git commit -m "feat(rag-eval): ephemeral pgvector ingestor (real ingestion components)"

Task 7: Question generator + cache​

Files:

  • Create: rag/evals/retrieval/generate_questions.py

Generates one paraphrased question per ingested chunk via OpenRouter, caches the labeled set to committed JSON, and replays the cache on subsequent runs.

  • Step 1: Implement generate_questions.py
"""Synthetic question generator (ground-truth approach A).

For each ingested chunk, an OpenRouter LLM writes ONE question the chunk
answers. The chunk's parent doc is the gold label. Output is cached to a
committed JSON file; reruns replay the cache (deterministic, no token spend).
Use --regenerate (run.py flag) to force regeneration.

PARAPHRASE, don't quote: the prompt instructs the model to avoid copying
distinctive phrases, to blunt lexical-overlap bias that would flatter the
keyword engine (design spec "Known caveat").
"""

from __future__ import annotations

import json
import os
from pathlib import Path

import httpx

from evals.retrieval.ingest import IngestedChunk
from evals.retrieval.types import LabeledQuery

CACHE_PATH = Path(__file__).with_name("questions.cache.json")
GEN_MODEL = os.environ.get("EVAL_GEN_MODEL", "openai/gpt-4o-mini")

_PROMPT = (
    "You are generating an evaluation question for a retrieval system.\n"
    "Read the passage below and write ONE natural question that a user would "
    "ask, which this passage answers.\n"
    "Rules:\n"
    "- Paraphrase. Do NOT reuse distinctive words or phrases from the passage.\n"
    "- The question must be answerable ONLY from this passage's information.\n"
    "- Output ONLY the question text, no preamble.\n\n"
    "Passage:\n{content}\n"
)


def load_cache() -> list[LabeledQuery] | None:
    if not CACHE_PATH.exists():
        return None
    rows = json.loads(CACHE_PATH.read_text(encoding="utf-8"))
    return [
        LabeledQuery(
            question=r["question"], gold_doc_id=r["gold_doc_id"],
            source_chunk_id=r["source_chunk_id"],
        )
        for r in rows
    ]


def _write_cache(queries: list[LabeledQuery]) -> None:
    CACHE_PATH.write_text(
        json.dumps(
            [
                {
                    "question": q.question, "gold_doc_id": q.gold_doc_id,
                    "source_chunk_id": q.source_chunk_id,
                }
                for q in queries
            ],
            indent=2, ensure_ascii=False,
        ),
        encoding="utf-8",
    )


def _generate_one(client: httpx.Client, content: str) -> str:
    resp = client.post(
        "https://openrouter.ai/api/v1/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['OPENROUTER_API_KEY']}"},
        json={
            "model": GEN_MODEL,
            "temperature": 0,
            "messages": [{"role": "user", "content": _PROMPT.format(content=content)}],
        },
        timeout=60.0,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"].strip()


def generate_questions(
    chunks: list[IngestedChunk], regenerate: bool = False
) -> tuple[list[LabeledQuery], int]:
    """Return (labeled_queries, n_errored). Fail-loud: a chunk that errors is
    logged, excluded, and counted — never silently dropped."""
    if not regenerate:
        cached = load_cache()
        if cached is not None:
            return cached, 0

    queries: list[LabeledQuery] = []
    errored = 0
    with httpx.Client() as client:
        for chunk in chunks:
            try:
                question = _generate_one(client, chunk.content)
            except Exception as exc:  # noqa: BLE001 - eval is fail-loud, record + continue
                errored += 1
                print(f"[generate] ERROR chunk={chunk.chunk_id}: {exc}")
                continue
            queries.append(
                LabeledQuery(
                    question=question, gold_doc_id=chunk.doc_id,
                    source_chunk_id=chunk.chunk_id,
                )
            )
    _write_cache(queries)
    return queries, errored
  • Step 2: Verify import

Run (from rag/): python -c "from evals.retrieval.generate_questions import generate_questions, load_cache" Expected: exit 0.

  • Step 3: Commit
git add rag/evals/retrieval/generate_questions.py
git commit -m "feat(rag-eval): synthetic question generator with committed cache"

Task 8: Engine A adapter + integration test​

Files:

  • Create: rag/evals/retrieval/engines/engine_a_haystack.py
  • Test: rag/evals/retrieval/tests/test_engine_a_haystack.py

Builds the real hybrid+rerank pipeline (mirroring rag/pipelines/retrieval/pipeline_wrapper.py) against the ephemeral store, with a config so reranker/top_k can be varied later.

  • Step 1: Implement engine_a_haystack.py
"""Engine A: hybrid (BM25 + vector + RRF) + Cohere rerank, built from the real
production Haystack components (mirrors rag/pipelines/retrieval/pipeline_wrapper.py).

Config seam (EngineAConfig) makes reranker model / top_k / rerank on-off / fusion
top_k tunable, so a later reranker sweep is a loop over configs.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any

from evals.retrieval.engines.base import RetrievalEngine

ORG_ID = "eval-org"
DATASET = "default-kb"


@dataclass(frozen=True)
class EngineAConfig:
    rerank: bool = True
    rerank_model: str = "cohere/rerank-v3.5"
    bm25_top_k: int = 30
    vec_top_k: int = 30
    fusion_top_k: int = 20


class HaystackEngine(RetrievalEngine):
    name = "engine_a_haystack"

    def __init__(self, conn_str: str, config: EngineAConfig | None = None) -> None:
        self.config = config or EngineAConfig()
        os.environ["PG_CONN_STR"] = conn_str
        self._build()

    def _build(self) -> None:
        from haystack.components.embedders import OpenAITextEmbedder
        from haystack.components.joiners import DocumentJoiner
        from haystack.utils import Secret
        from haystack_integrations.components.retrievers.pgvector import (
            PgvectorEmbeddingRetriever, PgvectorKeywordRetriever,
        )
        from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore

        from components.openrouter_ranker import OpenRouterRanker

        store = PgvectorDocumentStore(
            connection_string=Secret.from_env_var("PG_CONN_STR"),
            schema_name="haystack", table_name="documents",
            embedding_dimension=1536, vector_function="cosine_similarity",
            search_strategy="hnsw", hnsw_ef_search=40,
        )
        self._embedder = OpenAITextEmbedder(
            api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
            api_base_url="https://openrouter.ai/api/v1",
            model="openai/text-embedding-3-small", dimensions=1536,
        )
        self._bm25 = PgvectorKeywordRetriever(document_store=store, top_k=self.config.bm25_top_k)
        self._vec = PgvectorEmbeddingRetriever(document_store=store, top_k=self.config.vec_top_k)
        self._joiner = DocumentJoiner(
            join_mode="reciprocal_rank_fusion", top_k=self.config.fusion_top_k
        )
        self._ranker = (
            OpenRouterRanker(
                api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
                api_base_url="https://openrouter.ai/api/v1",
                model=self.config.rerank_model, top_k=self.config.fusion_top_k,
            )
            if self.config.rerank
            else None
        )

    def _filters(self) -> dict[str, Any]:
        return {
            "operator": "AND",
            "conditions": [
                {"field": "meta.org_id", "operator": "==", "value": ORG_ID},
                {"field": "meta.dataset", "operator": "==", "value": DATASET},
            ],
        }

    def retrieve(self, query: str, k: int) -> list[str]:
        filters = self._filters()
        q_emb = self._embedder.run(text=query)["embedding"]
        bm25_docs = self._bm25.run(query=query, filters=filters, top_k=self.config.bm25_top_k)["documents"]
        vec_docs = self._vec.run(query_embedding=q_emb, filters=filters, top_k=self.config.vec_top_k)["documents"]
        fused = self._joiner.run(documents=[bm25_docs, vec_docs], top_k=self.config.fusion_top_k)["documents"]
        if self._ranker is not None and fused:
            fused = self._ranker.run(query=query, documents=fused, top_k=k)["documents"]
        # Normalize chunks -> document granularity: first occurrence of each parent doc_id.
        ordered_doc_ids: list[str] = []
        seen: set[str] = set()
        for doc in fused:
            doc_id = (getattr(doc, "meta", {}) or {}).get("doc_id")
            if doc_id and doc_id not in seen:
                seen.add(doc_id)
                ordered_doc_ids.append(doc_id)
        return ordered_doc_ids[:k]
  • Step 2: Write the integration test (skipped without docker + key)
import os
import shutil

import pytest

from evals.retrieval.corpus.provider import SeedCorpusProvider
from evals.retrieval.engines.engine_a_haystack import EngineAConfig, HaystackEngine
from evals.retrieval.ingest import ephemeral_pgvector

pytestmark = pytest.mark.skipif(
    shutil.which("docker") is None or not os.environ.get("OPENROUTER_API_KEY"),
    reason="requires docker + OPENROUTER_API_KEY (live integration)",
)


def test_engine_a_retrieves_ingested_doc():
    docs = SeedCorpusProvider().get_documents()[:5]
    with ephemeral_pgvector(docs) as store:
        engine = HaystackEngine(store.conn_str, EngineAConfig(rerank=False))
        # Query with text drawn from the first doc's body should retrieve it.
        ranked = engine.retrieve(docs[0].content[:120], k=10)
        assert docs[0].doc_id in ranked
  • Step 3: Run the integration test

Run (from rag/, with docker running + key set): pytest evals/retrieval/tests/test_engine_a_haystack.py -v Expected: 1 passed (or skipped if docker/key absent — that is acceptable locally; it runs where infra exists).

  • Step 4: Commit
git add rag/evals/retrieval/engines/engine_a_haystack.py rag/evals/retrieval/tests/test_engine_a_haystack.py
git commit -m "feat(rag-eval): Engine A hybrid+rerank adapter (configurable) + integration test"

Task 9: Reporter​

Files:

  • Create: rag/evals/retrieval/report.py

  • Step 1: Implement report.py

"""Render eval results: a markdown comparison table + per-query JSONL drill-down."""

from __future__ import annotations

import json
from pathlib import Path

from evals.retrieval.types import EngineResult

K_VALUES = (1, 3, 5, 10, 20)


def render_markdown(aggregates: list[dict], n_gen_errored: int) -> str:
    lines = ["# Retrieval Eval Results", ""]
    n = aggregates[0]["n_queries"] if aggregates else 0
    lines.append(f"- Queries: **{n}**")
    lines.append(f"- Question-generation errors (excluded): **{n_gen_errored}**")
    lines.append("")
    header = "| Engine | n_errored | " + " | ".join(f"Hit@{k}" for k in K_VALUES) + " | MRR |"
    sep = "|" + "---|" * (len(K_VALUES) + 3)
    lines += [header, sep]
    for agg in aggregates:
        hits = " | ".join(f"{agg['hit_at_k'][k]:.3f}" for k in K_VALUES)
        lines.append(
            f"| {agg['engine']} | {agg['n_errored']} | {hits} | {agg['mrr']:.3f} |"
        )
    lines.append("")
    lines.append(
        "> Note: with one gold doc per query, Recall@k == Hit@k. "
        "Engine B scores title/description/filename only (no body), by design."
    )
    return "\n".join(lines) + "\n"


def write_reports(
    out_dir: Path,
    aggregates: list[dict],
    results_by_engine: dict[str, list[EngineResult]],
    n_gen_errored: int,
) -> None:
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / "results.md").write_text(
        render_markdown(aggregates, n_gen_errored), encoding="utf-8"
    )
    with (out_dir / "per_query.jsonl").open("w", encoding="utf-8") as handle:
        for engine, results in results_by_engine.items():
            for r in results:
                handle.write(
                    json.dumps(
                        {
                            "engine": engine, "question": r.question,
                            "gold_doc_id": r.gold_doc_id,
                            "ranked_doc_ids": r.ranked_doc_ids,
                            "errored": r.errored, "error_message": r.error_message,
                        },
                        ensure_ascii=False,
                    )
                    + "\n"
                )
  • Step 2: Verify import

Run (from rag/): python -c "from evals.retrieval.report import render_markdown, write_reports" Expected: exit 0.

  • Step 3: Commit
git add rag/evals/retrieval/report.py
git commit -m "feat(rag-eval): markdown + per-query JSONL reporter"

Task 10: Runner CLI​

Files:

  • Create: rag/evals/retrieval/run.py

  • Step 1: Implement run.py

"""CLI orchestrator: ingest -> generate/replay questions -> run each engine
in isolation -> score -> report. Fail-loud: engine errors per query are
recorded as misses and surfaced, never silently skipped.

Usage (from rag/):
  python -m evals.retrieval.run [--regenerate] [--out DIR] [--no-rerank]
"""

from __future__ import annotations

import argparse
from pathlib import Path

from evals.retrieval.corpus.provider import SeedCorpusProvider
from evals.retrieval.engines.engine_a_haystack import EngineAConfig, HaystackEngine
from evals.retrieval.engines.engine_b_keyword import KeywordEngine
from evals.retrieval.generate_questions import generate_questions
from evals.retrieval.ingest import ephemeral_pgvector
from evals.retrieval.metrics import aggregate
from evals.retrieval.report import K_VALUES, write_reports
from evals.retrieval.types import EngineResult

K_MAX = max(K_VALUES)


def _run_engine(engine, name, queries) -> list[EngineResult]:
    results: list[EngineResult] = []
    for q in queries:
        try:
            ranked = engine.retrieve(q.question, k=K_MAX)
            results.append(
                EngineResult(
                    engine=name, question=q.question,
                    gold_doc_id=q.gold_doc_id, ranked_doc_ids=ranked,
                )
            )
        except Exception as exc:  # noqa: BLE001 - fail-loud: record as error, continue
            print(f"[run] {name} ERROR q={q.question!r}: {exc}")
            results.append(
                EngineResult(
                    engine=name, question=q.question, gold_doc_id=q.gold_doc_id,
                    ranked_doc_ids=[], errored=True, error_message=str(exc),
                )
            )
    return results


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--regenerate", action="store_true", help="force question regeneration")
    parser.add_argument("--no-rerank", action="store_true", help="disable Engine A reranker")
    parser.add_argument("--out", default="evals/retrieval/out", help="report output dir")
    args = parser.parse_args()

    docs = SeedCorpusProvider().get_documents()
    with ephemeral_pgvector(docs) as store:
        queries, n_gen_errored = generate_questions(store.chunks, regenerate=args.regenerate)
        print(f"[run] {len(queries)} queries, {n_gen_errored} generation errors")

        engine_a = HaystackEngine(store.conn_str, EngineAConfig(rerank=not args.no_rerank))
        engine_b = KeywordEngine(docs)

        results_by_engine = {
            engine_a.name: _run_engine(engine_a, engine_a.name, queries),
            engine_b.name: _run_engine(engine_b, engine_b.name, queries),
        }

    aggregates = [
        aggregate(name, results, K_VALUES)
        for name, results in results_by_engine.items()
    ]
    write_reports(Path(args.out), aggregates, results_by_engine, n_gen_errored)
    for agg in aggregates:
        print(f"[run] {agg['engine']}: MRR={agg['mrr']:.3f} Hit@5={agg['hit_at_k'][5]:.3f}")
    print(f"[run] reports written to {args.out}/")


if __name__ == "__main__":
    main()
  • Step 2: Verify the CLI parses and imports

Run (from rag/): python -m evals.retrieval.run --help Expected: argparse usage text, exit 0.

  • Step 3: Commit
git add rag/evals/retrieval/run.py
git commit -m "feat(rag-eval): CLI orchestrator wiring full eval flow"

Task 11: Tiny-fixture wiring self-test​

Files:

  • Create: rag/evals/retrieval/tests/fixtures/tiny_corpus/*.json (3-4 docs)
  • Create: rag/evals/retrieval/tests/test_pipeline_wiring.py

Verifies the full non-LLM, non-pgvector wiring (Engine B + metrics + report) on a tiny fixed corpus, so the orchestration is covered even where docker/keys are absent.

  • Step 1: Author 3-4 tiny fixture docs

Each shaped like the seed docs (Task 5), short content, distinct topics, one with the gold term only in content.

  • Step 2: Write the wiring test
from pathlib import Path

from evals.retrieval.corpus.provider import SeedCorpusProvider
from evals.retrieval.engines.engine_b_keyword import KeywordEngine
from evals.retrieval.metrics import aggregate
from evals.retrieval.report import K_VALUES, render_markdown
from evals.retrieval.types import EngineResult, LabeledQuery

FIX = Path(__file__).with_name("fixtures") / "tiny_corpus"


def test_engine_b_plus_metrics_plus_report_wire_together():
    docs = SeedCorpusProvider(seed_dir=FIX).get_documents()
    assert len(docs) >= 3
    # Build a query whose answer is in doc[0]'s title so Engine B can hit it.
    q = LabeledQuery(question=docs[0].title, gold_doc_id=docs[0].doc_id, source_chunk_id=f"{docs[0].doc_id}:0")
    engine = KeywordEngine(docs)
    ranked = engine.retrieve(q.question, k=max(K_VALUES))
    result = EngineResult(engine="engine_b_keyword", question=q.question,
                          gold_doc_id=q.gold_doc_id, ranked_doc_ids=ranked)
    agg = aggregate("engine_b_keyword", [result], K_VALUES)
    assert agg["n_queries"] == 1
    md = render_markdown([agg], n_gen_errored=0)
    assert "Retrieval Eval Results" in md
    assert "engine_b_keyword" in md
  • Step 3: Run the test

Run (from rag/): pytest evals/retrieval/tests/test_pipeline_wiring.py -v Expected: 1 passed.

  • Step 4: Run the full eval suite

Run (from rag/): pytest evals/retrieval/tests/ -v Expected: all metric/engine-b/corpus/wiring tests pass; the Engine A integration test passes or is skipped.

  • Step 5: Commit
git add rag/evals/retrieval/tests/fixtures rag/evals/retrieval/tests/test_pipeline_wiring.py
git commit -m "test(rag-eval): tiny-fixture wiring self-test"

Task 12: Generate the committed question cache + first baseline run​

Files:

  • Create: rag/evals/retrieval/questions.cache.json (generated artifact, committed)
  • Create: rag/evals/retrieval/README.md

Requires docker + OPENROUTER_API_KEY. Run where that infra exists.

  • Step 1: Run the full eval once to generate the cache + baseline

Run (from rag/, docker + key): python -m evals.retrieval.run Expected: prints query count, per-engine MRR/Hit@5, writes out/results.md + out/per_query.jsonl, and writes questions.cache.json.

  • Step 2: Sanity-check the baseline numbers

Open evals/retrieval/out/results.md. Confirm:

  • Engine A (hybrid+rerank) Hit@5 is meaningfully higher than Engine B on body-only-content docs.

  • Engine B's misses cluster on docs whose distinctive content is body-only (the expected metadata blind spot). If Engine A is NOT clearly ahead, stop and investigate (likely an ingestion or filter bug) before trusting the harness.

  • Step 3: Write a short README

rag/evals/retrieval/README.md documenting: what it measures (Hit@k/MRR per engine), how to run (python -m evals.retrieval.run), env needed (OPENROUTER_API_KEY, optional EVAL_PG_CONN_STR, EVAL_GEN_MODEL), the single-gold caveat, the synthetic lexical-overlap caveat, and the deferred items (LLM judge, real-dev corpus, reranker sweep).

  • Step 4: Commit
git add rag/evals/retrieval/questions.cache.json rag/evals/retrieval/README.md
git commit -m "chore(rag-eval): commit baseline question cache + README"

Self-Review​

Spec coverage:

  • Corpus provider + seed (C, in-repo) → Task 5. Ephemeral pgvector ingest → Task 6. ✓
  • Synthetic question gen (A) + cache → Task 7. ✓
  • Engine A configurable adapter (rerank seam) → Task 8. Engine B Python port → Task 4. ✓
  • Metrics Hit@k/MRR (Recall@k≡Hit@k noted) → Task 3. ✓
  • Reporter (results.md + per_query.jsonl) → Task 9. Runner CLI → Task 10. ✓
  • Fail-loud error handling → Tasks 7 (gen), 10 (run). ✓ Teardown in finally/context-manager → Task 6. ✓
  • Metric unit tests (TDD) → Task 3. Tiny-fixture self-test → Task 11. ✓
  • Deferred items (judge B, human anchor, NDCG, dev corpus, reranker sweep) → documented, seams present (EngineAConfig, CorpusProvider). ✓

Placeholder scan: No TBD/TODO. Corpus/fixture content is authored by the engineer (Tasks 5/11) — these are data-authoring steps with explicit shape + constraints, not code placeholders.

Type consistency: Doc, LabeledQuery, EngineResult (Task 2) used consistently. RetrievalEngine.retrieve(query, k) -> list[str] honored by both engines. aggregate() shape consumed by render_markdown() (hit_at_k dict keyed by k, mrr, n_errored, n_queries). K_VALUES defined in report.py, imported by run.py. ✓

Known live-dependency caveat: Engine A's per-query embedding + rerank are live OpenRouter calls, so Engine A runs are near-deterministic (stable embeddings) but not byte-identical-guaranteed like the pure metric layer. Question set is frozen via the committed cache. Acceptable for v1; engine-output caching is a possible future addition.

Last updated on by Degen