Evaluation

varagity.eval.datasets

Golden Q&A dataset: loading, validation, and identity resolution (spec §16).

The golden set (data/eval/golden_qa.jsonl) is hand-authored over tests/fixtures/corpus. Each line is one query with the chunks required to answer it, identified portably by relative source path + chunk_index — never by chunk_id directly, whose doc_id half depends on file bytes:

{"query": "…", "relevant": [{"rel_source": "aurora_station.md", "chunk_index": 1}]}

rel_source is the document's POSIX path relative to the eval corpus root — exactly the string the ingest loader hashes into doc_id (plan decision #6) — so an entry resolves to concrete chunk_ids from the corpus files alone: derive_doc_id(rel_source, content_hash(bytes)) plus the deterministic {doc_id}::{chunk_index} composition. No store round-trip is needed, and the resolution is identical on every machine.

chunk_index values assume the chunk boundaries produced by the pinned eval pipeline settings (recursive_character, size 400 / overlap 50 — see varagity.eval.evaluate.PINNED_EVAL_SETTINGS); re-author the golden set if those pins change.

The conversation fixtures (tests/fixtures/conversations/*.json, spec_v3 §4.9) extend the same shape to multi-turn sequences: each turn is a golden entry plus a scripted assistant reply and a kind tag on follow-up turns (pronoun / elliptical / topic_shift). The scripted replies are what both chat engines see as history — identical by construction, so the chat eval compares engines, never their answers.

The graph golden set (data/eval/graph_golden_qa.jsonl, spec_graphrag §12) anchors differently again, because a graph engine returns prose rather than chunk ids: each entry carries expected_facts (an AND of OR-groups matched case-insensitively against the answer — :func:resolve_golden_by_fact's substring trick applied to answers instead of chunks) and required_guids (the messages a provenance-surfacing engine should cite). Its kind slices the report into the question shapes the bake-off discriminates on and is golden metadata only — never shown to an engine.

ConversationFixture

Bases: BaseModel

One hand-authored multi-turn conversation over the eval corpus.

Attributes:

Name	Type	Description
name	str	Unique fixture identity in reports.
turns	list[ConversationTurn]	The conversation's user turns, in order (≥ 2 — a single-turn conversation exercises no history).

Source code in varagity/eval/datasets.py

class ConversationFixture(BaseModel):
    """One hand-authored multi-turn conversation over the eval corpus.

    Attributes:
        name: Unique fixture identity in reports.
        turns: The conversation's user turns, in order (≥ 2 — a
            single-turn conversation exercises no history).
    """

    name: str = Field(min_length=1)
    turns: list[ConversationTurn] = Field(min_length=2)

    @model_validator(mode="after")
    def _check_turn_contract(self) -> "ConversationFixture":
        """Enforce the anchor/follow-up structure the chat eval relies on.

        Returns:
            The validated fixture.

        Raises:
            ValueError: If the first turn carries a ``kind`` (it is
                standalone by construction — both engines behave
                identically on it, so tagging it would dilute a follow-up
                bucket), a later turn lacks one, or a non-final turn lacks
                its scripted ``assistant`` reply.
        """
        if self.turns[0].kind is not None:
            raise ValueError(
                f"conversation {self.name!r}: the first turn is standalone by "
                "construction and must not carry a kind tag"
            )
        for index, turn in enumerate(self.turns):
            if index > 0 and turn.kind is None:
                raise ValueError(
                    f"conversation {self.name!r}: turn {index} is a follow-up and "
                    f"needs a kind tag (one of {CONVERSATION_KINDS})"
                )
            if index < len(self.turns) - 1 and turn.assistant is None:
                raise ValueError(
                    f"conversation {self.name!r}: turn {index} has later turns and "
                    "needs a scripted assistant reply for their history"
                )
        return self

ConversationTurn

Bases: BaseModel

One user turn of a multi-turn eval conversation (spec_v3 §4.9).

Attributes:

Name	Type	Description
query	str	The user's question for this turn, exactly as typed — follow-up turns deliberately keep their pronouns and ellipses; resolving them is what the eval measures.
kind	Literal['pronoun', 'elliptical', 'topic_shift'] | None	The follow-up case this turn exercises (one of :data:CONVERSATION_KINDS). None on a conversation's first turn, which is standalone by construction; required on every later turn so the by-kind report has no untagged follow-ups.
assistant	str | None	The scripted assistant reply appended to the history after this turn. Required on every non-final turn (the condenser resolves references against it); optional on the final turn, whose reply nothing reads.
relevant	list[GoldenChunkRef]	The chunks a perfect retriever would return for this turn (≥ 1), in the golden set's portable ref shape.

Source code in varagity/eval/datasets.py

class ConversationTurn(BaseModel):
    """One user turn of a multi-turn eval conversation (spec_v3 §4.9).

    Attributes:
        query: The user's question for this turn, exactly as typed —
            follow-up turns deliberately keep their pronouns and ellipses;
            resolving them is what the eval measures.
        kind: The follow-up case this turn exercises (one of
            :data:`CONVERSATION_KINDS`). ``None`` on a conversation's first
            turn, which is standalone by construction; required on every
            later turn so the by-kind report has no untagged follow-ups.
        assistant: The scripted assistant reply appended to the history
            after this turn. Required on every non-final turn (the
            condenser resolves references against it); optional on the
            final turn, whose reply nothing reads.
        relevant: The chunks a perfect retriever would return for this
            turn (≥ 1), in the golden set's portable ref shape.
    """

    query: str = Field(min_length=1)
    kind: Literal["pronoun", "elliptical", "topic_shift"] | None = None
    assistant: str | None = Field(default=None, min_length=1)
    relevant: list[GoldenChunkRef] = Field(min_length=1)

GoldenChunkRef

Bases: BaseModel

Portable identity of one relevant chunk.

chunk_index anchors the ref to the pinned chunk boundaries (the matrix's fixed configuration); fact anchors it to content, which survives any boundary change — the chunker sweep re-resolves refs by scanning the strategy-true chunks for the fact (spec_v2 §7.4), since an index authored against one strategy's boundaries points at arbitrary text under another's.

Attributes:

Name	Type	Description
rel_source	str	Document path relative to the eval corpus root (POSIX-style, e.g. "aurora_station.md" — the same string the loader derives doc_id from).
chunk_index	int	The chunk's position within its document (0-based), under the pinned eval chunk boundaries.
fact	str | None	A short verbatim snippet of the chunk's source text that uniquely identifies the golden content within its document (matched case-insensitively — OCR extraction may case-shift). Optional for the fixed matrix; required for a ref to participate in the chunker sweep.

Source code in varagity/eval/datasets.py

class GoldenChunkRef(BaseModel):
    """Portable identity of one relevant chunk.

    ``chunk_index`` anchors the ref to the **pinned** chunk boundaries (the
    matrix's fixed configuration); ``fact`` anchors it to **content**, which
    survives any boundary change — the chunker sweep re-resolves refs by
    scanning the strategy-true chunks for the fact (spec_v2 §7.4), since an
    index authored against one strategy's boundaries points at arbitrary
    text under another's.

    Attributes:
        rel_source: Document path relative to the eval corpus root
            (POSIX-style, e.g. ``"aurora_station.md"`` — the same string
            the loader derives ``doc_id`` from).
        chunk_index: The chunk's position within its document (0-based),
            under the pinned eval chunk boundaries.
        fact: A short verbatim snippet of the chunk's source text that
            uniquely identifies the golden content within its document
            (matched case-insensitively — OCR extraction may case-shift).
            Optional for the fixed matrix; required for a ref to
            participate in the chunker sweep.
    """

    rel_source: str = Field(min_length=1)
    chunk_index: int = Field(ge=0)
    fact: str | None = Field(default=None, min_length=1)

GoldenEntry

Bases: BaseModel

One golden query and every chunk required to answer it.

Attributes:

Name	Type	Description
query	str	The evaluation question, as a user would ask it.
relevant	list[GoldenChunkRef]	The chunks a perfect retriever would return (≥ 1).

Source code in varagity/eval/datasets.py

class GoldenEntry(BaseModel):
    """One golden query and every chunk required to answer it.

    Attributes:
        query: The evaluation question, as a user would ask it.
        relevant: The chunks a perfect retriever would return (≥ 1).
    """

    query: str = Field(min_length=1)
    relevant: list[GoldenChunkRef] = Field(min_length=1)

GraphGoldenEntry

Bases: BaseModel

One golden question over the synthetic message corpus (spec_graphrag §12).

Scoring is fact-anchored and house style — no LLM judge (plan decision

10). :attr:expected_facts is an AND of OR-groups: every group

must be satisfied, and a group is satisfied when any of its variants is a case-insensitive substring of the engine's answer. Variants exist so that a right answer phrased differently still scores ("march 14" / "3/14"), not so that a wrong one squeaks through — keep them specific enough that a distractor-fooled answer fails.

Attributes:

Name	Type	Description
query	str	The question, exactly as it is put to the engine.
kind	Literal['aggregation', 'verification', 'relation']	The question shape this entry exercises (one of :data:GRAPH_GOLDEN_KINDS) — report metadata only; engines never see it.
expected_facts	list[list[str]]	One OR-group of acceptable variants per fact the answer must contain (≥ 1 group, each ≥ 1 non-blank variant).
required_guids	list[str]	Message guids a perfect provenance answer would cite (≥ 1). Engines that surface no source ids score null provenance rather than zero — the honesty criterion.
notes	str | None	Why the entry is discriminative — which distractors it is built to defeat. Authoring documentation, never scored.

Source code in varagity/eval/datasets.py

class GraphGoldenEntry(BaseModel):
    """One golden question over the synthetic message corpus (spec_graphrag §12).

    Scoring is fact-anchored and house style — no LLM judge (plan decision
    #10). :attr:`expected_facts` is an **AND of OR-groups**: every group
    must be satisfied, and a group is satisfied when any of its variants is
    a case-insensitive substring of the engine's answer. Variants exist so
    that a right answer phrased differently still scores (``"march 14"`` /
    ``"3/14"``), not so that a wrong one squeaks through — keep them
    specific enough that a distractor-fooled answer fails.

    Attributes:
        query: The question, exactly as it is put to the engine.
        kind: The question shape this entry exercises (one of
            :data:`GRAPH_GOLDEN_KINDS`) — report metadata only; engines
            never see it.
        expected_facts: One OR-group of acceptable variants per fact the
            answer must contain (≥ 1 group, each ≥ 1 non-blank variant).
        required_guids: Message guids a perfect provenance answer would
            cite (≥ 1). Engines that surface no source ids score ``null``
            provenance rather than zero — the honesty criterion.
        notes: Why the entry is discriminative — which distractors it is
            built to defeat. Authoring documentation, never scored.
    """

    query: str = Field(min_length=1)
    kind: Literal["aggregation", "verification", "relation"]
    expected_facts: list[list[str]] = Field(min_length=1)
    required_guids: list[str] = Field(min_length=1)
    notes: str | None = Field(default=None, min_length=1)

    @model_validator(mode="after")
    def _check_fact_groups(self) -> "GraphGoldenEntry":
        """Reject empty OR-groups and blank variants.

        Returns:
            The validated entry.

        Raises:
            ValueError: If a group holds no variants (it could never be
                satisfied, guaranteeing a 0 score for every engine) or a
                variant is blank (which every answer trivially contains).
        """
        for index, group in enumerate(self.expected_facts):
            if not group:
                raise ValueError(
                    f"golden query {self.query!r}: fact group {index} is empty — "
                    "an unsatisfiable group scores every engine 0"
                )
            for variant in group:
                if not variant.strip():
                    raise ValueError(
                        f"golden query {self.query!r}: fact group {index} has a blank "
                        "variant, which every answer trivially matches"
                    )
        return self

ResolvedGoldenEntry

Bases: BaseModel

A golden entry with its refs resolved to concrete chunk_ids.

Attributes:

Name	Type	Description
query	str	The evaluation question.
relevant	list[GoldenChunkRef]	The original portable refs (kept for reporting).
chunk_ids	list[str]	One resolved {doc_id}::{chunk_index} per ref, in ref order.

Source code in varagity/eval/datasets.py

class ResolvedGoldenEntry(BaseModel):
    """A golden entry with its refs resolved to concrete ``chunk_id``s.

    Attributes:
        query: The evaluation question.
        relevant: The original portable refs (kept for reporting).
        chunk_ids: One resolved ``{doc_id}::{chunk_index}`` per ref, in
            ref order.
    """

    query: str
    relevant: list[GoldenChunkRef]
    chunk_ids: list[str]

load_conversations(directory)

Load and validate every conversation fixture in a directory.

Files are read in sorted name order (the report order); any malformed fixture fails the load loudly, mirroring :func:load_golden — a silently skipped conversation would quietly inflate every score.

Parameters:

Name	Type	Description	Default
directory	Path	The fixtures directory (one .json file per conversation).	required

Returns:

Type	Description
list[ConversationFixture]	The validated fixtures, in file-name order.

Raises:

Type	Description
FileNotFoundError	If directory does not exist.
ValueError	If it holds no .json files, a file is not valid JSON, a fixture fails validation (message names the file), or two fixtures share a name.

Source code in varagity/eval/datasets.py

def load_conversations(directory: Path) -> list[ConversationFixture]:
    """Load and validate every conversation fixture in a directory.

    Files are read in sorted name order (the report order); any malformed
    fixture fails the load loudly, mirroring :func:`load_golden` — a
    silently skipped conversation would quietly inflate every score.

    Args:
        directory: The fixtures directory (one ``.json`` file per
            conversation).

    Returns:
        The validated fixtures, in file-name order.

    Raises:
        FileNotFoundError: If ``directory`` does not exist.
        ValueError: If it holds no ``.json`` files, a file is not valid
            JSON, a fixture fails validation (message names the file), or
            two fixtures share a name.
    """
    if not directory.is_dir():
        raise FileNotFoundError(f"conversation fixtures directory {directory} does not exist")
    paths = sorted(directory.glob("*.json"))
    if not paths:
        raise ValueError(f"{directory}: no conversation fixtures (*.json) found")
    conversations: list[ConversationFixture] = []
    seen: dict[str, Path] = {}
    for path in paths:
        try:
            fixture = ConversationFixture.model_validate(
                json.loads(path.read_text(encoding="utf-8"))
            )
        except json.JSONDecodeError as exc:
            raise ValueError(f"{path}: not valid JSON — {exc}") from exc
        except ValidationError as exc:
            raise ValueError(f"{path}: invalid conversation fixture — {exc}") from exc
        if fixture.name in seen:
            raise ValueError(
                f"{path}: conversation name {fixture.name!r} already used by {seen[fixture.name]}"
            )
        seen[fixture.name] = path
        conversations.append(fixture)
    logger.info("loaded %d conversation fixture(s) from %s", len(conversations), directory)
    return conversations

load_golden(path)

Load and validate the golden dataset from a JSONL file.

Blank lines are ignored; any malformed line fails the load loudly — a silently skipped golden query would quietly inflate every score.

Parameters:

Name	Type	Description	Default
path	Path	The golden_qa.jsonl file.	required

Returns:

Type	Description
list[GoldenEntry]	The validated entries, in file order.

Raises:

Type	Description
FileNotFoundError	If path does not exist.
ValueError	If a line is not valid JSON or fails schema validation (message includes the 1-based line number).

Source code in varagity/eval/datasets.py

def load_golden(path: Path) -> list[GoldenEntry]:
    """Load and validate the golden dataset from a JSONL file.

    Blank lines are ignored; any malformed line fails the load loudly —
    a silently skipped golden query would quietly inflate every score.

    Args:
        path: The ``golden_qa.jsonl`` file.

    Returns:
        The validated entries, in file order.

    Raises:
        FileNotFoundError: If ``path`` does not exist.
        ValueError: If a line is not valid JSON or fails schema validation
            (message includes the 1-based line number).
    """
    entries: list[GoldenEntry] = []
    for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        try:
            entries.append(GoldenEntry.model_validate(json.loads(line)))
        except json.JSONDecodeError as exc:
            raise ValueError(f"{path}:{line_no}: not valid JSON — {exc}") from exc
        except ValidationError as exc:
            raise ValueError(f"{path}:{line_no}: invalid golden entry — {exc}") from exc
    if not entries:
        raise ValueError(f"{path}: golden dataset is empty")
    logger.info("loaded %d golden queries from %s", len(entries), path)
    return entries

load_graph_golden(path)

Load and validate the graph golden dataset from a JSONL file.

Mirrors :func:load_golden — blank lines ignored, any malformed line fails the load loudly with its 1-based line number — and additionally rejects a repeated query, which would silently double-weight one question in the per-kind averages.

Parameters:

Name	Type	Description	Default
path	Path	The graph_golden_qa.jsonl file.	required

Returns:

Type	Description
list[GraphGoldenEntry]	The validated entries, in file order.

Raises:

Type	Description
FileNotFoundError	If path does not exist.
ValueError	If a line is not valid JSON, fails schema validation, or repeats an earlier entry's query, or if the file holds no entries.

Source code in varagity/eval/datasets.py

def load_graph_golden(path: Path) -> list[GraphGoldenEntry]:
    """Load and validate the graph golden dataset from a JSONL file.

    Mirrors :func:`load_golden` — blank lines ignored, any malformed line
    fails the load loudly with its 1-based line number — and additionally
    rejects a repeated ``query``, which would silently double-weight one
    question in the per-kind averages.

    Args:
        path: The ``graph_golden_qa.jsonl`` file.

    Returns:
        The validated entries, in file order.

    Raises:
        FileNotFoundError: If ``path`` does not exist.
        ValueError: If a line is not valid JSON, fails schema validation, or
            repeats an earlier entry's query, or if the file holds no
            entries.
    """
    entries: list[GraphGoldenEntry] = []
    seen: dict[str, int] = {}
    for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        try:
            entry = GraphGoldenEntry.model_validate(json.loads(line))
        except json.JSONDecodeError as exc:
            raise ValueError(f"{path}:{line_no}: not valid JSON — {exc}") from exc
        except ValidationError as exc:
            raise ValueError(f"{path}:{line_no}: invalid graph golden entry — {exc}") from exc
        if entry.query in seen:
            raise ValueError(
                f"{path}:{line_no}: query {entry.query!r} already appears on line "
                f"{seen[entry.query]}"
            )
        seen[entry.query] = line_no
        entries.append(entry)
    if not entries:
        raise ValueError(f"{path}: graph golden dataset is empty")
    logger.info("loaded %d graph golden queries from %s", len(entries), path)
    return entries

resolve_golden(entries, corpus_root)

Resolve portable refs to the chunk_ids the stores will hold.

Reproduces the loader's identity derivation (plan decision #6) from the corpus files themselves: doc_id hashes rel_source together with the file's byte hash, so resolution needs no ingested store and is stable across machines and OCR engines (bytes, not extracted text).

Whether a resolved chunk_id actually exists (i.e. chunk_index is within the document's real chunk count) is checked by the harness after ingest — it depends on the live chunking, not on this file.

Parameters:

Name	Type	Description	Default
entries	list[GoldenEntry]	The validated golden entries.	required
corpus_root	Path	The eval corpus root the refs are relative to.	required

Returns:

Type	Description
list[ResolvedGoldenEntry]	One resolved entry per input entry, in order.

Raises:

Type	Description
FileNotFoundError	If a ref names a file missing from the corpus.

Source code in varagity/eval/datasets.py

def resolve_golden(entries: list[GoldenEntry], corpus_root: Path) -> list[ResolvedGoldenEntry]:
    """Resolve portable refs to the ``chunk_id``s the stores will hold.

    Reproduces the loader's identity derivation (plan decision #6) from
    the corpus files themselves: ``doc_id`` hashes ``rel_source`` together
    with the file's byte hash, so resolution needs no ingested store and
    is stable across machines and OCR engines (bytes, not extracted text).

    Whether a resolved ``chunk_id`` actually exists (i.e. ``chunk_index``
    is within the document's real chunk count) is checked by the harness
    after ingest — it depends on the live chunking, not on this file.

    Args:
        entries: The validated golden entries.
        corpus_root: The eval corpus root the refs are relative to.

    Returns:
        One resolved entry per input entry, in order.

    Raises:
        FileNotFoundError: If a ref names a file missing from the corpus.
    """
    doc_ids: dict[str, str] = {}
    resolved: list[ResolvedGoldenEntry] = []
    for entry in entries:
        chunk_ids: list[str] = []
        for ref in entry.relevant:
            if ref.rel_source not in doc_ids:
                source = corpus_root / ref.rel_source
                if not source.is_file():
                    raise FileNotFoundError(
                        f"golden ref {ref.rel_source!r} not found under {corpus_root} — "
                        "the golden set and the eval corpus are out of sync"
                    )
                doc_ids[ref.rel_source] = derive_doc_id(
                    ref.rel_source, content_hash(source.read_bytes())
                )
            chunk_ids.append(f"{doc_ids[ref.rel_source]}::{ref.chunk_index}")
        resolved.append(
            ResolvedGoldenEntry(query=entry.query, relevant=entry.relevant, chunk_ids=chunk_ids)
        )
    return resolved

varagity.eval.containers

Ephemeral backing stores via testcontainers (plan decision #4).

The eval harness measures against throwaway Postgres/Elasticsearch containers so the live corpus is never touched; the integration and e2e test suites spin the same containers. This module is the single home for that setup — image tags, schema application, and the single-node Elasticsearch tuning — shared by both consumers.

testcontainers lives in the eval dependency group (installed with the dev group, or via uv run --group eval); its import is deferred to call time so importing this module — which the CLI does on every start via the eval wiring — never requires it.

Operational notes baked in:

• The Elasticsearch container disables the disk-watermark allocation checks: on a host whose disk is >90% full, the default percentage watermarks refuse to allocate the throwaway index's primary shard and every operation times out. Ephemeral stores must never depend on host disk pressure.
• Single-node Elasticsearch is yellow by design; nothing here waits for green.

EphemeralStores dataclass

Connected store clients over the throwaway containers.

Attributes:

Name	Type	Description
store	ContextualVectorDB	Vector store on the ephemeral Postgres.
bm25	ElasticsearchBM25	BM25 store on the ephemeral Elasticsearch (index not yet created — the ingest loader creates it idempotently).
pg_conninfo	str	The Postgres conninfo, for raw assertions.
es_url	str	The Elasticsearch base URL, for raw assertions.

Source code in varagity/eval/containers.py

@dataclass(frozen=True)
class EphemeralStores:
    """Connected store clients over the throwaway containers.

    Attributes:
        store: Vector store on the ephemeral Postgres.
        bm25: BM25 store on the ephemeral Elasticsearch (index not yet
            created — the ingest loader creates it idempotently).
        pg_conninfo: The Postgres conninfo, for raw assertions.
        es_url: The Elasticsearch base URL, for raw assertions.
    """

    store: ContextualVectorDB
    bm25: ElasticsearchBM25
    pg_conninfo: str
    es_url: str

ephemeral_elasticsearch()

Run a throwaway single-node Elasticsearch (BM25 store backing).

Yields:

Type	Description
str	The container's base URL.

Raises:

Type	Description
ModuleNotFoundError	If testcontainers is not installed (the eval dependency group).

Source code in varagity/eval/containers.py

@contextmanager
def ephemeral_elasticsearch() -> Iterator[str]:
    """Run a throwaway single-node Elasticsearch (BM25 store backing).

    Yields:
        The container's base URL.

    Raises:
        ModuleNotFoundError: If ``testcontainers`` is not installed (the
            ``eval`` dependency group).
    """
    from testcontainers.elasticsearch import ElasticSearchContainer

    container = (
        ElasticSearchContainer(ES_IMAGE, mem_limit="2g")
        .with_env("discovery.type", "single-node")
        .with_env("ES_JAVA_OPTS", "-Xms512m -Xmx512m")
        # See the module docstring: never depend on host disk pressure.
        .with_env("cluster.routing.allocation.disk.threshold_enabled", "false")
    )
    with container as es:
        url = f"http://{es.get_container_host_ip()}:{es.get_exposed_port(9200)}"
        logger.info("ephemeral Elasticsearch up at %s", url)
        yield url

ephemeral_postgres()

Run a throwaway pgvector Postgres with schema.sql applied.

Yields:

Type	Description
str	A libpq conninfo string for the container.

Raises:

Type	Description
ModuleNotFoundError	If testcontainers is not installed (the eval dependency group).

Source code in varagity/eval/containers.py

@contextmanager
def ephemeral_postgres() -> Iterator[str]:
    """Run a throwaway pgvector Postgres with ``schema.sql`` applied.

    Yields:
        A libpq conninfo string for the container.

    Raises:
        ModuleNotFoundError: If ``testcontainers`` is not installed (the
            ``eval`` dependency group).
    """
    from testcontainers.postgres import PostgresContainer

    with PostgresContainer(POSTGRES_IMAGE) as container:
        conninfo = psycopg.conninfo.make_conninfo(
            host=container.get_container_host_ip(),
            port=int(container.get_exposed_port(5432)),
            dbname=container.dbname,
            user=container.username,
            password=container.password,
        )
        with psycopg.connect(conninfo, autocommit=True) as conn:
            conn.execute(SCHEMA_PATH.read_text())
        logger.info("ephemeral pgvector Postgres up (schema applied)")
        yield conninfo

ephemeral_stores(index_name='varagity_eval_bm25')

Run both throwaway containers and yield connected store clients.

Parameters:

Name	Type	Description	Default
index_name	str	BM25 index name inside the ephemeral Elasticsearch.	'varagity_eval_bm25'

Yields:

Type	Description
EphemeralStores	The connected stores (closed, with their containers stopped, on
EphemeralStores	exit).

Raises:

Type	Description
ModuleNotFoundError	If testcontainers is not installed (the eval dependency group).

Source code in varagity/eval/containers.py

@contextmanager
def ephemeral_stores(index_name: str = "varagity_eval_bm25") -> Iterator[EphemeralStores]:
    """Run both throwaway containers and yield connected store clients.

    Args:
        index_name: BM25 index name inside the ephemeral Elasticsearch.

    Yields:
        The connected stores (closed, with their containers stopped, on
        exit).

    Raises:
        ModuleNotFoundError: If ``testcontainers`` is not installed (the
            ``eval`` dependency group).
    """
    with ExitStack() as stack:
        conninfo = stack.enter_context(ephemeral_postgres())
        es_url = stack.enter_context(ephemeral_elasticsearch())
        store = stack.enter_context(ContextualVectorDB(conninfo))
        bm25 = stack.enter_context(ElasticsearchBM25(url=es_url, index_name=index_name))
        yield EphemeralStores(store=store, bm25=bm25, pg_conninfo=conninfo, es_url=es_url)

varagity.eval.evaluate

Retrieval evaluation: recall@k / pass@k over the 7-config matrix (spec §16).

Quantifies the retrieval-quality ladder, on the golden set over tests/fixtures/corpus:

1. semantic_noncontextual — the vanilla-RAG v1 baseline,
2. semantic_contextual — contextual embeddings (≈35% tier),
3. bm25_contextual — contextual BM25,
4. hybrid_contextual — rank fusion (≈49% tier, the default),
5. hybrid_rerank_contextual — + cross-encoder re-ranking (≈67% tier, spec_v2 §5.5),
6. hyde_contextual — HyDE over the same hybrid base (ADR-016): the dense arm searches with an LLM-written hypothetical passage,
7. hyde_rerank_contextual — the HyDE+rerank pairing (reranked composing hyde): HyDE steers the candidate pool, the cross-encoder judges the real query.

The chunker sweep (spec_v2 §7.4) follows the matrix: every registered chunking strategy gets its own contextual ingest over the same ephemeral stores and is measured across the contextual retrieval configs. Because the golden set's chunk_index refs are authored against the pinned default boundaries, the sweep re-resolves each ref by content: a ref's fact snippet is located in the strategy-true chunks (:func:resolve_golden_by_fact), and a ref counts as retrieved when any chunk containing its fact is in the top-k — index-anchored scoring under foreign boundaries would measure nothing but the boundary mismatch.

Measurement runs against ephemeral testcontainers stores (plan decision

4) so the live corpus is never touched, while embeddings, the

contextualizing LLM, and the reranker are the real GPU services from the running compose stack (they are stateless). Two ingests cover all seven configs: ingest A (CONTEXTUALIZE=false) backs config 1; ingest B (CONTEXTUALIZE=true, reingest over the same stores) backs configs 2–7 against the same index — and doubles as the sweep's recursive_character row. Each remaining strategy adds one reingest. The two HyDE configs additionally pay one live-LLM passage generation per query each (independent generations — sampling noise between configs 6 and 7 is part of what the pairing comparison measures).

Metric semantics: :func:recall_at_k is the Anthropic cookbook's evaluation number (there called Pass@n) — the per-query fraction of golden chunks present in the top-k, averaged over queries. :func:pass_at_k is the stricter complement reported alongside it: the fraction of queries whose golden chunks are all in the top-k. The sweep's :func:recall_at_k_any / :func:pass_at_k_any are the same two numbers with each golden ref generalized from one chunk id to an acceptable set (fact-containing chunks).

The eval pipeline settings are pinned (:data:PINNED_EVAL_SETTINGS) rather than inherited from .env: golden chunk_index values assume those chunk boundaries, and pinning keeps runs comparable across machines. Pins are applied by exporting environment variables and clearing the settings cache — the same mechanism the test suite uses — because deep pipeline code resolves get_settings() internally; the eval harness is a single-threaded CLI path, where this is safe.

The chat-engine eval (spec_v3 §4.9) is the multi-turn counterpart: every registered chat engine runs the hand-authored conversation fixtures (tests/fixtures/conversations/), and each turn's retrieval is scored fact-anchored exactly like the sweep — the engine's search_query (the condensed rewrite, under condense_context) drives retrieval, and a turn's ref is satisfied when any fact-carrying chunk lands in the top-k. Assistant replies are scripted in the fixtures, so both engines see byte-identical history: the comparison isolates the engine, never its answers. Depths are :data:CHAT_K_VALUES, shallower than the matrix's: the eval corpus ingests ~16 chunks, so recall@20 is 1.0 by definition and k=5 is the production context cut (RERANK_TOP_N).

FactRef

Bases: BaseModel

One golden ref resolved to the chunks that satisfy it (sweep form).

Attributes:

Name	Type	Description
label	str	Human-readable identity in reports — the ref's fact (or its index-anchored chunk_id for a fact-less ref).
chunk_ids	list[str]	Every chunk id whose content contains the fact under the currently ingested boundaries (overlap can spread a fact over several); retrieval of any of them satisfies the ref. Empty when the fact matched nothing — a guaranteed miss.

Source code in varagity/eval/evaluate.py

class FactRef(BaseModel):
    """One golden ref resolved to the chunks that satisfy it (sweep form).

    Attributes:
        label: Human-readable identity in reports — the ref's ``fact``
            (or its index-anchored ``chunk_id`` for a fact-less ref).
        chunk_ids: Every chunk id whose content contains the fact under the
            currently ingested boundaries (overlap can spread a fact over
            several); retrieval of **any** of them satisfies the ref. Empty
            when the fact matched nothing — a guaranteed miss.
    """

    label: str
    chunk_ids: list[str]

FactResolvedEntry

Bases: BaseModel

A golden entry re-resolved by content for one chunking strategy.

Attributes:

Name	Type	Description
query	str	The evaluation question.
refs	list[FactRef]	One :class:FactRef per golden ref, in ref order.

Source code in varagity/eval/evaluate.py

class FactResolvedEntry(BaseModel):
    """A golden entry re-resolved by content for one chunking strategy.

    Attributes:
        query: The evaluation question.
        refs: One :class:`FactRef` per golden ref, in ref order.
    """

    query: str
    refs: list[FactRef]

measure_chat_engine(engine, conversations, fact_entries, retrievers, *, llm, k_values=CHAT_K_VALUES)

Run every conversation through one chat engine and score each turn.

Per turn: the engine prepares the search query against the scripted history (one condense LLM call per follow-up under condense_context; none under simple), then that one search_query is scored under every retrieval config by swapping it into the turn's fact-resolved entry and reusing :func:measure_retriever_facts. History threads the fixture's scripted replies, so every engine sees the identical conversation — the comparison isolates the engine. Engine rendering is silenced (verbose=0): the harness reports rewrites itself, and per-turn console output across engines × turns would swamp it.

Parameters:

Name	Type	Description	Default
engine	ChatEngine	The chat engine under measurement.	required
conversations	Sequence[ConversationFixture]	The fixtures, in report order.	required
fact_entries	Sequence[FactResolvedEntry]	One fact-resolved entry per turn, flattened in conversation order (the fixtures' turns, depth-first).	required
retrievers	Mapping[str, Retriever]	Retrieval configs to score under, keyed by name.	required
llm	LLMClient | None	Chat client for engines that condense (the live service); None lets them resolve one via the model registry.	required
k_values	Sequence[int]	Depths to report.	CHAT_K_VALUES

Returns:

Type	Description
dict[str, Any]	The engine's results: "summary" (per retrieval config: the
dict[str, Any]	"all" / "follow_up" / "by_kind" slice averages —
dict[str, Any]	follow-ups are the discriminating slice, first turns being
dict[str, Any]	identity splits under every engine), "condense" (call count
dict[str, Any]	and latency stats), and "conversations" (per-turn detail:
dict[str, Any]	the search query used, per-config scores, and each ref's best
dict[str, Any]	golden rank).

Raises:

Type	Description
ValueError	If conversations, retrievers or k_values is empty, or fact_entries doesn't match the fixtures' total turn count.

Source code in varagity/eval/evaluate.py

def measure_chat_engine(
    engine: ChatEngine,
    conversations: Sequence[ConversationFixture],
    fact_entries: Sequence[FactResolvedEntry],
    retrievers: Mapping[str, Retriever],
    *,
    llm: LLMClient | None,
    k_values: Sequence[int] = CHAT_K_VALUES,
) -> dict[str, Any]:
    """Run every conversation through one chat engine and score each turn.

    Per turn: the engine prepares the search query against the scripted
    history (one condense LLM call per follow-up under
    ``condense_context``; none under ``simple``), then that **one**
    ``search_query`` is scored under every retrieval config by swapping it
    into the turn's fact-resolved entry and reusing
    :func:`measure_retriever_facts`. History threads the fixture's
    scripted replies, so every engine sees the identical conversation —
    the comparison isolates the engine. Engine rendering is silenced
    (``verbose=0``): the harness reports rewrites itself, and per-turn
    console output across engines × turns would swamp it.

    Args:
        engine: The chat engine under measurement.
        conversations: The fixtures, in report order.
        fact_entries: One fact-resolved entry per turn, flattened in
            conversation order (the fixtures' turns, depth-first).
        retrievers: Retrieval configs to score under, keyed by name.
        llm: Chat client for engines that condense (the live service);
            ``None`` lets them resolve one via the model registry.
        k_values: Depths to report.

    Returns:
        The engine's results: ``"summary"`` (per retrieval config: the
        ``"all"`` / ``"follow_up"`` / ``"by_kind"`` slice averages —
        follow-ups are the discriminating slice, first turns being
        identity splits under every engine), ``"condense"`` (call count
        and latency stats), and ``"conversations"`` (per-turn detail:
        the search query used, per-config scores, and each ref's best
        golden rank).

    Raises:
        ValueError: If ``conversations``, ``retrievers`` or ``k_values``
            is empty, or ``fact_entries`` doesn't match the fixtures'
            total turn count.
    """
    if not conversations:
        raise ValueError("measure_chat_engine needs at least one conversation")
    if not retrievers:
        raise ValueError("measure_chat_engine needs at least one retriever")
    if not k_values:
        raise ValueError("measure_chat_engine needs at least one k value")
    n_turns = sum(len(conversation.turns) for conversation in conversations)
    if len(fact_entries) != n_turns:
        raise ValueError(
            f"got {len(fact_entries)} fact-resolved entries for {n_turns} conversation turns"
        )

    flat_index = 0
    conversation_results: list[dict[str, Any]] = []
    all_records: list[dict[str, Any]] = []
    for conversation in conversations:
        history: list[Turn] = []
        turn_records: list[dict[str, Any]] = []
        for turn_index, turn in enumerate(conversation.turns):
            entry = fact_entries[flat_index]
            flat_index += 1
            prepared = engine.prepare(turn.query, history=history, llm=llm, verbose=0)
            methods: dict[str, Any] = {}
            for name, retriever in retrievers.items():
                scored = entry.model_copy(update={"query": prepared.search_query})
                summary, ranks = measure_retriever_facts(retriever, [scored], k_values=k_values)
                methods[name] = {
                    "recall": summary["recall"],
                    "pass": summary["pass"],
                    "golden_ranks": ranks[0],
                }
            record: dict[str, Any] = {
                "turn": turn_index,
                "kind": turn.kind,
                "query": turn.query,
                "search_query": prepared.search_query,
                "condensed": prepared.condensed,
                "condense_latency_s": (
                    round(prepared.condense_latency_s, 3)
                    if prepared.condense_latency_s is not None
                    else None
                ),
                "methods": methods,
            }
            turn_records.append(record)
            all_records.append(record)
            if turn_index < len(conversation.turns) - 1:
                history.append(Turn(role="user", content=turn.query))
                # Validation guarantees a scripted reply on non-final turns.
                history.append(Turn(role="assistant", content=turn.assistant or ""))
        conversation_results.append({"name": conversation.name, "turns": turn_records})

    follow_ups = [record for record in all_records if record["turn"] > 0]
    summary_by_method: dict[str, Any] = {}
    for name in retrievers:
        by_kind = {
            kind: _mean_turn_scores(selected, name, k_values)
            for kind in CONVERSATION_KINDS
            if (selected := [r for r in follow_ups if r["kind"] == kind])
        }
        summary_by_method[name] = {
            "all": _mean_turn_scores(all_records, name, k_values),
            "follow_up": _mean_turn_scores(follow_ups, name, k_values),
            "by_kind": by_kind,
        }
    latencies = [
        record["condense_latency_s"]
        for record in all_records
        if record["condense_latency_s"] is not None
    ]
    condense = {
        "calls": len(latencies),
        "mean_latency_s": round(sum(latencies) / len(latencies), 3) if latencies else None,
        "max_latency_s": max(latencies) if latencies else None,
    }
    return {
        "summary": summary_by_method,
        "condense": condense,
        "conversations": conversation_results,
    }

measure_retriever(retriever, entries, *, k_values=K_VALUES)

Score one retriever over the golden entries at every depth.

Each query retrieves once at max(k_values); shallower depths are prefix cuts of that single ranked list (retrieval order is deterministic given the stores' contents).

Parameters:

Name	Type	Description	Default
retriever	Retriever	The retrieval method under measurement.	required
entries	Sequence[ResolvedGoldenEntry]	Resolved golden entries.	required
k_values	Sequence[int]	Depths to report.	K_VALUES

Returns:

Type	Description
dict[str, dict[str, float]]	A (summary, golden_ranks) pair. summary holds the
list[dict[str, int | None]]	{"recall": {"5": …}, "pass": {"5": …}} averages over queries;
tuple[dict[str, dict[str, float]], list[dict[str, int | None]]]	golden_ranks holds, per entry in order, each golden
tuple[dict[str, dict[str, float]], list[dict[str, int | None]]]	chunk_id's 1-based position in the ranked list (None if
tuple[dict[str, dict[str, float]], list[dict[str, int | None]]]	absent) — the raw material for debugging a bad score.

Raises:

Type	Description
ValueError	If entries or k_values is empty.

Source code in varagity/eval/evaluate.py

def measure_retriever(
    retriever: Retriever,
    entries: Sequence[ResolvedGoldenEntry],
    *,
    k_values: Sequence[int] = K_VALUES,
) -> tuple[dict[str, dict[str, float]], list[dict[str, int | None]]]:
    """Score one retriever over the golden entries at every depth.

    Each query retrieves once at ``max(k_values)``; shallower depths are
    prefix cuts of that single ranked list (retrieval order is
    deterministic given the stores' contents).

    Args:
        retriever: The retrieval method under measurement.
        entries: Resolved golden entries.
        k_values: Depths to report.

    Returns:
        A ``(summary, golden_ranks)`` pair. ``summary`` holds the
        ``{"recall": {"5": …}, "pass": {"5": …}}`` averages over queries;
        ``golden_ranks`` holds, per entry in order, each golden
        ``chunk_id``'s 1-based position in the ranked list (``None`` if
        absent) — the raw material for debugging a bad score.

    Raises:
        ValueError: If ``entries`` or ``k_values`` is empty.
    """
    if not entries:
        raise ValueError("measure_retriever needs at least one golden entry")
    if not k_values:
        raise ValueError("measure_retriever needs at least one k value")
    max_k = max(k_values)
    recall_sums = dict.fromkeys(k_values, 0.0)
    pass_sums = dict.fromkeys(k_values, 0.0)
    golden_ranks: list[dict[str, int | None]] = []
    for entry in entries:
        chunks: list[RetrievedChunk] = retriever.retrieve(entry.query, k=max_k, verbose=0)
        retrieved_ids = [chunk.chunk_id for chunk in chunks]
        for k in k_values:
            recall_sums[k] += recall_at_k(entry.chunk_ids, retrieved_ids, k)
            pass_sums[k] += pass_at_k(entry.chunk_ids, retrieved_ids, k)
        positions = {chunk_id: rank for rank, chunk_id in enumerate(retrieved_ids, start=1)}
        golden_ranks.append({chunk_id: positions.get(chunk_id) for chunk_id in entry.chunk_ids})
    summary = {
        "recall": {str(k): recall_sums[k] / len(entries) for k in k_values},
        "pass": {str(k): pass_sums[k] / len(entries) for k in k_values},
    }
    return summary, golden_ranks

measure_retriever_facts(retriever, entries, *, k_values=K_VALUES)

Score one retriever over fact-resolved entries at every depth.

The sweep counterpart of :func:measure_retriever — identical shape, with each ref satisfied by any of its acceptable chunk ids.

Parameters:

Name	Type	Description	Default
retriever	Retriever	The retrieval method under measurement.	required
entries	Sequence[FactResolvedEntry]	Fact-resolved golden entries.	required
k_values	Sequence[int]	Depths to report.	K_VALUES

Returns:

Type	Description
dict[str, dict[str, float]]	A (summary, golden_ranks) pair; golden_ranks holds, per
list[dict[str, int | None]]	entry in order, each ref label's best 1-based rank among its
tuple[dict[str, dict[str, float]], list[dict[str, int | None]]]	acceptable ids (None if none was retrieved).

Raises:

Type	Description
ValueError	If entries or k_values is empty.

Source code in varagity/eval/evaluate.py

def measure_retriever_facts(
    retriever: Retriever,
    entries: Sequence[FactResolvedEntry],
    *,
    k_values: Sequence[int] = K_VALUES,
) -> tuple[dict[str, dict[str, float]], list[dict[str, int | None]]]:
    """Score one retriever over fact-resolved entries at every depth.

    The sweep counterpart of :func:`measure_retriever` — identical shape,
    with each ref satisfied by *any* of its acceptable chunk ids.

    Args:
        retriever: The retrieval method under measurement.
        entries: Fact-resolved golden entries.
        k_values: Depths to report.

    Returns:
        A ``(summary, golden_ranks)`` pair; ``golden_ranks`` holds, per
        entry in order, each ref label's best 1-based rank among its
        acceptable ids (``None`` if none was retrieved).

    Raises:
        ValueError: If ``entries`` or ``k_values`` is empty.
    """
    if not entries:
        raise ValueError("measure_retriever_facts needs at least one golden entry")
    if not k_values:
        raise ValueError("measure_retriever_facts needs at least one k value")
    max_k = max(k_values)
    recall_sums = dict.fromkeys(k_values, 0.0)
    pass_sums = dict.fromkeys(k_values, 0.0)
    golden_ranks: list[dict[str, int | None]] = []
    for entry in entries:
        chunks: list[RetrievedChunk] = retriever.retrieve(entry.query, k=max_k, verbose=0)
        retrieved_ids = [chunk.chunk_id for chunk in chunks]
        acceptable = [ref.chunk_ids for ref in entry.refs]
        for k in k_values:
            recall_sums[k] += recall_at_k_any(acceptable, retrieved_ids, k)
            pass_sums[k] += pass_at_k_any(acceptable, retrieved_ids, k)
        positions = {chunk_id: rank for rank, chunk_id in enumerate(retrieved_ids, start=1)}
        golden_ranks.append(
            {
                ref.label: min(
                    (positions[chunk_id] for chunk_id in ref.chunk_ids if chunk_id in positions),
                    default=None,
                )
                for ref in entry.refs
            }
        )
    summary = {
        "recall": {str(k): recall_sums[k] / len(entries) for k in k_values},
        "pass": {str(k): pass_sums[k] / len(entries) for k in k_values},
    }
    return summary, golden_ranks

pass_at_k(golden, retrieved, k)

Whether all golden chunk ids are in the top-k (0.0 or 1.0).

The strict complement of :func:recall_at_k: a query passes only if nothing a perfect retriever would return is missing. Callers average over queries into a pass rate.

Parameters:

Name	Type	Description	Default
golden	Collection[str]	The query's golden chunk ids (non-empty).	required
retrieved	Sequence[str]	Retrieved chunk ids, best first.	required
k	int	Retrieval depth to evaluate at.	required

Returns:

Type	Description
float	1.0 if every golden id is in retrieved[:k], else 0.0.

Raises:

Type	Description
ValueError	If golden is empty or k is not positive.

Source code in varagity/eval/evaluate.py

def pass_at_k(golden: Collection[str], retrieved: Sequence[str], k: int) -> float:
    """Whether **all** golden chunk ids are in the top-k (0.0 or 1.0).

    The strict complement of :func:`recall_at_k`: a query passes only if
    nothing a perfect retriever would return is missing. Callers average
    over queries into a pass rate.

    Args:
        golden: The query's golden chunk ids (non-empty).
        retrieved: Retrieved chunk ids, best first.
        k: Retrieval depth to evaluate at.

    Returns:
        ``1.0`` if every golden id is in ``retrieved[:k]``, else ``0.0``.

    Raises:
        ValueError: If ``golden`` is empty or ``k`` is not positive.
    """
    return 1.0 if recall_at_k(golden, retrieved, k) == 1.0 else 0.0

pass_at_k_any(acceptable, retrieved, k)

Whether every golden ref is satisfied in the top-k (0.0 or 1.0).

Parameters:

Name	Type	Description	Default
acceptable	Sequence[Collection[str]]	Per golden ref, the chunk ids that satisfy it.	required
retrieved	Sequence[str]	Retrieved chunk ids, best first.	required
k	int	Retrieval depth to evaluate at.	required

Returns:

Type	Description
float	1.0 if every ref has an acceptable id in retrieved[:k],
float	else 0.0.

Raises:

Type	Description
ValueError	If acceptable is empty or k is not positive.

Source code in varagity/eval/evaluate.py

def pass_at_k_any(acceptable: Sequence[Collection[str]], retrieved: Sequence[str], k: int) -> float:
    """Whether **every** golden ref is satisfied in the top-k (0.0 or 1.0).

    Args:
        acceptable: Per golden ref, the chunk ids that satisfy it.
        retrieved: Retrieved chunk ids, best first.
        k: Retrieval depth to evaluate at.

    Returns:
        ``1.0`` if every ref has an acceptable id in ``retrieved[:k]``,
        else ``0.0``.

    Raises:
        ValueError: If ``acceptable`` is empty or ``k`` is not positive.
    """
    return 1.0 if recall_at_k_any(acceptable, retrieved, k) == 1.0 else 0.0

pinned_eval_settings(**overrides)

Apply :data:PINNED_EVAL_SETTINGS (plus overrides) for a block.

Exports the pins as environment variables and clears the settings cache, restoring both on exit — the eval-harness counterpart of the test suite's settings_env fixture (see the module docstring for why an environment override rather than parameter injection).

Parameters:

Name	Type	Description	Default
**overrides	str	Additional setting pins for this block (e.g. CONTEXTUALIZE="true", OCR_ENGINE="tesseract"); they take precedence over the standing pins.	{}

Yields:

Type	Description
None	Nothing; the pinned settings are active inside the block.

Source code in varagity/eval/evaluate.py

@contextmanager
def pinned_eval_settings(**overrides: str) -> Iterator[None]:
    """Apply :data:`PINNED_EVAL_SETTINGS` (plus overrides) for a block.

    Exports the pins as environment variables and clears the settings
    cache, restoring both on exit — the eval-harness counterpart of the
    test suite's ``settings_env`` fixture (see the module docstring for
    why an environment override rather than parameter injection).

    Args:
        **overrides: Additional setting pins for this block (e.g.
            ``CONTEXTUALIZE="true"``, ``OCR_ENGINE="tesseract"``); they
            take precedence over the standing pins.

    Yields:
        Nothing; the pinned settings are active inside the block.
    """
    pins = {**PINNED_EVAL_SETTINGS, **overrides}
    saved = {name: os.environ.get(name) for name in pins}
    os.environ.update(pins)
    get_settings.cache_clear()
    try:
        yield
    finally:
        for name, value in saved.items():
            if value is None:
                os.environ.pop(name, None)
            else:
                os.environ[name] = value
        get_settings.cache_clear()

recall_at_k(golden, retrieved, k)

Fraction of golden chunk ids present in the top-k retrieved.

The Anthropic cookbook's evaluation metric (there reported as Pass@n): one query's score; callers average over queries.

Parameters:

Name	Type	Description	Default
golden	Collection[str]	The query's golden chunk ids (non-empty).	required
retrieved	Sequence[str]	Retrieved chunk ids, best first.	required
k	int	Retrieval depth to evaluate at.	required

Returns:

Type	Description
float	|golden ∩ retrieved[:k]| / |golden|.

Raises:

Type	Description
ValueError	If golden is empty or k is not positive.

Source code in varagity/eval/evaluate.py

def recall_at_k(golden: Collection[str], retrieved: Sequence[str], k: int) -> float:
    """Fraction of golden chunk ids present in the top-k retrieved.

    The Anthropic cookbook's evaluation metric (there reported as
    *Pass@n*): one query's score; callers average over queries.

    Args:
        golden: The query's golden chunk ids (non-empty).
        retrieved: Retrieved chunk ids, best first.
        k: Retrieval depth to evaluate at.

    Returns:
        ``|golden ∩ retrieved[:k]| / |golden|``.

    Raises:
        ValueError: If ``golden`` is empty or ``k`` is not positive.
    """
    if not golden:
        raise ValueError("recall_at_k needs at least one golden chunk id")
    if k <= 0:
        raise ValueError(f"k must be positive; got {k}")
    top = set(retrieved[:k])
    return sum(1 for chunk_id in golden if chunk_id in top) / len(golden)

recall_at_k_any(acceptable, retrieved, k)

Fraction of golden refs with any acceptable chunk in the top-k.

:func:recall_at_k generalized for the chunker sweep: a ref is satisfied by whichever strategy-true chunk carries its fact. An empty acceptable set (the fact matched no chunk) can never be satisfied.

Parameters:

Name	Type	Description	Default
acceptable	Sequence[Collection[str]]	Per golden ref, the chunk ids that satisfy it.	required
retrieved	Sequence[str]	Retrieved chunk ids, best first.	required
k	int	Retrieval depth to evaluate at.	required

Returns:

Type	Description
float	|satisfied refs| / |refs|.

Raises:

Type	Description
ValueError	If acceptable is empty or k is not positive.

Source code in varagity/eval/evaluate.py

def recall_at_k_any(
    acceptable: Sequence[Collection[str]], retrieved: Sequence[str], k: int
) -> float:
    """Fraction of golden refs with **any** acceptable chunk in the top-k.

    :func:`recall_at_k` generalized for the chunker sweep: a ref is
    satisfied by whichever strategy-true chunk carries its fact. An empty
    acceptable set (the fact matched no chunk) can never be satisfied.

    Args:
        acceptable: Per golden ref, the chunk ids that satisfy it.
        retrieved: Retrieved chunk ids, best first.
        k: Retrieval depth to evaluate at.

    Returns:
        ``|satisfied refs| / |refs|``.

    Raises:
        ValueError: If ``acceptable`` is empty or ``k`` is not positive.
    """
    if not acceptable:
        raise ValueError("recall_at_k_any needs at least one golden ref")
    if k <= 0:
        raise ValueError(f"k must be positive; got {k}")
    top = set(retrieved[:k])
    return sum(1 for ids in acceptable if top.intersection(ids)) / len(acceptable)

resolve_golden_by_fact(entries, store)

Re-resolve golden refs by content against the ingested boundaries.

For each ref with a fact, scans its document's stored chunks (case-insensitively — OCR extraction may case-shift) and collects every chunk containing the fact. A ref without a fact falls back to its index-anchored chunk_id (only honest under the pinned default boundaries — the golden set carries facts precisely so the sweep never relies on this). A fact that matches no chunk is logged and left with an empty acceptable set: a guaranteed miss, mirroring the OCR benchmark's drift posture rather than silently inflating scores.

Parameters:

Name	Type	Description	Default
entries	Sequence[ResolvedGoldenEntry]	The index-resolved golden entries (their relevant refs carry the facts; their chunk_ids carry the doc ids).	required
store	ContextualVectorDB	The vector store holding the current strategy's ingest.	required

Returns:

Type	Description
list[FactResolvedEntry]	One fact-resolved entry per input entry, in order.

Source code in varagity/eval/evaluate.py

def resolve_golden_by_fact(
    entries: Sequence[ResolvedGoldenEntry], store: ContextualVectorDB
) -> list[FactResolvedEntry]:
    """Re-resolve golden refs by content against the ingested boundaries.

    For each ref with a ``fact``, scans its document's stored chunks
    (case-insensitively — OCR extraction may case-shift) and collects every
    chunk containing the fact. A ref without a fact falls back to its
    index-anchored ``chunk_id`` (only honest under the pinned default
    boundaries — the golden set carries facts precisely so the sweep never
    relies on this). A fact that matches no chunk is logged and left with
    an empty acceptable set: a guaranteed miss, mirroring the OCR
    benchmark's drift posture rather than silently inflating scores.

    Args:
        entries: The index-resolved golden entries (their ``relevant`` refs
            carry the facts; their ``chunk_ids`` carry the doc ids).
        store: The vector store holding the current strategy's ingest.

    Returns:
        One fact-resolved entry per input entry, in order.
    """
    chunks_by_doc: dict[str, list[RetrievedChunk]] = {}
    resolved: list[FactResolvedEntry] = []
    for entry in entries:
        refs: list[FactRef] = []
        for ref, chunk_id in zip(entry.relevant, entry.chunk_ids, strict=True):
            if ref.fact is None:
                refs.append(FactRef(label=chunk_id, chunk_ids=[chunk_id]))
                continue
            doc_id = chunk_id.split("::", 1)[0]
            if doc_id not in chunks_by_doc:
                chunks_by_doc[doc_id] = store.document_chunks(doc_id)
            needle = ref.fact.lower()
            matches = [
                chunk.chunk_id for chunk in chunks_by_doc[doc_id] if needle in chunk.content.lower()
            ]
            if not matches:
                logger.warning(
                    "golden fact %r not found in any ingested chunk of %s "
                    "(%d chunks) — counted as a guaranteed miss",
                    ref.fact,
                    ref.rel_source,
                    len(chunks_by_doc[doc_id]),
                )
            refs.append(FactRef(label=ref.fact, chunk_ids=matches))
        resolved.append(FactResolvedEntry(query=entry.query, refs=refs))
    return resolved

run_chat_eval(*, corpus_root=EVAL_CORPUS, conversations_dir=CONVERSATIONS_DIR, results_dir=RESULTS_DIR, ingest=ingest_corpus, verbose=None)

Run the multi-turn chat-engine eval; persist results (spec_v3 §4.9).

One contextual ingest into ephemeral stores backs the whole run; every registered chat engine (enumerated from the registry, like the chunker sweep) then plays each conversation fixture, and every turn is scored fact-anchored under :data:CHAT_EVAL_RETRIEVAL_CONFIGS. The condense settings are pinned (:data:PINNED_CHAT_SETTINGS) so a host env with the kill switch off can't silently measure condense_context as simple. This harness decides the shipped CHAT_ENGINE default — and "the default stays simple" is a valid result, not a failure.

Parameters:

Name	Type	Description	Default
corpus_root	Path	The eval corpus (defaults to the fixtures corpus the conversation refs are authored over).	EVAL_CORPUS
conversations_dir	Path	The conversation fixtures directory.	CONVERSATIONS_DIR
results_dir	Path	Directory for the timestamped results JSON.	RESULTS_DIR
ingest	IngestCallable	Ingest callable (ingest_corpus, or the Prefect ingest_flow when invoked through the eval flow).	ingest_corpus
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (also written to results_dir), with a
dict[str, Any]	"results_path" key naming the file.

Raises:

Type	Description
ValueError	If verbose is invalid or a fixture fails validation.
FileNotFoundError	If the fixtures directory or a referenced corpus file is missing.
RuntimeError	If the eval ingest reports failed files (a partial index would silently under-report every score).

Source code in varagity/eval/evaluate.py

def run_chat_eval(
    *,
    corpus_root: Path = EVAL_CORPUS,
    conversations_dir: Path = CONVERSATIONS_DIR,
    results_dir: Path = RESULTS_DIR,
    ingest: IngestCallable = ingest_corpus,
    verbose: int | None = None,
) -> dict[str, Any]:
    """Run the multi-turn chat-engine eval; persist results (spec_v3 §4.9).

    One contextual ingest into ephemeral stores backs the whole run; every
    registered chat engine (enumerated from the registry, like the chunker
    sweep) then plays each conversation fixture, and every turn is scored
    fact-anchored under :data:`CHAT_EVAL_RETRIEVAL_CONFIGS`. The condense
    settings are pinned (:data:`PINNED_CHAT_SETTINGS`) so a host env with
    the kill switch off can't silently measure ``condense_context`` as
    ``simple``. This harness decides the shipped ``CHAT_ENGINE`` default —
    and "the default stays ``simple``" is a valid result, not a failure.

    Args:
        corpus_root: The eval corpus (defaults to the fixtures corpus the
            conversation refs are authored over).
        conversations_dir: The conversation fixtures directory.
        results_dir: Directory for the timestamped results JSON.
        ingest: Ingest callable (``ingest_corpus``, or the Prefect
            ``ingest_flow`` when invoked through the eval flow).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (also written to ``results_dir``), with a
        ``"results_path"`` key naming the file.

    Raises:
        ValueError: If ``verbose`` is invalid or a fixture fails
            validation.
        FileNotFoundError: If the fixtures directory or a referenced
            corpus file is missing.
        RuntimeError: If the eval ingest reports failed files (a partial
            index would silently under-report every score).
    """
    verbose = check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    conversations = load_conversations(conversations_dir)
    flat_golden = [
        GoldenEntry(query=turn.query, relevant=turn.relevant)
        for conversation in conversations
        for turn in conversation.turns
    ]
    resolved = resolve_golden(flat_golden, corpus_root)

    # Resolve the live model clients up front: a misconfigured endpoint
    # should fail before containers spin, not after an ingest.
    embeddings = get_model("embedding")
    llm = get_model("default")
    rerank = get_model("rerank")

    with ephemeral_stores() as stores:
        hybrid = HybridRetriever(store=stores.store, bm25=stores.bm25, embeddings=embeddings)
        # Keyed to match CHAT_EVAL_RETRIEVAL_CONFIGS (defined together above).
        retrievers: dict[str, Retriever] = {
            "hybrid": hybrid,
            "reranked": RerankedRetriever(base=hybrid, rerank=rerank),
        }
        with pinned_eval_settings(CONTEXTUALIZE="true", **PINNED_CHAT_SETTINGS):
            logger.info("chat eval: contextual ingest into ephemeral stores")
            started = time.monotonic()
            summary = ingest(
                str(corpus_root),
                store=stores.store,
                bm25=stores.bm25,
                embeddings=embeddings,
                llm=llm,
                verbose=verbose,
            )
            ingest_seconds = round(time.monotonic() - started, 2)
            if summary.failed:
                raise RuntimeError(f"chat-eval ingest failed for {summary.failed} file(s)")
            fact_entries = resolve_golden_by_fact(resolved, stores.store)
            unresolved = [
                ref.label for entry in fact_entries for ref in entry.refs if not ref.chunk_ids
            ]
            engines = {
                name: measure_chat_engine(
                    CHAT_ENGINE_REGISTRY[name],
                    conversations,
                    fact_entries,
                    retrievers,
                    llm=llm,
                    k_values=CHAT_K_VALUES,
                )
                for name in sorted(CHAT_ENGINE_REGISTRY)
            }

    settings = get_settings()
    results: dict[str, Any] = {
        "kind": "chat_eval",
        "timestamp": datetime.now(UTC).isoformat(),
        "corpus": str(corpus_root),
        "conversations_dir": str(conversations_dir),
        "n_conversations": len(conversations),
        "n_turns": len(flat_golden),
        "n_follow_up_turns": sum(len(c.turns) - 1 for c in conversations),
        "k_values": list(CHAT_K_VALUES),
        "retrieval_configs": list(CHAT_EVAL_RETRIEVAL_CONFIGS),
        "embedding_model": settings.EMBEDDING_MODEL,
        "rerank_model": settings.RERANK_MODEL,
        "pinned_settings": {
            **PINNED_EVAL_SETTINGS,
            "CONTEXTUALIZE": "true",
            **PINNED_CHAT_SETTINGS,
        },
        "chunks_ingested": summary.chunks,
        "ingest_seconds": ingest_seconds,
        "unresolved_facts": unresolved,
        "engines": engines,
    }
    results["results_path"] = str(write_results("chat", results, results_dir=results_dir))
    return results

run_matrix(*, corpus_root=EVAL_CORPUS, golden_path=GOLDEN_PATH, results_dir=RESULTS_DIR, ingest=ingest_corpus, verbose=None)

Run the 7-config retrieval matrix + the chunker sweep; persist results.

Two ingests into ephemeral stores cover all seven configs (see the module docstring); the chunker sweep then reingests once per remaining registered strategy (recursive_character reuses ingest B) and measures each across :data:SWEEP_RETRIEVAL_CONFIGS with fact-anchored golden resolution — deliberately without the HyDE configs: the hypothetical passage depends only on the query, never on chunk boundaries, so a per-strategy re-measure would re-pay every LLM generation to measure the same transformation. Embeddings, the contextualizing LLM (which also writes the HyDE passages), and the reranker resolve from settings — the live GPU services (the semantic strategy additionally embeds at ingest time through the same service).

Parameters:

Name	Type	Description	Default
corpus_root	Path	The eval corpus (defaults to the fixtures corpus the golden set is authored over).	EVAL_CORPUS
golden_path	Path	The golden dataset file.	GOLDEN_PATH
results_dir	Path	Directory for the timestamped results JSON.	RESULTS_DIR
ingest	IngestCallable	Ingest callable (ingest_corpus, or the Prefect ingest_flow when invoked through the eval flow so both ingests are tracked subflows).	ingest_corpus
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (also written to results_dir), with a
dict[str, Any]	"results_path" key naming the file.

Raises:

Type	Description
ValueError	If verbose is invalid, or the golden set doesn't match the ingested chunk boundaries.
FileNotFoundError	If the golden set or a referenced corpus file is missing.
RuntimeError	If either eval ingest reports failed files (a partial index would silently under-report every score).

Source code in varagity/eval/evaluate.py

def run_matrix(
    *,
    corpus_root: Path = EVAL_CORPUS,
    golden_path: Path = GOLDEN_PATH,
    results_dir: Path = RESULTS_DIR,
    ingest: IngestCallable = ingest_corpus,
    verbose: int | None = None,
) -> dict[str, Any]:
    """Run the 7-config retrieval matrix + the chunker sweep; persist results.

    Two ingests into ephemeral stores cover all seven configs (see the
    module docstring); the chunker sweep then reingests once per remaining
    registered strategy (``recursive_character`` reuses ingest B) and
    measures each across :data:`SWEEP_RETRIEVAL_CONFIGS` with fact-anchored
    golden resolution — deliberately *without* the HyDE configs: the
    hypothetical passage depends only on the query, never on chunk
    boundaries, so a per-strategy re-measure would re-pay every LLM
    generation to measure the same transformation. Embeddings, the
    contextualizing LLM (which also writes the HyDE passages), and the
    reranker resolve from settings — the live GPU services (the ``semantic``
    strategy additionally embeds at ingest time through the same service).

    Args:
        corpus_root: The eval corpus (defaults to the fixtures corpus the
            golden set is authored over).
        golden_path: The golden dataset file.
        results_dir: Directory for the timestamped results JSON.
        ingest: Ingest callable (``ingest_corpus``, or the Prefect
            ``ingest_flow`` when invoked through the eval flow so both
            ingests are tracked subflows).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (also written to ``results_dir``), with a
        ``"results_path"`` key naming the file.

    Raises:
        ValueError: If ``verbose`` is invalid, or the golden set doesn't
            match the ingested chunk boundaries.
        FileNotFoundError: If the golden set or a referenced corpus file
            is missing.
        RuntimeError: If either eval ingest reports failed files (a partial
            index would silently under-report every score).
    """
    verbose = check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    entries = resolve_golden(load_golden(golden_path), corpus_root)

    # Resolve the live model clients up front: a misconfigured endpoint
    # should fail before containers spin, not after an ingest.
    embeddings = get_model("embedding")
    llm = get_model("default")
    rerank = get_model("rerank")

    configs: dict[str, dict[str, dict[str, float]]] = {}
    ranks_by_config: dict[str, list[dict[str, int | None]]] = {}
    ingest_seconds: dict[str, float] = {}

    with ephemeral_stores() as stores:
        hybrid = HybridRetriever(store=stores.store, bm25=stores.bm25, embeddings=embeddings)
        hyde = HydeRetriever(base=hybrid, llm=llm, embeddings=embeddings)
        retrievers: dict[str, Retriever] = {
            "semantic": SemanticRetriever(store=stores.store, embeddings=embeddings),
            "bm25": BM25Retriever(bm25=stores.bm25, store=stores.store),
            "hybrid": hybrid,
            # Config 5 composes the same store-wired hybrid; the reranker
            # is the live infinity container's /rerank (spec_v2 §5.5).
            "reranked": RerankedRetriever(base=hybrid, rerank=rerank),
            # Configs 6–7 (ADR-016): hyde steers the same hybrid's dense
            # arm with a live-LLM hypothetical passage; the pairing stacks
            # rerank OUTSIDE hyde, so the cross-encoder judges the real
            # query while HyDE shapes the candidate pool.
            "hyde": hyde,
            "hyde_reranked": RerankedRetriever(base=hyde, rerank=rerank),
        }

        # Ingest A — non-contextual baseline → config 1.
        with pinned_eval_settings(CONTEXTUALIZE="false"):
            logger.info("ingest A (non-contextual baseline) into ephemeral stores")
            started = time.monotonic()
            summary_a = ingest(
                str(corpus_root),
                store=stores.store,
                bm25=stores.bm25,
                embeddings=embeddings,
                verbose=verbose,
            )
            ingest_seconds["noncontextual"] = round(time.monotonic() - started, 2)
            if summary_a.failed:
                raise RuntimeError(f"ingest A failed for {summary_a.failed} file(s)")
            validate_golden_against_store(entries, stores, strict=True)
            configs["semantic_noncontextual"], ranks_by_config["semantic_noncontextual"] = (
                measure_retriever(retrievers["semantic"], entries)
            )

        # Ingest B — contextual (LLM blurbs) → configs 2–7 on one index.
        with pinned_eval_settings(CONTEXTUALIZE="true"):
            logger.info("ingest B (contextual) into the same ephemeral stores (reingest)")
            started = time.monotonic()
            summary_b = ingest(
                str(corpus_root),
                store=stores.store,
                bm25=stores.bm25,
                embeddings=embeddings,
                llm=llm,
                reingest=True,
                verbose=verbose,
            )
            ingest_seconds["contextual"] = round(time.monotonic() - started, 2)
            if summary_b.failed:
                raise RuntimeError(f"ingest B failed for {summary_b.failed} file(s)")
            for config, retriever_name in (
                ("semantic_contextual", "semantic"),
                ("bm25_contextual", "bm25"),
                ("hybrid_contextual", "hybrid"),
                ("hybrid_rerank_contextual", "reranked"),
                ("hyde_contextual", "hyde"),
                ("hyde_rerank_contextual", "hyde_reranked"),
            ):
                configs[config], ranks_by_config[config] = measure_retriever(
                    retrievers[retriever_name], entries
                )

        # ── Chunker sweep (spec_v2 §7.4): one contextual ingest per
        # strategy, measured across the contextual retrieval configs with
        # fact-anchored golden resolution. recursive_character goes first,
        # reusing ingest B's index before any reingest replaces it.
        sweep: dict[str, dict[str, Any]] = {}
        strategies = ["recursive_character"] + sorted(
            name for name in CHUNKER_REGISTRY if name != "recursive_character"
        )
        for strategy in strategies:
            with pinned_eval_settings(CONTEXTUALIZE="true", CHUNKING_STRATEGY=strategy):
                if strategy == "recursive_character":
                    chunks_ingested, seconds = summary_b.chunks, ingest_seconds["contextual"]
                else:
                    logger.info("chunker sweep: reingesting with %r", strategy)
                    started = time.monotonic()
                    summary = ingest(
                        str(corpus_root),
                        store=stores.store,
                        bm25=stores.bm25,
                        embeddings=embeddings,
                        llm=llm,
                        reingest=True,
                        verbose=verbose,
                    )
                    seconds = round(time.monotonic() - started, 2)
                    if summary.failed:
                        raise RuntimeError(
                            f"chunker-sweep ingest ({strategy}) failed for {summary.failed} file(s)"
                        )
                    chunks_ingested = summary.chunks
                fact_entries = resolve_golden_by_fact(entries, stores.store)
                unresolved = [
                    ref.label for entry in fact_entries for ref in entry.refs if not ref.chunk_ids
                ]
                methods: dict[str, dict[str, dict[str, float]]] = {}
                method_ranks: dict[str, list[dict[str, int | None]]] = {}
                for name in SWEEP_RETRIEVAL_CONFIGS:
                    methods[name], method_ranks[name] = measure_retriever_facts(
                        retrievers[name], fact_entries
                    )
                sweep[strategy] = {
                    "chunks": chunks_ingested,
                    "ingest_seconds": seconds,
                    "unresolved_facts": unresolved,
                    "configs": methods,
                    "golden_ranks": method_ranks,
                }

    settings = get_settings()
    results: dict[str, Any] = {
        "kind": "retrieval_matrix",
        "timestamp": datetime.now(UTC).isoformat(),
        "corpus": str(corpus_root),
        "golden_path": str(golden_path),
        "n_queries": len(entries),
        "k_values": list(K_VALUES),
        "embedding_model": settings.EMBEDDING_MODEL,
        "rerank_model": settings.RERANK_MODEL,
        "pinned_settings": dict(PINNED_EVAL_SETTINGS),
        "chunks_ingested": summary_b.chunks,
        "ingest_seconds": ingest_seconds,
        "configs": configs,
        "sweep_retrieval_configs": list(SWEEP_RETRIEVAL_CONFIGS),
        "chunker_sweep": sweep,
        "per_query": [
            {
                "query": entry.query,
                "golden": entry.chunk_ids,
                "golden_ranks": {
                    config: ranks_by_config[config][index] for config in ranks_by_config
                },
            }
            for index, entry in enumerate(entries)
        ],
    }
    results["results_path"] = str(write_results("matrix", results, results_dir=results_dir))
    return results

validate_golden_against_store(entries, stores, *, strict=True)

Check that every golden ref exists in the freshly ingested store.

A golden chunk_index beyond a document's actual chunk count means the golden set has drifted from the chunking pipeline (or, in the OCR benchmark, that an engine's text produced different chunk boundaries) — a score computed against it would silently under-report.

Parameters:

Name	Type	Description	Default
entries	Sequence[ResolvedGoldenEntry]	Resolved golden entries.	required
stores	EphemeralStores	The ingested ephemeral stores.	required
strict	bool	Raise on unresolvable refs (the retrieval matrix) instead of warning and returning them (the OCR benchmark, where per-engine boundary drift is a known, reported caveat).	True

Returns:

Type	Description
list[str]	The unresolvable chunk_ids (always empty when strict).

Raises:

Type	Description
ValueError	If strict and any ref is unresolvable.

Source code in varagity/eval/evaluate.py

def validate_golden_against_store(
    entries: Sequence[ResolvedGoldenEntry],
    stores: EphemeralStores,
    *,
    strict: bool = True,
) -> list[str]:
    """Check that every golden ref exists in the freshly ingested store.

    A golden ``chunk_index`` beyond a document's actual chunk count means
    the golden set has drifted from the chunking pipeline (or, in the OCR
    benchmark, that an engine's text produced different chunk boundaries)
    — a score computed against it would silently under-report.

    Args:
        entries: Resolved golden entries.
        stores: The ingested ephemeral stores.
        strict: Raise on unresolvable refs (the retrieval matrix) instead
            of warning and returning them (the OCR benchmark, where
            per-engine boundary drift is a known, reported caveat).

    Returns:
        The unresolvable ``chunk_id``s (always empty when ``strict``).

    Raises:
        ValueError: If ``strict`` and any ref is unresolvable.
    """
    n_chunks: dict[str, int | None] = {}
    unresolvable: list[str] = []
    for entry in entries:
        for ref, chunk_id in zip(entry.relevant, entry.chunk_ids, strict=True):
            doc_id = chunk_id.split("::", 1)[0]
            if doc_id not in n_chunks:
                n_chunks[doc_id] = stores.store.document_n_chunks(doc_id)
            count = n_chunks[doc_id]
            if count is None or ref.chunk_index >= count:
                unresolvable.append(chunk_id)
    if unresolvable:
        message = (
            f"{len(unresolvable)} golden ref(s) don't exist in the ingested store "
            f"(chunk boundaries drifted?): {unresolvable}"
        )
        if strict:
            raise ValueError(message)
        logger.warning("%s — counted as guaranteed misses", message)
    return unresolvable

write_results(kind, payload, *, results_dir=RESULTS_DIR)

Persist one evaluation run's results as timestamped JSON (spec §16).

Parameters:

Name	Type	Description	Default
kind	str	Short result-file tag ("matrix" or "ocr").	required
payload	dict[str, Any]	The full results document.	required
results_dir	Path	Directory for result files (created if missing).	RESULTS_DIR

Returns:

Type	Description
Path	The path written.

Source code in varagity/eval/evaluate.py

def write_results(kind: str, payload: dict[str, Any], *, results_dir: Path = RESULTS_DIR) -> Path:
    """Persist one evaluation run's results as timestamped JSON (spec §16).

    Args:
        kind: Short result-file tag (``"matrix"`` or ``"ocr"``).
        payload: The full results document.
        results_dir: Directory for result files (created if missing).

    Returns:
        The path written.
    """
    results_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    path = results_dir / f"{stamp}-{kind}.json"
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    logger.info("wrote %s results to %s", kind, path)
    return path

varagity.eval.graph_fixtures

Synthetic iMessage fixture corpus for the graph eval (spec_graphrag §12).

Writes a real chat.db — the schema subset :mod:varagity.graph.sources.imessage reads — so the bake-off and the permanent regression harness parse the fixture through the product code path (find_message_source → batch_for_path), never a test shortcut.

Two profiles (plan decision #11):

• smoke — the hand-scripted conversations only (~226 messages, 10 threads, 5 named participants + the owner). Adapter development, harness unit tests, the incremental-reindex check.
• full — the same scripted messages verbatim, plus deterministic filler up to > 10,000 messages spanning a decade (the §12 Q2 bake-off corpus).

Golden QA (data/eval/graph_golden_qa.jsonl) is authored strictly against scripted messages, which are byte-identical in both profiles, so a golden never depends on filler. Filler is held to the opposite discipline: it must never mention a golden entity or fact term (:data:FILLER_BLOCKLIST, asserted at generation time) — that is what keeps the corpus discriminative at 10⁴ scale rather than merely large.

The scripted content is synthetic: invented people, invented events, no real data (spec_graphrag §6). The one exception is the attributedBody blob reused from the Phase-1 decode fixture (innocuous, owner-reviewed), so that the NULL-text decode path runs inside the eval too.

FixtureManifest

Bases: BaseModel

What one :func:build_fixture_chat_db call wrote (never persisted).

Returned for assertions and for the harness's settings pins: the parser maps handles to names through GRAPH_HANDLE_NAMES and the owner through GRAPH_OWNER_ALIASES, so a caller that wants the scripted display names ("Bob Nakamura", not "+12145550101") pins :attr:handle_names and :attr:owner_label before parsing.

Attributes:

Name	Type	Description
profile	str	The profile built (smoke | full).
seed	int	The filler RNG seed (irrelevant for smoke, recorded anyway).
message_count	int	Parsed messages written (scripted + filler); tapback rows are reactions, not messages, and are excluded.
scripted_count	int	Messages from the hand-authored script.
filler_count	int	Generated filler messages (0 for smoke).
tapback_count	int	Reaction rows written (they fold onto their targets).
thread_counts	dict[str, int]	Message count per thread guid.
guids	list[str]	Every message guid written, in write order — the set the golden QA's required_guids is validated against.
first_timestamp	datetime	Earliest message time (aware UTC).
last_timestamp	datetime	Latest message time (aware UTC).
owner_label	str	Display name the script was authored with for is_from_me messages.
handle_names	dict[str, str]	handle → display name for every participant (scripted and filler).

Source code in varagity/eval/graph_fixtures.py

class FixtureManifest(BaseModel):
    """What one :func:`build_fixture_chat_db` call wrote (never persisted).

    Returned for assertions and for the harness's settings pins: the parser
    maps handles to names through ``GRAPH_HANDLE_NAMES`` and the owner
    through ``GRAPH_OWNER_ALIASES``, so a caller that wants the scripted
    display names ("Bob Nakamura", not "+12145550101") pins
    :attr:`handle_names` and :attr:`owner_label` before parsing.

    Attributes:
        profile: The profile built (``smoke`` | ``full``).
        seed: The filler RNG seed (irrelevant for ``smoke``, recorded anyway).
        message_count: Parsed messages written (scripted + filler); tapback
            rows are reactions, not messages, and are excluded.
        scripted_count: Messages from the hand-authored script.
        filler_count: Generated filler messages (0 for ``smoke``).
        tapback_count: Reaction rows written (they fold onto their targets).
        thread_counts: Message count per thread guid.
        guids: Every message guid written, in write order — the set the
            golden QA's ``required_guids`` is validated against.
        first_timestamp: Earliest message time (aware UTC).
        last_timestamp: Latest message time (aware UTC).
        owner_label: Display name the script was authored with for
            ``is_from_me`` messages.
        handle_names: ``handle`` → display name for every participant
            (scripted and filler).
    """

    profile: str
    seed: int
    message_count: int
    scripted_count: int
    filler_count: int
    tapback_count: int
    thread_counts: dict[str, int]
    guids: list[str]
    first_timestamp: datetime
    last_timestamp: datetime
    owner_label: str
    handle_names: dict[str, str]

apple_date(when)

Encode a datetime as a message.date value in its era's units.

Rows before :data:_ERA_BOUNDARY are written in seconds (the pre-macOS 10.13/iOS 11 format) and later rows in nanoseconds, so a single fixture database contains both eras exactly as a decade-spanning real export does.

Parameters:

Name	Type	Description	Default
when	datetime	The send time (aware UTC).	required

Returns:

Type	Description
int	The raw message.date integer.

Source code in varagity/eval/graph_fixtures.py

def apple_date(when: datetime) -> int:
    """Encode a datetime as a ``message.date`` value in its era's units.

    Rows before :data:`_ERA_BOUNDARY` are written in seconds (the pre-macOS
    10.13/iOS 11 format) and later rows in nanoseconds, so a single fixture
    database contains both eras exactly as a decade-spanning real export
    does.

    Args:
        when: The send time (aware UTC).

    Returns:
        The raw ``message.date`` integer.
    """
    seconds = int((when - _APPLE_EPOCH).total_seconds())
    return seconds if when < _ERA_BOUNDARY else seconds * _NANOS_PER_SECOND

assert_filler_clean(text, *, context)

Reject filler text that mentions a golden entity or fact term.

The discipline that keeps the full profile discriminative: every designed distractor lives in the scripted conversations, so filler must be inert. Matching is whole-word (optionally pluralised) and case-insensitive — "warm" is not "arm".

Parameters:

Name	Type	Description	Default
text	str	The generated filler string (message, name, or thread label).	required
context	str	Where it came from, for the error message.	required

Raises:

Type	Description
ValueError	If text contains a :data:FILLER_BLOCKLIST term.

Source code in varagity/eval/graph_fixtures.py

def assert_filler_clean(text: str, *, context: str) -> None:
    """Reject filler text that mentions a golden entity or fact term.

    The discipline that keeps the ``full`` profile discriminative: every
    designed distractor lives in the scripted conversations, so filler must
    be inert. Matching is whole-word (optionally pluralised) and
    case-insensitive — "warm" is not "arm".

    Args:
        text: The generated filler string (message, name, or thread label).
        context: Where it came from, for the error message.

    Raises:
        ValueError: If ``text`` contains a :data:`FILLER_BLOCKLIST` term.
    """
    hit = _BLOCKED_RE.search(text)
    if hit is not None:
        raise ValueError(
            f"filler {context} mentions the golden term {hit.group(0)!r}: {text!r} — "
            "filler must never compete with the scripted distractors"
        )

build_fixture_chat_db(dest, *, profile='smoke', seed=13, message_target=None, script_path=SCRIPT_PATH, blob_path=ATTRIBUTED_BODY_PATH)

Build the synthetic chat.db for a profile and describe what it holds.

The scripted conversations are written verbatim in every profile (golden QA is authored against them); full appends deterministic filler until the message target is met. Determinism is over parsed content, not file bytes — SQLite's page layout is not the contract.

Parameters:

Name	Type	Description	Default
dest	Path	Destination .db path; any previous build is replaced.	required
profile	str	smoke (scripted only) or full (scripted + filler to 10,000 messages).	'smoke'
seed	int	RNG seed for the filler (same seed ⇒ same parsed content).	13
message_target	int | None	Overrides the profile's minimum parsed-message count — the knob unit tests use to exercise full cheaply.	None
script_path	Path	The hand-authored conversation script.	SCRIPT_PATH
blob_path	Path	The captured attributedBody blob for NULL-text rows.	ATTRIBUTED_BODY_PATH

Returns:

Name	Type	Description
The	FixtureManifest	class:FixtureManifest describing the build.

Raises:

Type	Description
FileNotFoundError	If the script or blob fixture is missing.
ValueError	If profile is unknown, the script is inconsistent, or generated filler mentions a golden term.

Source code in varagity/eval/graph_fixtures.py

def build_fixture_chat_db(
    dest: Path,
    *,
    profile: str = "smoke",
    seed: int = 13,
    message_target: int | None = None,
    script_path: Path = SCRIPT_PATH,
    blob_path: Path = ATTRIBUTED_BODY_PATH,
) -> FixtureManifest:
    """Build the synthetic ``chat.db`` for a profile and describe what it holds.

    The scripted conversations are written verbatim in every profile (golden
    QA is authored against them); ``full`` appends deterministic filler until
    the message target is met. Determinism is over *parsed content*, not file
    bytes — SQLite's page layout is not the contract.

    Args:
        dest: Destination ``.db`` path; any previous build is replaced.
        profile: ``smoke`` (scripted only) or ``full`` (scripted + filler to
            > 10,000 messages).
        seed: RNG seed for the filler (same seed ⇒ same parsed content).
        message_target: Overrides the profile's minimum parsed-message count
            — the knob unit tests use to exercise ``full`` cheaply.
        script_path: The hand-authored conversation script.
        blob_path: The captured ``attributedBody`` blob for NULL-``text``
            rows.

    Returns:
        The :class:`FixtureManifest` describing the build.

    Raises:
        FileNotFoundError: If the script or blob fixture is missing.
        ValueError: If ``profile`` is unknown, the script is inconsistent, or
            generated filler mentions a golden term.
    """
    if profile not in PROFILE_TARGETS:
        raise ValueError(
            f"unknown fixture profile {profile!r}; expected one of {list(PROFILE_TARGETS)}"
        )
    target = PROFILE_TARGETS[profile] if message_target is None else message_target

    script = _load_script(script_path)
    blob = blob_path.read_bytes()
    rng = random.Random(seed)

    handle_rowids: dict[str, int] = {}
    handles: list[tuple[int, str]] = []
    handle_names: dict[str, str] = {}
    for person in script["participants"]:
        rowid = len(handles) + 1
        handle_rowids[person["key"]] = rowid
        handles.append((rowid, person["handle"]))
        handle_names[person["handle"]] = person["name"]

    chat_rowids: dict[str, int] = {}
    chat_guids: dict[int, str] = {}
    chats: list[tuple[int, str, str]] = []
    chat_handles: list[tuple[int, int]] = []
    for thread in script["threads"]:
        rowid = len(chats) + 1
        chat_rowids[thread["key"]] = rowid
        chat_guids[rowid] = thread["guid"]
        chats.append((rowid, thread["guid"], thread["display_name"]))
        chat_handles.extend((rowid, handle_rowids[key]) for key in thread["participants"])

    messages, reactions = _scripted_rows(
        script, handle_rowids=handle_rowids, chat_rowids=chat_rowids, blob=blob
    )
    scripted_count = len(messages)

    filler: list[_Row] = []
    if target > scripted_count:
        people = _filler_people(rng)
        filler_handle_rowids: dict[str, int] = {}
        for handle, name in people:
            rowid = len(handles) + 1
            filler_handle_rowids[handle] = rowid
            handles.append((rowid, handle))
            handle_names[handle] = name
        filler_chats: list[tuple[int, list[str]]] = []
        for index in range(_FILLER_THREAD_COUNT):
            rowid = len(chats) + 1
            guid = f"fx-thread-fill-{index:02d}"
            # Every third filler thread is an unnamed group, so the parser's
            # participant-name fallback runs at scale too.
            named = _FILLER_THREAD_NAMES[index % len(_FILLER_THREAD_NAMES)]
            label = "" if index % 3 == 0 else named
            if label:
                assert_filler_clean(label, context="thread name")
            chat_guids[rowid] = guid
            chats.append((rowid, guid, label))
            start = index % len(people)
            members = [
                people[(start + offset) % len(people)][0]
                for offset in range(rng.randint(*_FILLER_PARTICIPANTS_PER_THREAD))
            ]
            chat_handles.extend((rowid, filler_handle_rowids[handle]) for handle in members)
            filler_chats.append((rowid, members))
        filler = _filler_rows(
            count=target - scripted_count,
            rng=rng,
            handle_rowids=filler_handle_rowids,
            filler_chats=filler_chats,
        )

    all_messages = [*messages, *filler]
    _write_db(
        dest,
        handles=handles,
        chats=chats,
        chat_handles=chat_handles,
        rows=[*all_messages, *reactions],
    )

    thread_counts: dict[str, int] = {}
    for row in all_messages:
        guid = chat_guids[row.chat_rowid]
        thread_counts[guid] = thread_counts.get(guid, 0) + 1
    times = [row.when for row in all_messages]
    manifest = FixtureManifest(
        profile=profile,
        seed=seed,
        message_count=len(all_messages),
        scripted_count=scripted_count,
        filler_count=len(filler),
        tapback_count=len(reactions),
        thread_counts=thread_counts,
        guids=[row.guid for row in all_messages],
        first_timestamp=min(times),
        last_timestamp=max(times),
        owner_label=script["owner_name"],
        handle_names=handle_names,
    )
    logger.info(
        "built %s fixture at %s: %d message(s) (%d scripted + %d filler), %d thread(s), %s..%s",
        profile,
        dest,
        manifest.message_count,
        manifest.scripted_count,
        manifest.filler_count,
        len(thread_counts),
        manifest.first_timestamp.date(),
        manifest.last_timestamp.date(),
    )
    return manifest

varagity.eval.graph_eval

The graph harness — eval graph (spec_graphrag §12, §8.2).

The fourth eval subcommand, and the one that decided ADR-017: build the synthetic message corpus, run every selected registered graph engine over it sequentially, and score each engine's answers — because a graph engine returns prose, not chunk ids. With the ADR accepted it stays on as the graph regression guard: a retrieval-side change (query mode, the e5 prefix, a rerank hook) is one --skip-build re-score away from a number.

Structurally this is :func:varagity.eval.evaluate.run_chat_eval again (registry-enumerated engine loop, pinned settings, fact-anchored scoring, :func:~varagity.eval.evaluate.write_results persistence, a thin Prefect shell in :mod:varagity.pipeline.eval_flow) with three differences that come from the subject matter:

• No stores, no testcontainers. Graph engines self-store in their own working directories under data/eval/graph/ (plan decision #9), so the only external needs are the live llama.cpp and infinity services.
• Scoring is over answers. Each golden entry carries expected_facts (an AND of OR-groups, matched case-insensitively as substrings of the answer — :func:~varagity.eval.evaluate.resolve_golden_by_fact's trick applied to prose) and required_guids (provenance). An engine that surfaces no source ids scores None provenance rather than 0: that honesty is criterion §8.2#4 data. Question kind slices the report and is never shown to an engine (plan decision #10).
• The parser's naming settings are pinned from the fixture manifest. Sender display names come from GRAPH_OWNER_ALIASES / GRAPH_HANDLE_NAMES, so a parse without those pins yields raw phone handles and every golden naming "Bob" fails against a corpus that only ever says "+12145550101". :func:manifest_settings_pins derives them from what the generator actually wrote, and :func:pinned_graph_settings applies them the way the rest of the harness pins settings (environment export + cache clear — see :func:varagity.eval.evaluate.pinned_eval_settings).

Three cost controls shape the run, because a full-profile index is hours per engine on a single-slot llama.cpp: --engine runs one engine per session, --skip-build reuses the corpus and the engines' working directories so scoring can be iterated without re-paying an index, and --message-target caps the corpus below its profile's size — Graphiti indexes at ~51 s/message, which makes an uncapped 10,001-message build about six days, so its full-profile seat is a capped slice and the cap is reported with the numbers. A capped run is a different corpus: :func:corpus_stem gives it its own database, manifest, and per-engine working directories (full-mt1000), so a capped run can never clobber the uncapped corpus another engine is still indexing against. The scripted messages are written in every corpus regardless of the cap, so the golden set's provenance anchors always resolve.

The incremental re-index check (criterion §8.2#3) runs only on the smoke profile and only when building: a second build over an overlapping batch — the same messages plus :data:INCREMENTAL_DELTA_MESSAGES new ones — measures what a re-export of a grown chat.db costs. The delta messages are generator filler, which is blocklist-guaranteed never to mention a golden term, so the check cannot perturb the scores that follow it.

Engine libraries stay call-time imports (stage-1 decision #8) even though the shipped engine is a main dependency now: nothing here needs an engine installed to import, and selecting one whose library is missing raises a clear error naming the sync that would fix it.

corpus_stem(profile, message_target=None)

Name the corpus (and the engines' workdir tier) for one run.

A --message-target run indexes a different corpus from its profile's uncapped one, so it may not share either the generated database or the engines' working directories: overwriting full.db while another engine's multi-day full-profile build is still running against it would corrupt that run's provenance silently, and reusing full/<engine>/ would score a capped run against an uncapped graph.

Parameters:

Name	Type	Description	Default
profile	str	smoke or full.	required
message_target	int | None	The --message-target override, or None for the profile's own size.	None

Returns:

Type	Description
str	The file stem for the corpus and its manifest, and the directory name
str	holding each engine's workdir: full uncapped, full-mt1000
str	capped.

Source code in varagity/eval/graph_eval.py

def corpus_stem(profile: str, message_target: int | None = None) -> str:
    """Name the corpus (and the engines' workdir tier) for one run.

    A ``--message-target`` run indexes a *different corpus* from its profile's
    uncapped one, so it may not share either the generated database or the
    engines' working directories: overwriting ``full.db`` while another
    engine's multi-day full-profile build is still running against it would
    corrupt that run's provenance silently, and reusing ``full/<engine>/``
    would score a capped run against an uncapped graph.

    Args:
        profile: ``smoke`` or ``full``.
        message_target: The ``--message-target`` override, or ``None`` for the
            profile's own size.

    Returns:
        The file stem for the corpus and its manifest, and the directory name
        holding each engine's workdir: ``full`` uncapped, ``full-mt1000``
        capped.
    """
    return profile if message_target is None else f"{profile}-mt{message_target}"

engine_versions(names)

Stamp the installed version of each engine's library.

Parameters:

Name	Type	Description	Default
names	Sequence[str]	Engine names being run.	required

Returns:

Type	Description
dict[str, str | None]	Engine name → installed distribution version, or None when the
dict[str, str | None]	library is not installed (which a run cannot actually reach, but a
dict[str, str | None]	results file re-read later can).

Source code in varagity/eval/graph_eval.py

def engine_versions(names: Sequence[str]) -> dict[str, str | None]:
    """Stamp the installed version of each engine's library.

    Args:
        names: Engine names being run.

    Returns:
        Engine name → installed distribution version, or ``None`` when the
        library is not installed (which a run cannot actually reach, but a
        results file re-read later can).
    """
    versions: dict[str, str | None] = {}
    for name in names:
        distribution = ENGINE_DISTRIBUTIONS.get(name)
        if distribution is None:
            versions[name] = None
            continue
        try:
            versions[name] = version(distribution)
        except PackageNotFoundError:
            versions[name] = None
    return versions

manifest_settings_pins(*manifests)

Derive the parser's naming pins from what the generator wrote.

Sender display names are a settings concern (:attr:~varagity.config.Settings.graph_handle_name_map, :attr:~varagity.config.Settings.graph_owner_label), so parsing a fixture without these pins yields raw handles and every golden question naming a participant is unanswerable. GRAPH_HANDLE_NAMES_FILE is pinned empty alongside them: a host .env pointing it at a contacts file that does not exist on this machine would otherwise fail the parse.

Parameters:

Name	Type	Description	Default
*manifests	FixtureManifest	The manifests of every corpus this run will parse (the base corpus, plus the incremental delta corpus when the check runs). Their handle maps are merged, later manifests winning.	()

Returns:

Type	Description
dict[str, str]	The pins, ready for :func:pinned_graph_settings.

Raises:

Type	Description
ValueError	If no manifest is given, or two manifests disagree about the owner label (a mismatch would silently split the owner into two graph entities).

Source code in varagity/eval/graph_eval.py

def manifest_settings_pins(*manifests: FixtureManifest) -> dict[str, str]:
    """Derive the parser's naming pins from what the generator wrote.

    Sender display names are a settings concern
    (:attr:`~varagity.config.Settings.graph_handle_name_map`,
    :attr:`~varagity.config.Settings.graph_owner_label`), so parsing a fixture
    without these pins yields raw handles and every golden question naming a
    participant is unanswerable. ``GRAPH_HANDLE_NAMES_FILE`` is pinned empty
    alongside them: a host ``.env`` pointing it at a contacts file that does
    not exist on this machine would otherwise fail the parse.

    Args:
        *manifests: The manifests of every corpus this run will parse (the
            base corpus, plus the incremental delta corpus when the check
            runs). Their handle maps are merged, later manifests winning.

    Returns:
        The pins, ready for :func:`pinned_graph_settings`.

    Raises:
        ValueError: If no manifest is given, or two manifests disagree about
            the owner label (a mismatch would silently split the owner into
            two graph entities).
    """
    if not manifests:
        raise ValueError("manifest_settings_pins needs at least one manifest")
    labels = {manifest.owner_label for manifest in manifests}
    if len(labels) > 1:
        raise ValueError(f"corpora disagree about the owner label: {sorted(labels)}")
    handle_names: dict[str, str] = {}
    for manifest in manifests:
        handle_names.update(manifest.handle_names)
    return {
        "GRAPH_OWNER_ALIASES": manifests[0].owner_label,
        "GRAPH_HANDLE_NAMES": ",".join(
            f"{handle}={name}" for handle, name in sorted(handle_names.items())
        ),
        "GRAPH_HANDLE_NAMES_FILE": "",
    }

match_fact_groups(answer, expected_facts)

Report which of an entry's OR-groups the answer satisfies.

expected_facts is an AND of OR-groups (spec_graphrag §12): a group is satisfied when any of its variants appears in the answer, matched case-insensitively as a substring — the same rule the chunk-RAG sweep uses against chunk text, applied to prose.

Parameters:

Name	Type	Description	Default
answer	str	The engine's answer text.	required
expected_facts	Sequence[Sequence[str]]	The entry's fact groups.	required

Returns:

Type	Description
list[bool]	One boolean per group, in group order.

Source code in varagity/eval/graph_eval.py

def match_fact_groups(answer: str, expected_facts: Sequence[Sequence[str]]) -> list[bool]:
    """Report which of an entry's OR-groups the answer satisfies.

    ``expected_facts`` is an AND of OR-groups (spec_graphrag §12): a group is
    satisfied when any of its variants appears in the answer, matched
    case-insensitively as a substring — the same rule the chunk-RAG sweep uses
    against chunk text, applied to prose.

    Args:
        answer: The engine's answer text.
        expected_facts: The entry's fact groups.

    Returns:
        One boolean per group, in group order.
    """
    haystack = answer.lower()
    return [any(variant.lower() in haystack for variant in group) for group in expected_facts]

measure_graph_engine(session, entries, *, mode=None, verbose=0)

Ask one open engine session every golden question and score the answers.

Questions are put verbatim — an entry's kind is report metadata and is never shown to an engine (plan decision #10). A query that raises is recorded as a failed query (zero fact recall, None provenance) rather than aborting the engine: at hours per index, losing a whole engine's numbers to one bad question would be the expensive kind of strict, and engine failure counts are themselves criterion §8.2#2 data — the same posture :attr:~varagity.graph.records.BuildReport.failures takes for indexing.

Parameters:

Name	Type	Description	Default
session	GraphSession	The open engine session to interrogate.	required
entries	Sequence[GraphGoldenEntry]	The golden questions, in report order.	required
mode	str | None	Engine query mode override (None uses each adapter's primary mode — the --mode escape hatch of plan decision #13).	None
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
dict[str, Any]	{"summary": …, "by_kind": …, "queries": […]} — the overall slice,
dict[str, Any]	one slice per :data:~varagity.eval.datasets.GRAPH_GOLDEN_KINDS that
dict[str, Any]	has questions, and the full per-query detail (answer text included,
dict[str, Any]	for the ADR autopsy).

Raises:

Type	Description
ValueError	If entries is empty.

Source code in varagity/eval/graph_eval.py

def measure_graph_engine(
    session: GraphSession,
    entries: Sequence[GraphGoldenEntry],
    *,
    mode: str | None = None,
    verbose: int = 0,
) -> dict[str, Any]:
    """Ask one open engine session every golden question and score the answers.

    Questions are put verbatim — an entry's ``kind`` is report metadata and is
    never shown to an engine (plan decision #10). A query that raises is
    recorded as a failed query (zero fact recall, ``None`` provenance) rather
    than aborting the engine: at hours per index, losing a whole engine's
    numbers to one bad question would be the expensive kind of strict, and
    engine failure counts are themselves criterion §8.2#2 data — the same
    posture :attr:`~varagity.graph.records.BuildReport.failures` takes for
    indexing.

    Args:
        session: The open engine session to interrogate.
        entries: The golden questions, in report order.
        mode: Engine query mode override (``None`` uses each adapter's primary
            mode — the ``--mode`` escape hatch of plan decision #13).
        verbose: Validated console verbosity (0–2).

    Returns:
        ``{"summary": …, "by_kind": …, "queries": […]}`` — the overall slice,
        one slice per :data:`~varagity.eval.datasets.GRAPH_GOLDEN_KINDS` that
        has questions, and the full per-query detail (answer text included,
        for the ADR autopsy).

    Raises:
        ValueError: If ``entries`` is empty.
    """
    if not entries:
        raise ValueError("measure_graph_engine needs at least one golden entry")
    records: list[dict[str, Any]] = []
    for entry in entries:
        required = set(entry.required_guids)
        record: dict[str, Any] = {
            "query": entry.query,
            "kind": entry.kind,
            "mode": mode,
            "latency_s": None,
            "fact_recall": 0.0,
            "missed_facts": [list(group) for group in entry.expected_facts],
            "provenance_recall": None,
            "required_guids": list(entry.required_guids),
            "matched_guids": [],
            "evidence": None,
            "answer": "",
            "error": None,
        }
        try:
            answer = session.query(entry.query, mode=mode, verbose=verbose)
        except Exception as exc:  # broad on purpose: an engine failure is §8.2#2 data
            logger.warning("query failed for %r: %s", entry.query, exc)
            record["error"] = f"{type(exc).__name__}: {exc}"
            records.append(record)
            continue
        matched = match_fact_groups(answer.answer, entry.expected_facts)
        evidence = answer.evidence
        surfaced = [guid for guid in evidence.message_guids if guid in required]
        record.update(
            {
                "mode": answer.mode,
                "latency_s": round(answer.latency_s, 2),
                "fact_recall": round(sum(matched) / len(matched), 4),
                "missed_facts": [
                    list(group)
                    for group, hit in zip(entry.expected_facts, matched, strict=True)
                    if not hit
                ],
                "provenance_recall": provenance_recall(
                    evidence.message_guids, entry.required_guids
                ),
                "matched_guids": surfaced,
                "evidence": {
                    "entities": len(evidence.entities),
                    "relations": len(evidence.relations),
                    "communities": len(evidence.communities),
                    "message_guids": len(evidence.message_guids),
                },
                "answer": answer.answer,
            }
        )
        records.append(record)
    return {
        "summary": _summarize(records),
        "by_kind": {
            kind: _summarize(selected)
            for kind in GRAPH_GOLDEN_KINDS
            if (selected := [record for record in records if record["kind"] == kind])
        },
        "queries": records,
    }

open_engine_session(name, workdir)

Open one registered engine's session, translating a missing library.

Adapters import their engine libraries inside session() (plan decision

8), so a missing library surfaces here as a bare ModuleNotFoundError

from somewhere inside a third-party package. Translating it names the engine and the install command instead.

Parameters:

Name	Type	Description	Default
name	str	Registered engine name.	required
workdir	Path	The engine's working directory.	required

Yields:

Type	Description
GraphSession	The open session (closed on exit — teardown is not optional; Graphiti
GraphSession	runs an embedded redis-server subprocess).

Raises:

Type	Description
RuntimeError	If the engine's library is not installed.

Source code in varagity/eval/graph_eval.py

@contextmanager
def open_engine_session(name: str, workdir: Path) -> Iterator[GraphSession]:
    """Open one registered engine's session, translating a missing library.

    Adapters import their engine libraries inside ``session()`` (plan decision
    #8), so a missing library surfaces here as a bare ``ModuleNotFoundError``
    from somewhere inside a third-party package. Translating it names the
    engine and the install command instead.

    Args:
        name: Registered engine name.
        workdir: The engine's working directory.

    Yields:
        The open session (closed on exit — teardown is not optional; Graphiti
        runs an embedded ``redis-server`` subprocess).

    Raises:
        RuntimeError: If the engine's library is not installed.
    """
    engine = get_graph_engine(name)
    try:
        with engine.session(workdir) as session:
            yield session
    except ModuleNotFoundError as exc:
        raise RuntimeError(
            f"graph engine {name!r} needs a library that is not installed "
            f"(no module named {exc.name!r}). The shipped engine is a main dependency — "
            "run `uv sync`, then `uv run main.py eval graph`."
        ) from exc

pinned_graph_settings(**pins)

Apply graph settings pins for a block, restoring them on exit.

The graph counterpart of :func:varagity.eval.evaluate.pinned_eval_settings, and the same mechanism for the same reason: deep code (here the iMessage parser) resolves get_settings() internally, and the eval harness is a single-threaded CLI path where an environment override is safe. There is no standing pin set — every pin is derived from the fixture manifest the run just built (:func:manifest_settings_pins), because the names the corpus was written with are the only correct ones.

Parameters:

Name	Type	Description	Default
**pins	str	Settings to export for this block (e.g. GRAPH_OWNER_ALIASES="Sam").	{}

Yields:

Type	Description
None	Nothing; the pinned settings are active inside the block.

Source code in varagity/eval/graph_eval.py

@contextmanager
def pinned_graph_settings(**pins: str) -> Iterator[None]:
    """Apply graph settings pins for a block, restoring them on exit.

    The graph counterpart of
    :func:`varagity.eval.evaluate.pinned_eval_settings`, and the same
    mechanism for the same reason: deep code (here the iMessage parser)
    resolves ``get_settings()`` internally, and the eval harness is a
    single-threaded CLI path where an environment override is safe. There is
    no standing pin set — every pin is derived from the fixture manifest the
    run just built (:func:`manifest_settings_pins`), because the names the
    corpus was written with are the only correct ones.

    Args:
        **pins: Settings to export for this block (e.g.
            ``GRAPH_OWNER_ALIASES="Sam"``).

    Yields:
        Nothing; the pinned settings are active inside the block.
    """
    saved = {name: os.environ.get(name) for name in pins}
    os.environ.update(pins)
    get_settings.cache_clear()
    try:
        yield
    finally:
        for name, value in saved.items():
            if value is None:
                os.environ.pop(name, None)
            else:
                os.environ[name] = value
        get_settings.cache_clear()

prepare_corpus(corpus_dir, *, profile, seed=FIXTURE_SEED, message_target=None, name=None, reuse=False)

Build (or reuse) one fixture corpus and return it with its manifest.

Parameters:

Name	Type	Description	Default
corpus_dir	Path	Directory holding the generated databases (created if absent).	required
profile	str	smoke or full (see :data:~varagity.eval.graph_fixtures.PROFILE_TARGETS).	required
seed	int	Filler RNG seed.	FIXTURE_SEED
message_target	int | None	Overrides the profile's message count (the incremental delta corpus uses it to ask for "the corpus plus a few").	None
name	str | None	File stem for the database and its manifest; defaults to profile.	None
reuse	bool	Reuse an existing database and its manifest instead of regenerating (--skip-build). A missing or unreadable manifest falls back to building, since without it the run cannot pin the parser's naming settings.	False

Returns:

Type	Description
tuple[Path, FixtureManifest]	The (database path, manifest) pair.

Raises:

Type	Description
FileNotFoundError	If the scripted-message or blob fixture is missing.
ValueError	If profile is unknown or the script is inconsistent.

Source code in varagity/eval/graph_eval.py

def prepare_corpus(
    corpus_dir: Path,
    *,
    profile: str,
    seed: int = FIXTURE_SEED,
    message_target: int | None = None,
    name: str | None = None,
    reuse: bool = False,
) -> tuple[Path, FixtureManifest]:
    """Build (or reuse) one fixture corpus and return it with its manifest.

    Args:
        corpus_dir: Directory holding the generated databases (created if
            absent).
        profile: ``smoke`` or ``full`` (see
            :data:`~varagity.eval.graph_fixtures.PROFILE_TARGETS`).
        seed: Filler RNG seed.
        message_target: Overrides the profile's message count (the
            incremental delta corpus uses it to ask for "the corpus plus a
            few").
        name: File stem for the database and its manifest; defaults to
            ``profile``.
        reuse: Reuse an existing database **and** its manifest instead of
            regenerating (``--skip-build``). A missing or unreadable manifest
            falls back to building, since without it the run cannot pin the
            parser's naming settings.

    Returns:
        The ``(database path, manifest)`` pair.

    Raises:
        FileNotFoundError: If the scripted-message or blob fixture is missing.
        ValueError: If ``profile`` is unknown or the script is inconsistent.
    """
    stem = name or profile
    db_path = corpus_dir / f"{stem}.db"
    manifest_path = corpus_dir / f"{stem}{MANIFEST_SUFFIX}"
    if reuse and db_path.is_file() and manifest_path.is_file():
        manifest = FixtureManifest.model_validate_json(manifest_path.read_text(encoding="utf-8"))
        logger.info(
            "reusing the %s fixture corpus at %s (%d message(s))",
            stem,
            db_path,
            manifest.message_count,
        )
        return db_path, manifest
    if reuse:
        logger.info("no reusable %s corpus beside %s — building it", stem, manifest_path)
    manifest = build_fixture_chat_db(
        db_path, profile=profile, seed=seed, message_target=message_target
    )
    manifest_path.write_text(manifest.model_dump_json(indent=2) + "\n", encoding="utf-8")
    return db_path, manifest

provenance_recall(surfaced, required)

Score how much of an entry's required provenance the engine surfaced.

Parameters:

Name	Type	Description	Default
surfaced	Sequence[str]	Message guids the engine's evidence carried.	required
required	Sequence[str]	The entry's required_guids (non-empty by schema).	required

Returns:

Type	Description
float | None	|surfaced ∩ required| / |required|, or None when the engine
float | None	surfaced no guids at all — reported honestly as "this engine cannot
float | None	cite messages" rather than as a zero score (criterion §8.2#4).

Raises:

Type	Description
ValueError	If required is empty (nothing to divide by).

Source code in varagity/eval/graph_eval.py

def provenance_recall(surfaced: Sequence[str], required: Sequence[str]) -> float | None:
    """Score how much of an entry's required provenance the engine surfaced.

    Args:
        surfaced: Message guids the engine's evidence carried.
        required: The entry's ``required_guids`` (non-empty by schema).

    Returns:
        ``|surfaced ∩ required| / |required|``, or ``None`` when the engine
        surfaced no guids at all — reported honestly as "this engine cannot
        cite messages" rather than as a zero score (criterion §8.2#4).

    Raises:
        ValueError: If ``required`` is empty (nothing to divide by).
    """
    if not required:
        raise ValueError("provenance_recall needs at least one required guid")
    if not surfaced:
        return None
    hits = set(surfaced) & set(required)
    return len(hits) / len(required)

run_graph_eval(*, profile='smoke', engines=None, mode=None, skip_build=False, message_target=None, eval_root=GRAPH_EVAL_ROOT, golden_path=GRAPH_GOLDEN_PATH, results_dir=RESULTS_DIR, seed=FIXTURE_SEED, verbose=None)

Run the graph bake-off over the fixture corpus; persist results (§12).

The sequence, once per selected engine (sequentially — stage-1 decision

15): wipe its working directory, index the corpus, run the incremental

re-index check, then ask every golden question and score the answers. The corpus is generated first and its manifest pins the parser's naming settings, so the messages the engines index say "Bob Nakamura" rather than a phone number.

This harness decided ADR-017 and stays on as the graph regression guard, exactly as the chat eval did for ADR-011.

Parameters:

Name	Type	Description	Default
profile	str	Fixture corpus profile — smoke (~226 scripted messages, the adapter/iteration corpus) or full (> 10,000 messages across a decade, the bake-off corpus).	'smoke'
engines	Sequence[str] | None	Engine names to run; None runs every registered engine.	None
mode	str | None	Engine query mode override for every question (None uses each adapter's primary mode).	None
skip_build	bool	Reuse the existing corpus and each engine's already-built working directory, and skip the incremental check — the scoring iteration loop, which must not re-pay a multi-hour index.	False
message_target	int | None	Cap the corpus at this many messages instead of the profile's own size — how an engine too slow for the uncapped corpus (Graphiti, ~51 s/message) still gets a full-profile seat. The scripted messages are always written, so the golden set holds; filler tops the corpus up to the cap. A capped run gets its own corpus and working directories (:func:corpus_stem).	None
eval_root	Path	Root for the generated corpus and the engines' working directories; resolved to an absolute path before use.	GRAPH_EVAL_ROOT
golden_path	Path	The graph golden Q&A file.	GRAPH_GOLDEN_PATH
results_dir	Path	Directory for the timestamped results JSON.	RESULTS_DIR
seed	int	Filler RNG seed for the generated corpus.	FIXTURE_SEED
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (also written to results_dir), with a
dict[str, Any]	"results_path" key naming the file.

Raises:

Type	Description
ValueError	If profile, message_target or verbose is invalid, an engine name is not registered, or a golden provenance anchor is missing from the corpus.
FileNotFoundError	If the golden set or a fixture input is missing.
RuntimeError	If a selected engine's library is not installed.

Source code in varagity/eval/graph_eval.py

def run_graph_eval(
    *,
    profile: str = "smoke",
    engines: Sequence[str] | None = None,
    mode: str | None = None,
    skip_build: bool = False,
    message_target: int | None = None,
    eval_root: Path = GRAPH_EVAL_ROOT,
    golden_path: Path = GRAPH_GOLDEN_PATH,
    results_dir: Path = RESULTS_DIR,
    seed: int = FIXTURE_SEED,
    verbose: int | None = None,
) -> dict[str, Any]:
    """Run the graph bake-off over the fixture corpus; persist results (§12).

    The sequence, once per selected engine (sequentially — stage-1 decision
    #15): wipe its working directory, index the corpus, run the incremental
    re-index check, then ask every golden question and score the answers. The
    corpus is generated first and its manifest pins the parser's naming
    settings, so the messages the engines index say "Bob Nakamura" rather than
    a phone number.

    This harness decided ADR-017 and stays on as the graph regression guard,
    exactly as the chat eval did for ADR-011.

    Args:
        profile: Fixture corpus profile — ``smoke`` (~226 scripted messages,
            the adapter/iteration corpus) or ``full`` (> 10,000 messages
            across a decade, the bake-off corpus).
        engines: Engine names to run; ``None`` runs every registered engine.
        mode: Engine query mode override for every question (``None`` uses
            each adapter's primary mode).
        skip_build: Reuse the existing corpus and each engine's already-built
            working directory, and skip the incremental check — the scoring
            iteration loop, which must not re-pay a multi-hour index.
        message_target: Cap the corpus at this many messages instead of the
            profile's own size — how an engine too slow for the uncapped
            corpus (Graphiti, ~51 s/message) still gets a full-profile seat.
            The scripted messages are always written, so the golden set holds;
            filler tops the corpus up to the cap. A capped run gets its own
            corpus and working directories (:func:`corpus_stem`).
        eval_root: Root for the generated corpus and the engines' working
            directories; resolved to an absolute path before use.
        golden_path: The graph golden Q&A file.
        results_dir: Directory for the timestamped results JSON.
        seed: Filler RNG seed for the generated corpus.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (also written to ``results_dir``), with a
        ``"results_path"`` key naming the file.

    Raises:
        ValueError: If ``profile``, ``message_target`` or ``verbose`` is
            invalid, an engine name is not registered, or a golden provenance
            anchor is missing from the corpus.
        FileNotFoundError: If the golden set or a fixture input is missing.
        RuntimeError: If a selected engine's library is not installed.
    """
    if profile not in PROFILE_TARGETS:
        raise ValueError(
            f"unknown corpus profile {profile!r}; expected one of {[*PROFILE_TARGETS]}"
        )
    if message_target is not None and message_target < 1:
        raise ValueError(f"message_target must be a positive count, got {message_target}")
    verbose = check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    selected = select_engines(engines)

    # Engines consume their workdir verbatim, from whatever cwd they happen
    # to run in (a bake-off candidate rejected a relative path outright,
    # 2026-07-26), so the CLI's repo-relative default must become absolute
    # before any session opens. `GraphService` resolves for the same reason.
    eval_root = eval_root.resolve()
    corpus_dir = eval_root / "corpus"
    stem = corpus_stem(profile, message_target)
    db_path, manifest = prepare_corpus(
        corpus_dir,
        profile=profile,
        seed=seed,
        message_target=message_target,
        name=stem,
        reuse=skip_build,
    )
    if message_target is not None and manifest.message_count > message_target:
        logger.warning(
            "message_target %d is below the %d scripted message(s), which every corpus "
            "carries verbatim (the golden set is authored against them): the %s corpus "
            "holds %d",
            message_target,
            manifest.scripted_count,
            stem,
            manifest.message_count,
        )

    # The incremental re-index check (criterion §8.2#3) measures the marginal
    # cost of a re-export: the same messages plus a few new ones. Smoke only —
    # on the full profile the base index already costs hours — and never under
    # --skip-build, whose entire purpose is not to index.
    delta_manifest: FixtureManifest | None = None
    delta_path: Path | None = None
    if profile == "smoke" and not skip_build:
        delta_path, delta_manifest = prepare_corpus(
            corpus_dir,
            profile="full",
            seed=seed,
            message_target=manifest.message_count + INCREMENTAL_DELTA_MESSAGES,
            name=f"{stem}-delta",
        )

    corpora = [manifest] if delta_manifest is None else [manifest, delta_manifest]
    pins = manifest_settings_pins(*corpora)
    engine_results: dict[str, Any] = {}

    with pinned_graph_settings(**pins):
        batch = batch_for_path(db_path, corpus_dir, verbose=verbose)
        delta_batch: MessageBatch | None = (
            None if delta_path is None else batch_for_path(delta_path, corpus_dir, verbose=verbose)
        )
        entries = load_graph_golden(golden_path)
        validate_golden_against_manifest(entries, manifest)

        for name in selected:
            workdir = eval_root / stem / name
            if not skip_build:
                shutil.rmtree(workdir, ignore_errors=True)
            workdir.mkdir(parents=True, exist_ok=True)
            logger.info(
                "graph eval: %s over the %s corpus (%d messages) in %s",
                name,
                stem,
                len(batch.messages),
                workdir,
            )
            started = time.monotonic()
            with open_engine_session(name, workdir) as session:
                build: dict[str, Any] | None = None
                if not skip_build:
                    report = session.build([batch], verbose=verbose)
                    build = _build_block(report, session.stats())
                    logger.info(
                        "%s indexed %d message(s) in %.1f s (%d failure(s)): %s",
                        name,
                        report.messages_seen,
                        report.wall_clock_s,
                        len(report.failures),
                        build["stats"],
                    )
                incremental: dict[str, Any] | None = None
                if delta_batch is not None:
                    before = session.stats()
                    delta_report = session.build([delta_batch], verbose=verbose)
                    incremental = _build_block(delta_report, session.stats())
                    incremental["new_messages"] = INCREMENTAL_DELTA_MESSAGES
                    incremental["stats_before"] = before.model_dump()
                    logger.info(
                        "%s re-indexed an overlapping batch (+%d new of %d) in %.1f s",
                        name,
                        INCREMENTAL_DELTA_MESSAGES,
                        delta_report.messages_seen,
                        delta_report.wall_clock_s,
                    )
                measured = measure_graph_engine(session, entries, mode=mode, verbose=verbose)
            engine_results[name] = {
                "workdir": str(workdir),
                "build": build,
                "incremental": incremental,
                "session_wall_clock_s": round(time.monotonic() - started, 2),
                **measured,
            }

    settings = get_settings()
    results: dict[str, Any] = {
        "kind": "graph_eval",
        "timestamp": datetime.now(UTC).isoformat(),
        "profile": profile,
        # The cap and the name it produced: a capped run's numbers are only
        # comparable to another run over the same corpus, and the stem is what
        # names both the corpus and every engine workdir recorded below.
        "message_target": message_target,
        "corpus_stem": stem,
        "seed": seed,
        "corpus_path": str(db_path),
        "golden_path": str(golden_path),
        "n_queries": len(entries),
        "kinds": list(GRAPH_GOLDEN_KINDS),
        "mode": mode,
        "skip_build": skip_build,
        "incremental_new_messages": None if delta_path is None else INCREMENTAL_DELTA_MESSAGES,
        "corpus": {
            "messages": manifest.message_count,
            "scripted": manifest.scripted_count,
            "filler": manifest.filler_count,
            "tapbacks": manifest.tapback_count,
            "threads": len(manifest.thread_counts),
            "first_timestamp": manifest.first_timestamp.isoformat(),
            "last_timestamp": manifest.last_timestamp.isoformat(),
        },
        "pinned_settings": pins,
        "chat_model": settings.BASE_MODEL,
        "embedding_model": settings.EMBEDDING_MODEL,
        "engine_versions": engine_versions(selected),
        "engines": engine_results,
    }
    results["results_path"] = str(write_results("graph", results, results_dir=results_dir))
    return results

select_engines(engines)

Resolve the --engine filter against the registry.

Parameters:

Name	Type	Description	Default
engines	Sequence[str] | None	Requested engine names, or None for every registered engine. The harness enumerates the registry rather than reading GRAPH_ENGINE: comparing engines is the whole point, and a candidate re-entering the seam must be measurable without an app-wide setting change.	required

Returns:

Type	Description
list[str]	The engine names to run, sorted — the harness runs them sequentially
list[str]	(stage-1 decision #15: there is one llama.cpp slot, so parallel
list[str]	engines would only queue behind each other).

Raises:

Type	Description
ValueError	If a requested name is not registered, or the filter selects nothing.

Source code in varagity/eval/graph_eval.py

def select_engines(engines: Sequence[str] | None) -> list[str]:
    """Resolve the ``--engine`` filter against the registry.

    Args:
        engines: Requested engine names, or ``None`` for every registered
            engine. The harness enumerates the registry rather than reading
            ``GRAPH_ENGINE``: comparing engines is the whole point, and a
            candidate re-entering the seam must be measurable without an
            app-wide setting change.

    Returns:
        The engine names to run, sorted — the harness runs them sequentially
        (stage-1 decision #15: there is one llama.cpp slot, so parallel
        engines would only queue behind each other).

    Raises:
        ValueError: If a requested name is not registered, or the filter
            selects nothing.
    """
    if engines is None:
        selected = sorted(GRAPH_ENGINE_REGISTRY)
    else:
        unknown = sorted({name for name in engines if name not in GRAPH_ENGINE_REGISTRY})
        if unknown:
            raise ValueError(
                f"unknown graph engine(s) {unknown}; registered: {sorted(GRAPH_ENGINE_REGISTRY)}"
            )
        selected = sorted(set(engines))
    if not selected:
        raise ValueError(f"no graph engines selected; registered: {sorted(GRAPH_ENGINE_REGISTRY)}")
    return selected

validate_golden_against_manifest(entries, manifest)

Check that every golden provenance anchor exists in the built corpus.

The golden-validation precedent (:func:varagity.eval.evaluate.validate_golden_against_store), and strict for the same reason: a required_guid that no longer exists would score every engine's provenance down for a drift in the fixture, silently. Runs before any engine opens a session, so the failure costs seconds rather than hours.

Parameters:

Name	Type	Description	Default
entries	Sequence[GraphGoldenEntry]	The loaded golden entries.	required
manifest	FixtureManifest	The built corpus's manifest.	required

Raises:

Type	Description
ValueError	If any required_guids entry is absent from the corpus.

Source code in varagity/eval/graph_eval.py

def validate_golden_against_manifest(
    entries: Sequence[GraphGoldenEntry], manifest: FixtureManifest
) -> None:
    """Check that every golden provenance anchor exists in the built corpus.

    The golden-validation precedent
    (:func:`varagity.eval.evaluate.validate_golden_against_store`), and strict
    for the same reason: a ``required_guid`` that no longer exists would score
    every engine's provenance down for a drift in the *fixture*, silently.
    Runs before any engine opens a session, so the failure costs seconds
    rather than hours.

    Args:
        entries: The loaded golden entries.
        manifest: The built corpus's manifest.

    Raises:
        ValueError: If any ``required_guids`` entry is absent from the corpus.
    """
    available = set(manifest.guids)
    missing = sorted(
        {guid for entry in entries for guid in entry.required_guids if guid not in available}
    )
    if missing:
        raise ValueError(
            f"{len(missing)} golden provenance anchor(s) are not in the {manifest.profile} "
            f"corpus ({manifest.message_count} messages): {missing} — the golden set and the "
            "fixture generator have drifted apart"
        )

varagity.eval.ocr_benchmark

OCR engine benchmark: EasyOCR vs Tesseract (plan decision #10, spec §16).

EasyOCR shipped as a provisional fallback engine behind the pluggable OCR_ENGINE factory; this benchmark supplies the data that picks the shipped default (recorded in ADR-004). Two measurements per engine:

• Intrinsic quality — the scanned/mixed fixture PDFs are parsed through the real two-pass parser and compared against their known ground-truth text (data/eval/ocr_truth/, transcribed from the images the fixtures were generated with): CER/WER via jiwer after normalization (lowercase, punctuation stripped — which also drops Docling's markdown markup — whitespace collapsed), plus pages/sec wall-clock over a warmed engine (one untimed warm-up conversion loads models first).

• Retrieval impact (supplementary) — the full fixtures corpus is ingested once per engine into ephemeral testcontainers stores (real embeddings; CONTEXTUALIZE off so the engine is the only variable), then recall@k / pass@k are reported for the scanned-document golden queries. The non-scanned corpus documents ride along as distractors — without them every query would trivially retrieve the whole index. OCR noise hits BM25 keyword matching hardest, so all three retrieval methods are reported. Directional on a tiny corpus, per the plan; a golden ref an engine's chunk boundaries can't resolve is counted as a guaranteed miss and reported.

jiwer lives in the eval dependency group; its import is deferred to call time like the group's other members.

character_error_rate(truth, hypothesis)

CER between normalized ground truth and OCR output.

Parameters:

Name	Type	Description	Default
truth	str	Ground-truth text (non-empty after normalization).	required
hypothesis	str	The engine's extracted text.	required

Returns:

Type	Description
float	The character error rate (0.0 = perfect; can exceed 1.0).

Raises:

Type	Description
ValueError	If the normalized ground truth is empty.

Source code in varagity/eval/ocr_benchmark.py

def character_error_rate(truth: str, hypothesis: str) -> float:
    """CER between normalized ground truth and OCR output.

    Args:
        truth: Ground-truth text (non-empty after normalization).
        hypothesis: The engine's extracted text.

    Returns:
        The character error rate (0.0 = perfect; can exceed 1.0).

    Raises:
        ValueError: If the normalized ground truth is empty.
    """
    import jiwer

    reference = normalize_ocr_text(truth)
    if not reference:
        raise ValueError("ground truth is empty after normalization")
    return float(jiwer.cer(reference, normalize_ocr_text(hypothesis)))

normalize_ocr_text(text)

Normalize text for OCR error metrics.

Lowercases, strips punctuation (which also removes Docling's markdown markup — # headings, table pipes — so the comparison measures recovered text, not export formatting), and collapses whitespace. Applied identically to ground truth and hypothesis.

Parameters:

Name	Type	Description	Default
text	str	Raw ground-truth or extracted text.	required

Returns:

Type	Description
str	The normalized text.

Source code in varagity/eval/ocr_benchmark.py

def normalize_ocr_text(text: str) -> str:
    """Normalize text for OCR error metrics.

    Lowercases, strips punctuation (which also removes Docling's markdown
    markup — ``#`` headings, table pipes — so the comparison measures
    recovered *text*, not export formatting), and collapses whitespace.
    Applied identically to ground truth and hypothesis.

    Args:
        text: Raw ground-truth or extracted text.

    Returns:
        The normalized text.
    """
    cleaned = text.lower().translate(str.maketrans("", "", string.punctuation))
    return " ".join(cleaned.split())

run_ocr_benchmark(*, corpus_root=EVAL_CORPUS, golden_path=GOLDEN_PATH, results_dir=RESULTS_DIR, ingest=ingest_corpus, verbose=None)

Benchmark the OCR engines and persist the results (spec §16).

Parameters:

Name	Type	Description	Default
corpus_root	Path	The eval corpus holding the OCR fixtures.	EVAL_CORPUS
golden_path	Path	The golden dataset file (its scanned-document queries drive the retrieval-impact measurement).	GOLDEN_PATH
results_dir	Path	Directory for the timestamped results JSON.	RESULTS_DIR
ingest	IngestCallable	Ingest callable (ingest_corpus, or the Prefect ingest_flow when invoked through the eval flow).	ingest_corpus
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (also written to results_dir), with a
dict[str, Any]	"results_path" key naming the file.

Raises:

Type	Description
ValueError	If verbose is invalid or no scanned-document golden queries exist.
FileNotFoundError	If a fixture, ground-truth file, or the golden set is missing.
RuntimeError	If an engine's ingest reports failed files (a partial index would silently under-report its retrieval scores).

Source code in varagity/eval/ocr_benchmark.py

def run_ocr_benchmark(
    *,
    corpus_root: Path = EVAL_CORPUS,
    golden_path: Path = GOLDEN_PATH,
    results_dir: Path = RESULTS_DIR,
    ingest: IngestCallable = ingest_corpus,
    verbose: int | None = None,
) -> dict[str, Any]:
    """Benchmark the OCR engines and persist the results (spec §16).

    Args:
        corpus_root: The eval corpus holding the OCR fixtures.
        golden_path: The golden dataset file (its scanned-document queries
            drive the retrieval-impact measurement).
        results_dir: Directory for the timestamped results JSON.
        ingest: Ingest callable (``ingest_corpus``, or the Prefect
            ``ingest_flow`` when invoked through the eval flow).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (also written to ``results_dir``), with a
        ``"results_path"`` key naming the file.

    Raises:
        ValueError: If ``verbose`` is invalid or no scanned-document golden
            queries exist.
        FileNotFoundError: If a fixture, ground-truth file, or the golden
            set is missing.
        RuntimeError: If an engine's ingest reports failed files (a partial
            index would silently under-report its retrieval scores).
    """
    verbose = check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    entries = resolve_golden(load_golden(golden_path), corpus_root)
    scanned = _scanned_entries(entries)
    if not scanned:
        raise ValueError(
            f"no golden queries target the OCR fixtures {sorted(OCR_FIXTURE_PAGES)} — "
            "the retrieval-impact measurement needs at least one"
        )
    embeddings = get_model("embedding")

    engines: dict[str, Any] = {}
    with ephemeral_stores(index_name="varagity_ocr_bench_bm25") as stores:
        retrievers: dict[str, Retriever] = {
            "semantic": SemanticRetriever(store=stores.store, embeddings=embeddings),
            "bm25": BM25Retriever(bm25=stores.bm25, store=stores.store),
            "hybrid": HybridRetriever(store=stores.store, bm25=stores.bm25, embeddings=embeddings),
        }
        for engine in ENGINES:
            # CONTEXTUALIZE off: the engine's text is the only variable
            # (and no LLM nondeterminism between the two engine runs).
            with pinned_eval_settings(OCR_ENGINE=engine, CONTEXTUALIZE="false"):
                intrinsic = _measure_intrinsic(engine, corpus_root, verbose)

                logger.info("[%s] ingesting the fixtures corpus into ephemeral stores", engine)
                started = time.monotonic()
                summary = ingest(
                    str(corpus_root),
                    store=stores.store,
                    bm25=stores.bm25,
                    embeddings=embeddings,
                    reingest=True,  # replace the previous engine's chunks
                    verbose=verbose,
                )
                ingest_seconds = round(time.monotonic() - started, 2)
                if summary.failed:
                    raise RuntimeError(f"[{engine}] ingest failed for {summary.failed} file(s)")
                unresolvable = validate_golden_against_store(scanned, stores, strict=False)

                retrieval: dict[str, Any] = {
                    "ingest_seconds": ingest_seconds,
                    "chunks_ingested": summary.chunks,
                    "unresolvable_golden_refs": unresolvable,
                    "methods": {},
                }
                for method, retriever in retrievers.items():
                    scores, _ranks = measure_retriever(retriever, scanned)
                    retrieval["methods"][method] = scores
                engines[engine] = {"intrinsic": intrinsic, "retrieval": retrieval}

    results: dict[str, Any] = {
        "kind": "ocr_benchmark",
        "timestamp": datetime.now(UTC).isoformat(),
        "corpus": str(corpus_root),
        "golden_path": str(golden_path),
        "fixtures": dict(OCR_FIXTURE_PAGES),
        "n_scanned_queries": len(scanned),
        "k_values": list(K_VALUES),
        "embedding_model": get_settings().EMBEDDING_MODEL,
        "contextualize": False,
        "engines": engines,
    }
    results["results_path"] = str(write_results("ocr", results, results_dir=results_dir))
    return results

word_error_rate(truth, hypothesis)

WER between normalized ground truth and OCR output.

Parameters:

Name	Type	Description	Default
truth	str	Ground-truth text (non-empty after normalization).	required
hypothesis	str	The engine's extracted text.	required

Returns:

Type	Description
float	The word error rate (0.0 = perfect; can exceed 1.0).

Raises:

Type	Description
ValueError	If the normalized ground truth is empty.

Source code in varagity/eval/ocr_benchmark.py

def word_error_rate(truth: str, hypothesis: str) -> float:
    """WER between normalized ground truth and OCR output.

    Args:
        truth: Ground-truth text (non-empty after normalization).
        hypothesis: The engine's extracted text.

    Returns:
        The word error rate (0.0 = perfect; can exceed 1.0).

    Raises:
        ValueError: If the normalized ground truth is empty.
    """
    import jiwer

    reference = normalize_ocr_text(truth)
    if not reference:
        raise ValueError("ground truth is empty after normalization")
    return float(jiwer.wer(reference, normalize_ocr_text(hypothesis)))