120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""EvaluationRunner: offline fixture-corpus runs with reproducible results."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from ragcore.evaluation import (
|
|
AnswerabilityStatus,
|
|
EvaluationCase,
|
|
EvaluationDataset,
|
|
EvaluationDatasetError,
|
|
EvaluationRunner,
|
|
FixtureCorpusRetriever,
|
|
RetrievalRunProfile,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[3]
|
|
|
|
|
|
def make_dataset() -> tuple[EvaluationDataset, dict]:
|
|
matched_chunk, missing_chunk, unexpected_chunk = uuid4(), uuid4(), uuid4()
|
|
case = EvaluationCase(
|
|
id="case-1",
|
|
question="What is the timeout?",
|
|
language="en",
|
|
expected_answerability=AnswerabilityStatus.ANSWERABLE,
|
|
expected_document_ids=(uuid4(),),
|
|
expected_chunk_ids=(matched_chunk, missing_chunk),
|
|
required_facts=("the timeout is bounded",),
|
|
)
|
|
dataset = EvaluationDataset(id="offline", version="v1", cases=(case,))
|
|
corpus = {"case-1": (matched_chunk, unexpected_chunk)}
|
|
return dataset, {
|
|
"corpus": corpus,
|
|
"matched": matched_chunk,
|
|
"missing": missing_chunk,
|
|
"unexpected": unexpected_chunk,
|
|
}
|
|
|
|
|
|
def make_profile() -> RetrievalRunProfile:
|
|
return RetrievalRunProfile(
|
|
embedding_profile_digest="a" * 64,
|
|
sparse_profile_version="lexical-v1",
|
|
index_alias="rag_dense_hybrid_active",
|
|
index_collection_version="v1",
|
|
)
|
|
|
|
|
|
async def test_runner_works_offline_with_a_fixture_corpus() -> None:
|
|
dataset, ids = make_dataset()
|
|
retriever = FixtureCorpusRetriever(ids["corpus"])
|
|
runner = EvaluationRunner(retriever, clock=lambda: datetime(2026, 8, 2, tzinfo=UTC))
|
|
|
|
run = await runner.run(dataset, run_profile=make_profile())
|
|
|
|
assert run.dataset_id == "offline"
|
|
assert run.dataset_version == "v1"
|
|
assert len(run.case_results) == 1
|
|
result = run.case_results[0]
|
|
assert result.matched_expected_ids == (ids["matched"],)
|
|
assert result.missing_expected_ids == (ids["missing"],)
|
|
assert result.unexpected_ids == (ids["unexpected"],)
|
|
|
|
|
|
async def test_run_records_profile_model_and_index_revisions() -> None:
|
|
dataset, ids = make_dataset()
|
|
runner = EvaluationRunner(FixtureCorpusRetriever(ids["corpus"]))
|
|
profile = RetrievalRunProfile(
|
|
embedding_profile_digest="b" * 64,
|
|
sparse_profile_version="lexical-v2",
|
|
index_alias="rag_dense_hybrid_active",
|
|
index_collection_version="v3",
|
|
reranker_profile_digest="c" * 64,
|
|
)
|
|
|
|
run = await runner.run(dataset, run_profile=profile)
|
|
|
|
assert run.run_profile.embedding_profile_digest == "b" * 64
|
|
assert run.run_profile.sparse_profile_version == "lexical-v2"
|
|
assert run.run_profile.index_collection_version == "v3"
|
|
assert run.run_profile.reranker_profile_digest == "c" * 64
|
|
|
|
|
|
async def test_results_are_reproducible_given_identical_inputs() -> None:
|
|
dataset, ids = make_dataset()
|
|
run_id = uuid4()
|
|
fixed_time = datetime(2026, 8, 2, 9, tzinfo=UTC)
|
|
runner_a = EvaluationRunner(FixtureCorpusRetriever(ids["corpus"]), clock=lambda: fixed_time)
|
|
runner_b = EvaluationRunner(FixtureCorpusRetriever(ids["corpus"]), clock=lambda: fixed_time)
|
|
|
|
run_a = await runner_a.run(dataset, run_profile=make_profile(), run_id=run_id)
|
|
run_b = await runner_b.run(dataset, run_profile=make_profile(), run_id=run_id)
|
|
|
|
assert run_a == run_b
|
|
|
|
|
|
async def test_runner_rejects_a_retriever_exceeding_the_bounded_limit() -> None:
|
|
dataset, _ids = make_dataset()
|
|
oversized = {"case-1": tuple(uuid4() for _ in range(201))}
|
|
# FixtureCorpusRetriever itself would reject this at construction time; a
|
|
# raw retriever proves the runner independently enforces the limit too,
|
|
# regardless of which Reranker-style implementation is plugged in.
|
|
runner = EvaluationRunner(_RawRetriever(oversized))
|
|
|
|
with pytest.raises(EvaluationDatasetError, match="bounded limit"):
|
|
await runner.run(dataset, run_profile=make_profile())
|
|
|
|
|
|
class _RawRetriever:
|
|
def __init__(self, corpus: dict) -> None:
|
|
self._corpus = corpus
|
|
|
|
async def retrieve(self, case: EvaluationCase) -> tuple:
|
|
return self._corpus.get(case.id, ())
|