Evaluation¶

Every report holds counts and identifiers only, never document, question or answer text. Published numbers and what they do and do not show are on the Benchmarks page; the evaluation guide is here.

De-identification¶

openbtk.eval.deid.evaluate_deid ¶

evaluate_deid(engine: DeidEngine, documents: Iterable[_DocumentLike]) -> EvaluationResult

Run engine over documents and score it against their ground-truth spans.

documents is consumed once (streaming, ADR-0004): a full n2c2 corpus is never held in memory. A document's optional unscored_spans count (see openbtk.deid.labelled.LabelledDocument) is summed into the result so a report states how much annotation it did not score.

Source code in src/openbtk/eval/deid.py

def evaluate_deid(
    engine: DeidEngine,
    documents: Iterable[_DocumentLike],
) -> EvaluationResult:
    """Run ``engine`` over ``documents`` and score it against their
    ground-truth spans.

    ``documents`` is consumed once (streaming, ADR-0004): a full n2c2 corpus
    is never held in memory. A document's optional ``unscored_spans`` count
    (see ``openbtk.deid.labelled.LabelledDocument``) is summed into the
    result so a report states how much annotation it did not score.
    """
    totals: dict[PHICategory, list[int]] = {}
    binary_totals = [0, 0, 0]
    unscored = 0
    n = 0
    for doc in documents:
        result = engine.deidentify(doc.text, patient_id=doc.document_id)
        counts, binary = _score_document(doc, list(result.report.detections))
        for category, (tp, fp, fn) in counts.items():
            bucket = totals.setdefault(category, [0, 0, 0])
            bucket[0] += tp
            bucket[1] += fp
            bucket[2] += fn
        for j in range(3):
            binary_totals[j] += binary[j]
        unscored += int(getattr(doc, "unscored_spans", 0))
        n += 1
    return EvaluationResult(
        per_category={
            category: CategoryMetrics(
                true_positives=v[0], false_positives=v[1], false_negatives=v[2]
            )
            for category, v in totals.items()
        },
        binary=CategoryMetrics(
            true_positives=binary_totals[0],
            false_positives=binary_totals[1],
            false_negatives=binary_totals[2],
        ),
        n_documents=n,
        unscored_gold_spans=unscored,
    )

openbtk.eval.deid.EvaluationResult ¶

Bases: BaseModel

Scores for one benchmark run. Counts and category names only.

overall `property` ¶

overall: CategoryMetrics

Micro-averaged over categories (category-aware).

openbtk.eval.deid.CategoryMetrics ¶

Bases: BaseModel

True/false positive and false-negative counts, with derived scores.

An undefined ratio (no predictions, or no ground truth) is reported as 1.0 -- "nothing to get wrong" -- matching M2's accuracy gate; consult true_positives/false_negatives to tell "perfect" from "vacuous".

Example

m = CategoryMetrics(true_positives=3, false_positives=1, false_negatives=1) m.precision, m.recall (0.75, 0.75)

openbtk.eval.deid.report_dict ¶

report_dict(result: EvaluationResult) -> dict[str, object]

A JSON-serialisable report including the derived precision/recall/F1 (Pydantic properties are not part of model_dump). Counts and category names only -- safe to share.

Source code in src/openbtk/eval/deid.py

def report_dict(result: EvaluationResult) -> dict[str, object]:
    """A JSON-serialisable report including the derived precision/recall/F1
    (Pydantic properties are not part of ``model_dump``). Counts and category
    names only -- safe to share.
    """

    def scores(m: CategoryMetrics) -> dict[str, float | int]:
        return {
            "true_positives": m.true_positives,
            "false_positives": m.false_positives,
            "false_negatives": m.false_negatives,
            "precision": m.precision,
            "recall": m.recall,
            "f1": m.f1,
        }

    return {
        "n_documents": result.n_documents,
        "unscored_gold_spans": result.unscored_gold_spans,
        "overall": scores(result.overall),
        "binary": scores(result.binary),
        "per_category": {
            c.value: scores(m)
            for c, m in sorted(result.per_category.items(), key=lambda kv: kv[0].value)
        },
    }

openbtk.eval.deid.format_markdown ¶

format_markdown(result: EvaluationResult, *, title: str = 'De-identification') -> str

Render result as a Markdown report: counts and scores only, never any document text.

Source code in src/openbtk/eval/deid.py

def format_markdown(
    result: EvaluationResult, *, title: str = "De-identification"
) -> str:
    """Render ``result`` as a Markdown report: counts and scores only, never
    any document text."""
    lines = [
        f"### {title}",
        "",
        f"{result.n_documents} documents; matching: relaxed span overlap "
        "(any overlapping detection counts).",
        "",
        "| Category | TP | FP | FN | Precision | Recall | F1 |",
        "|---|---:|---:|---:|---:|---:|---:|",
    ]
    for category in sorted(result.per_category, key=lambda c: c.value):
        m = result.per_category[category]
        lines.append(_row(category.value, m))
    lines.append(_row("**overall (category-aware)**", result.overall))
    lines.append(_row("**binary (any PHI)**", result.binary))
    if result.unscored_gold_spans:
        lines += [
            "",
            f"{result.unscored_gold_spans} ground-truth annotations had no Safe "
            "Harbor category and were not scored.",
        ]
    return "\n".join(lines) + "\n"

openbtk.deid.labelled.LabelledDocument ¶

Bases: BaseModel

A document plus its ground-truth PHI annotations.

unscored_spans counts annotations in the source dataset that have no Safe Harbor PHICategory equivalent (e.g. i2b2's PROFESSION or AGE tags) and were therefore left out of spans -- carried as a count, not dropped silently, so a benchmark report can state exactly how much of the dataset's annotation it did not score.

Example

doc = LabelledDocument(document_id="d1", text="x", spans=[]) doc.unscored_spans 0

openbtk.deid.labelled.LabelledSpan ¶

Bases: BaseModel

One annotated PHI span: category plus half-open [start, end) offsets.

Example

LabelledSpan(category=PHICategory.SSN, start=4, end=15).end 15

openbtk.data.clinical_text.datasets.N2C2DeidDataset ¶

N2C2DeidDataset(*, path: str | None = None)

Bases: BaseDatasetAdapter

i2b2/n2c2 2014 de-identification corpus, from a user-supplied directory.

Parameters:

Name	Type	Description	Default
`path`	`str \| None`	Directory of `*.xml` files. May instead be passed to :meth:`load`. Never fetched or created by this class.	`None`

Example

import contextlib, io, tempfile from pathlib import Path xml = ( ... '' ... '' ... ) with tempfile.TemporaryDirectory() as d: ... _ = (Path(d) / "doc1.xml").write_text(xml) ... with contextlib.redirect_stdout(io.StringIO()): ... docs = list(N2C2DeidDataset(path=d).load()) docs[0].spans[0].category.value 'date'

Source code in src/openbtk/data/clinical_text/datasets.py

def __init__(self, *, path: str | None = None) -> None:
    self._path = path

load ¶

load(**kwargs: Any) -> Iterator[LabelledDocument]

Validate the source now, then stream documents lazily.

Raises:

Type	Description
`DatasetError`	If no directory was supplied or it does not exist (the message names the registration page -- this class never downloads), or, while iterating, if a file is malformed or its tag offsets do not match its text.

Source code in src/openbtk/data/clinical_text/datasets.py

def load(self, **kwargs: Any) -> Iterator[LabelledDocument]:
    """Validate the source now, then stream documents lazily.

    Raises:
        DatasetError: If no directory was supplied or it does not exist
            (the message names the registration page -- this class never
            downloads), or, while iterating, if a file is malformed or
            its tag offsets do not match its text.
    """
    path = kwargs.get("path", self._path)
    directory = Path(path) if path else None
    if directory is None or not directory.is_dir():
        raise DatasetError(
            "The n2c2 de-identification corpus is released only under a "
            "Data Use Agreement and is never downloaded automatically. "
            f"Register at {_REGISTRATION_URL}, obtain the data, and pass "
            "its directory as `path`.",
            context={"dataset": "n2c2_deid", "registration": _REGISTRATION_URL},
        )
    log.info("dataset.start", dataset="n2c2_deid", source=str(directory))
    return self._iter_documents(directory)

Retrieval¶

openbtk.eval.retrieval.evaluate_retrieval ¶

evaluate_retrieval(retrieve: Callable[[str], Sequence[str]], queries: Iterable[RetrievalQuery], *, ks: Sequence[int] = DEFAULT_KS) -> RetrievalReport

Score retrieve (query text -> ranked document ids) over queries.

queries is consumed once (streaming); memory does not grow with the number of queries.

Raises:

Type	Description
`ValueError`	If `ks` is empty or contains a value < 1, or `queries` is empty (an average over nothing is not a metric).

Example

qs = [RetrievalQuery(query_id="q", query="x", relevance={"a": 1.0})] report = evaluate_retrieval(lambda text: ["a", "b"], qs, ks=(1,)) report.mrr, report.recall_at_k[1] (1.0, 1.0)

Source code in src/openbtk/eval/retrieval.py

def evaluate_retrieval(
    retrieve: Callable[[str], Sequence[str]],
    queries: Iterable[RetrievalQuery],
    *,
    ks: Sequence[int] = DEFAULT_KS,
) -> RetrievalReport:
    """Score ``retrieve`` (query text -> ranked document ids) over ``queries``.

    ``queries`` is consumed once (streaming); memory does not grow with the
    number of queries.

    Raises:
        ValueError: If ``ks`` is empty or contains a value < 1, or ``queries``
            is empty (an average over nothing is not a metric).

    Example:
        >>> qs = [RetrievalQuery(query_id="q", query="x", relevance={"a": 1.0})]
        >>> report = evaluate_retrieval(lambda text: ["a", "b"], qs, ks=(1,))
        >>> report.mrr, report.recall_at_k[1]
        (1.0, 1.0)
    """
    if not ks:
        raise ValueError("ks must not be empty")
    for k in ks:
        _check_k(k)
    recall_sums = dict.fromkeys(ks, 0.0)
    ndcg_sums = dict.fromkeys(ks, 0.0)
    mrr_sum = 0.0
    n = 0
    for q in queries:
        ranked = list(retrieve(q.query))
        relevant = q.relevant_ids
        for k in ks:
            recall_sums[k] += recall_at_k(ranked, relevant, k)
            ndcg_sums[k] += ndcg_at_k(ranked, q.relevance, k)
        mrr_sum += reciprocal_rank(ranked, relevant)
        n += 1
    if n == 0:
        raise ValueError("no queries to evaluate")
    return RetrievalReport(
        n_queries=n,
        ks=list(ks),
        recall_at_k={k: recall_sums[k] / n for k in ks},
        ndcg_at_k={k: ndcg_sums[k] / n for k in ks},
        mrr=mrr_sum / n,
    )

openbtk.eval.retrieval.retriever_from ¶

retriever_from(embedding: BaseEmbeddingProvider, vectorstore: BaseVectorStore, *, top_k: int = 10, id_of: Callable[[SearchResult], str] | None = None) -> Callable[[str], list[str]]

Build a query -> ranked ids callable from a real embedding provider and vector store, so the same metrics score whatever retrieval stack a user actually runs. id_of defaults to the hit's own id; pass e.g. lambda r: r.source.record_id to judge at record rather than chunk level.

Source code in src/openbtk/eval/retrieval.py

def retriever_from(
    embedding: BaseEmbeddingProvider,
    vectorstore: BaseVectorStore,
    *,
    top_k: int = 10,
    id_of: Callable[[SearchResult], str] | None = None,
) -> Callable[[str], list[str]]:
    """Build a ``query -> ranked ids`` callable from a real embedding
    provider and vector store, so the same metrics score whatever retrieval
    stack a user actually runs. ``id_of`` defaults to the hit's own id; pass
    e.g. ``lambda r: r.source.record_id`` to judge at record rather than
    chunk level.
    """
    pick = id_of if id_of is not None else (lambda r: r.id)

    def retrieve(query: str) -> list[str]:
        hits = vectorstore.query(embedding.embed_one(query), top_k=top_k)
        return [pick(h) for h in hits]

    return retrieve

openbtk.eval.retrieval.RetrievalQuery ¶

Bases: BaseModel

One evaluation query with its graded relevance judgements.

Example

q = RetrievalQuery(query_id="q1", query="diabetes", relevance={"c3": 1.0}) q.relevant_ids ['c3']

openbtk.eval.retrieval.RetrievalReport ¶

Bases: BaseModel

Macro-averaged metrics over every evaluated query.

openbtk.eval.retrieval.recall_at_k ¶

recall_at_k(ranked_ids: Sequence[str], relevant: Iterable[str], k: int) -> float

Fraction of the relevant documents found in the top k.

Example

recall_at_k(["a", "b", "c"], {"a", "z"}, k=2) 0.5

Source code in src/openbtk/eval/retrieval.py

def recall_at_k(ranked_ids: Sequence[str], relevant: Iterable[str], k: int) -> float:
    """Fraction of the relevant documents found in the top ``k``.

    Example:
        >>> recall_at_k(["a", "b", "c"], {"a", "z"}, k=2)
        0.5
    """
    _check_k(k)
    relevant_set = set(relevant)
    if not relevant_set:
        raise ValueError("recall is undefined with no relevant documents")
    top_k = set(_dedupe(ranked_ids)[:k])
    return len(top_k & relevant_set) / len(relevant_set)

openbtk.eval.retrieval.reciprocal_rank ¶

reciprocal_rank(ranked_ids: Sequence[str], relevant: Iterable[str]) -> float

1 / rank of the first relevant result, or 0.0 if none was retrieved.

Example

reciprocal_rank(["x", "a", "b"], {"a"}) 0.5

Source code in src/openbtk/eval/retrieval.py

def reciprocal_rank(ranked_ids: Sequence[str], relevant: Iterable[str]) -> float:
    """1 / rank of the first relevant result, or 0.0 if none was retrieved.

    Example:
        >>> reciprocal_rank(["x", "a", "b"], {"a"})
        0.5
    """
    relevant_set = set(relevant)
    for rank, item in enumerate(_dedupe(ranked_ids), start=1):
        if item in relevant_set:
            return 1.0 / rank
    return 0.0

openbtk.eval.retrieval.ndcg_at_k ¶

ndcg_at_k(ranked_ids: Sequence[str], relevance: Mapping[str, float], k: int) -> float

Normalised discounted cumulative gain at k (linear gain).

Example

ndcg_at_k(["a", "b"], {"a": 1.0, "b": 1.0}, k=2) 1.0 round(ndcg_at_k(["x", "a"], {"a": 1.0}, k=2), 4) 0.6309

Source code in src/openbtk/eval/retrieval.py

def ndcg_at_k(
    ranked_ids: Sequence[str], relevance: Mapping[str, float], k: int
) -> float:
    """Normalised discounted cumulative gain at ``k`` (linear gain).

    Example:
        >>> ndcg_at_k(["a", "b"], {"a": 1.0, "b": 1.0}, k=2)
        1.0
        >>> round(ndcg_at_k(["x", "a"], {"a": 1.0}, k=2), 4)
        0.6309
    """
    _check_k(k)
    ideal_gains = sorted((g for g in relevance.values() if g > 0), reverse=True)[:k]
    idcg = sum(g / math.log2(i + 1) for i, g in enumerate(ideal_gains, start=1))
    if idcg == 0.0:
        raise ValueError("nDCG is undefined with no positively-judged documents")
    dcg = sum(
        max(relevance.get(item, 0.0), 0.0) / math.log2(i + 1)
        for i, item in enumerate(_dedupe(ranked_ids)[:k], start=1)
    )
    return dcg / idcg

Clinical QA¶

openbtk.eval.qa.evaluate_qa ¶

evaluate_qa(answerer: Callable[[str], str], items: Iterable[MCQItem], *, keep_results: bool = False) -> QAReport

Score answerer -- a function from a prompt to a reply -- on items.

items is consumed once and never materialised, so a full MedMCQA training split streams in constant memory. An exception from answerer (a provider outage, say) propagates: a QA score silently computed over the questions that happened to succeed would be a wrong number.

Parameters:

Name	Type	Description	Default
`answerer`	`Callable[[str], str]`	`prompt -> reply`. See :class:`LLMAnswerer`.	required
`items`	`Iterable[MCQItem]`	Questions to ask.	required
`keep_results`	`bool`	Also record each question's id, chosen letter and correctness (ids and letters only).	`False`

Source code in src/openbtk/eval/qa.py

def evaluate_qa(
    answerer: Callable[[str], str],
    items: Iterable[MCQItem],
    *,
    keep_results: bool = False,
) -> QAReport:
    """Score ``answerer`` -- a function from a prompt to a reply -- on ``items``.

    ``items`` is consumed once and never materialised, so a full MedMCQA
    training split streams in constant memory. An exception from ``answerer``
    (a provider outage, say) propagates: a QA score silently computed over the
    questions that happened to succeed would be a wrong number.

    Args:
        answerer: ``prompt -> reply``. See :class:`LLMAnswerer`.
        items: Questions to ask.
        keep_results: Also record each question's id, chosen letter and
            correctness (ids and letters only).
    """
    started_at = datetime.now(UTC)
    n = correct = unanswered = 0
    subjects: Counter[str] = Counter()
    subject_correct: Counter[str] = Counter()
    results: list[ItemResult] = []
    for item in items:
        n += 1
        predicted = parse_choice(answerer(format_prompt(item)), item.options)
        hit = predicted == item.answer
        correct += hit
        unanswered += predicted is None
        if item.subject is not None:
            subjects[item.subject] += 1
            subject_correct[item.subject] += hit
        if keep_results:
            results.append(
                ItemResult(item_id=item.item_id, predicted=predicted, correct=hit)
            )
    usage = getattr(answerer, "usage", None)
    return QAReport(
        n=n,
        correct=correct,
        unanswered=unanswered,
        per_subject={
            s: SubjectScore(n=c, correct=subject_correct[s])
            for s, c in subjects.items()
        },
        results=results,
        token_usage=usage if isinstance(usage, TokenUsage) else None,
        started_at=started_at,
        ended_at=datetime.now(UTC),
    )

openbtk.eval.qa.LLMAnswerer ¶

LLMAnswerer(provider: BaseLLMProvider, **generation: Any)

Adapt a BaseLLMProvider to evaluate_qa's prompt -> reply, accumulating its token usage across the run.

Parameters:

Name	Type	Description	Default
`provider`	`BaseLLMProvider`	The model under test.	required
`**generation`	`Any`	Passed to `provider.generate` on every call (provider-specific, e.g. `temperature=0`).	`{}`

Source code in src/openbtk/eval/qa.py

def __init__(self, provider: BaseLLMProvider, **generation: Any) -> None:
    self.provider = provider
    self._generation = generation
    self.usage: TokenUsage | None = None

openbtk.eval.qa.QAReport ¶

Bases: BaseModel

Scores for one QA run. Counts and ids only.

Unanswered questions (no parsable option) count as wrong and are also reported separately, so a model that rambles is visible rather than merely scoring low.

accuracy `property` ¶

accuracy: float

Correct / n (0.0 for an empty run).

accuracy_ci95 `property` ¶

accuracy_ci95: tuple[float, float]

95% Wilson score interval for accuracy: how much a score of this size can move by chance alone. (0.0, 1.0) for an empty run.

as_dict ¶

as_dict() -> dict[str, JsonValue]

A JSON-serialisable summary (no per-item results), counts only.

Source code in src/openbtk/eval/qa.py

def as_dict(self) -> dict[str, JsonValue]:
    """A JSON-serialisable summary (no per-item results), counts only."""
    lo, hi = self.accuracy_ci95
    return {
        "n": self.n,
        "correct": self.correct,
        "unanswered": self.unanswered,
        "accuracy": self.accuracy,
        "accuracy_ci95": [lo, hi],
        "per_subject": {
            k: {"n": v.n, "correct": v.correct, "accuracy": v.accuracy}
            for k, v in sorted(self.per_subject.items())
        },
    }

openbtk.eval.qa.MCQItem ¶

Bases: BaseModel

One multiple-choice question.

Example

item = MCQItem( ... item_id="q1", ... question="Which vitamin deficiency causes scurvy?", ... options={"A": "Vitamin A", "B": "Vitamin C"}, ... answer="B", ... ) item.options[item.answer] 'Vitamin C'

openbtk.eval.qa.read_medqa_jsonl ¶

read_medqa_jsonl(path: str | Path) -> Iterator[MCQItem]

Stream MedQA-format questions from a JSON Lines file.

Raises:

Type	Description
`DatasetError`	On unreadable or malformed input, or a question whose correct option cannot be determined. The message names the line, never the question.

Source code in src/openbtk/eval/qa.py

def read_medqa_jsonl(path: str | Path) -> Iterator[MCQItem]:
    """Stream MedQA-format questions from a JSON Lines file.

    Raises:
        DatasetError: On unreadable or malformed input, or a question whose
            correct option cannot be determined. The message names the line,
            never the question.
    """
    p = Path(path)
    for n, row in _lines(p):
        ctx = {"path": str(p), "line": n}
        options = row.get("options")
        if not isinstance(options, dict) or len(options) < 2:
            raise DatasetError(f"{p.name} line {n} has no options object.", context=ctx)
        options = {str(k): str(v) for k, v in options.items()}
        key = row.get("answer_idx")
        if key is None:
            matches = [k for k, v in options.items() if v == row.get("answer")]
            key = matches[0] if len(matches) == 1 else None
        if key is None or str(key) not in options:
            raise DatasetError(
                f"{p.name} line {n}: cannot determine the correct option.",
                context=ctx,
            )
        try:
            yield MCQItem(
                item_id=str(row.get("id", f"{p.stem}:{n}")),
                question=str(row.get("question", "")),
                options=options,
                answer=str(key),
                subject=row.get("meta_info"),
            )
        except ValueError as e:
            raise DatasetError(
                f"{p.name} line {n} is not a valid item.", context=ctx
            ) from e

openbtk.eval.qa.read_medmcqa_jsonl ¶

read_medmcqa_jsonl(path: str | Path) -> Iterator[MCQItem]

Stream MedMCQA-format questions from a JSON Lines file.

Raises:

Type	Description
`DatasetError`	On unreadable or malformed input, a hidden answer key, or an ambiguous `cop` value (see the module docstring).

Source code in src/openbtk/eval/qa.py

def read_medmcqa_jsonl(path: str | Path) -> Iterator[MCQItem]:
    """Stream MedMCQA-format questions from a JSON Lines file.

    Raises:
        DatasetError: On unreadable or malformed input, a hidden answer key, or
            an ambiguous ``cop`` value (see the module docstring).
    """
    p = Path(path)
    for n, row in _lines(p):
        missing = [k for k in _MEDMCQA_KEYS if k not in row]
        if missing:
            raise DatasetError(
                f"{p.name} line {n} lacks option fields {missing}.",
                context={"path": str(p), "line": n},
            )
        answer = _medmcqa_answer(row.get("cop"), n, p)
        try:
            yield MCQItem(
                item_id=str(row.get("id", f"{p.stem}:{n}")),
                question=str(row.get("question", "")),
                options={"ABCD"[i]: str(row[k]) for i, k in enumerate(_MEDMCQA_KEYS)},
                answer=answer,
                subject=row.get("subject_name"),
            )
        except ValueError as e:
            raise DatasetError(
                f"{p.name} line {n} is not a valid item.",
                context={"path": str(p), "line": n},
            ) from e

openbtk.eval.qa.parse_choice ¶

parse_choice(text: str, letters: Iterable[str]) -> str | None

Extract the chosen option letter from a model's free-text reply.

Recognises "B", "(B)", "B. text", "B) text" and "...the answer is B" / "Answer: B". It does not accept a bare letter buried in prose ("A patient with ..." starts with the article "A"), so an evasive or rambling reply is counted as unanswered instead of being credited by accident. Returns None when no option can be identified, or when the letter is not one of letters.

Example

parse_choice("The answer is C.", "ABCD") 'C' parse_choice("(b) Vitamin C", "ABCD") 'B' parse_choice("A patient presents with fever.", "ABCD") is None True

Source code in src/openbtk/eval/qa.py

def parse_choice(text: str, letters: Iterable[str]) -> str | None:
    """Extract the chosen option letter from a model's free-text reply.

    Recognises ``"B"``, ``"(B)"``, ``"B. text"``, ``"B) text"`` and
    ``"...the answer is B"`` / ``"Answer: B"``. It does *not* accept a bare
    letter buried in prose (``"A patient with ..."`` starts with the article
    "A"), so an evasive or rambling reply is counted as unanswered instead of
    being credited by accident. Returns ``None`` when no option can be
    identified, or when the letter is not one of ``letters``.

    Example:
        >>> parse_choice("The answer is C.", "ABCD")
        'C'
        >>> parse_choice("(b) Vitamin C", "ABCD")
        'B'
        >>> parse_choice("A patient presents with fever.", "ABCD") is None
        True
    """
    valid = {x.upper() for x in letters}
    stripped = text.strip()
    for pattern in (_ANSWER_IS, _LEADING):
        m = pattern.search(stripped)
        if m:
            letter = next(g for g in m.groups() if g).upper()
            return letter if letter in valid else None
    return None

openbtk.eval.qa.format_prompt ¶

format_prompt(item: MCQItem) -> str

The zero-shot prompt this module sends for item.

Example

item = MCQItem( ... item_id="q", question="Q?", options={"A": "x", "B": "y"}, answer="A" ... ) prompt = format_prompt(item) "A. x" in prompt and prompt.endswith("Answer:") True

Source code in src/openbtk/eval/qa.py

def format_prompt(item: MCQItem) -> str:
    """The zero-shot prompt this module sends for ``item``.

    Example:
        >>> item = MCQItem(
        ...     item_id="q", question="Q?", options={"A": "x", "B": "y"}, answer="A"
        ... )
        >>> prompt = format_prompt(item)
        >>> "A. x" in prompt and prompt.endswith("Answer:")
        True
    """
    options = "\n".join(f"{k}. {v}" for k, v in item.options.items())
    return _PROMPT.format(question=item.question, options=options)

openbtk.eval.qa.qa_manifest ¶

qa_manifest(report: QAReport, *, component: ComponentProvenance | None = None, source: str | Path | None = None) -> EvalManifest

The provenance record for a QA run (FR-X-06): which model, which file (by SHA-256), when, and the counts -- no question or answer text.

Source code in src/openbtk/eval/qa.py

def qa_manifest(
    report: QAReport,
    *,
    component: ComponentProvenance | None = None,
    source: str | Path | None = None,
) -> EvalManifest:
    """The provenance record for a QA run (FR-X-06): which model, which file
    (by SHA-256), when, and the counts -- no question or answer text."""
    digests = [file_digest(source, record_count=report.n)] if source is not None else []
    return build_manifest(
        "qa",
        report.as_dict(),
        started_at=report.started_at,
        ended_at=report.ended_at,
        component=component,
        input_digests=digests,
    )

Groundedness¶

openbtk.eval.groundedness.score_groundedness ¶

score_groundedness(examples: Iterable[GroundedExample], guardrail: GroundednessGuardrail | None = None) -> GroundednessReport

Score how faithful a set of generated answers is to their contexts.

examples is consumed once. guardrail defaults to the word-overlap heuristic (see the module docstring for what that can and cannot see).

Example

ok = GroundedExample( ... example_id="a", ... answer="The patient has type 2 diabetes.", ... context=["Assessment: type 2 diabetes mellitus, stable."], ... ) bad = GroundedExample( ... example_id="b", ... answer="The patient has a fractured femur.", ... context=["Assessment: type 2 diabetes mellitus, stable."], ... ) report = score_groundedness([ok, bad]) report.faithfulness, report.fully_grounded_examples (0.5, 1)

Source code in src/openbtk/eval/groundedness.py

def score_groundedness(
    examples: Iterable[GroundedExample],
    guardrail: GroundednessGuardrail | None = None,
) -> GroundednessReport:
    """Score how faithful a set of generated answers is to their contexts.

    ``examples`` is consumed once. ``guardrail`` defaults to the word-overlap
    heuristic (see the module docstring for what that can and cannot see).

    Example:
        >>> ok = GroundedExample(
        ...     example_id="a",
        ...     answer="The patient has type 2 diabetes.",
        ...     context=["Assessment: type 2 diabetes mellitus, stable."],
        ... )
        >>> bad = GroundedExample(
        ...     example_id="b",
        ...     answer="The patient has a fractured femur.",
        ...     context=["Assessment: type 2 diabetes mellitus, stable."],
        ... )
        >>> report = score_groundedness([ok, bad])
        >>> report.faithfulness, report.fully_grounded_examples
        (0.5, 1)
    """
    checker = guardrail or GroundednessGuardrail()
    started_at = datetime.now(UTC)
    n = claims = unsupported = grounded = 0
    for ex in examples:
        result = checker.check(
            GroundednessCheckInput(answer=ex.answer, context=ex.context)
        )
        n += 1
        count = result.details.get("claim_count")
        if not isinstance(count, int) or isinstance(count, bool):
            raise ProcessingError(
                "The groundedness guardrail did not report details['claim_count'], "
                "so faithfulness cannot be computed. A custom guardrail must "
                "include it (the built-in GroundednessGuardrail does).",
                context={"example_id": ex.example_id},
            )
        claims += count
        bad = len(result.spans)
        unsupported += bad
        grounded += bad == 0
    return GroundednessReport(
        n_examples=n,
        n_claims=claims,
        unsupported_claims=unsupported,
        fully_grounded_examples=grounded,
        started_at=started_at,
        ended_at=datetime.now(UTC),
    )

openbtk.eval.groundedness.evaluate_detector ¶

evaluate_detector(examples: Iterable[LabelledExample], guardrail: GroundednessGuardrail | None = None) -> DetectorReport

Measure how well guardrail separates grounded from ungrounded answers against human labels.

Example

good = LabelledExample( ... example_id="a", grounded=True, ... answer="The patient has type 2 diabetes.", ... context=["Assessment: type 2 diabetes mellitus, stable."], ... ) bad = LabelledExample( ... example_id="b", grounded=False, ... answer="The patient has a fractured femur.", ... context=["Assessment: type 2 diabetes mellitus, stable."], ... ) evaluate_detector([good, bad]).f1 1.0

Source code in src/openbtk/eval/groundedness.py

def evaluate_detector(
    examples: Iterable[LabelledExample],
    guardrail: GroundednessGuardrail | None = None,
) -> DetectorReport:
    """Measure how well ``guardrail`` separates grounded from ungrounded answers
    against human labels.

    Example:
        >>> good = LabelledExample(
        ...     example_id="a", grounded=True,
        ...     answer="The patient has type 2 diabetes.",
        ...     context=["Assessment: type 2 diabetes mellitus, stable."],
        ... )
        >>> bad = LabelledExample(
        ...     example_id="b", grounded=False,
        ...     answer="The patient has a fractured femur.",
        ...     context=["Assessment: type 2 diabetes mellitus, stable."],
        ... )
        >>> evaluate_detector([good, bad]).f1
        1.0
    """
    checker = guardrail or GroundednessGuardrail()
    tp = fp = fn = tn = 0
    for ex in examples:
        flagged = not checker.check(
            GroundednessCheckInput(answer=ex.answer, context=ex.context)
        ).passed
        if flagged and not ex.grounded:
            tp += 1
        elif flagged:
            fp += 1
        elif not ex.grounded:
            fn += 1
        else:
            tn += 1
    return DetectorReport(
        true_positives=tp, false_positives=fp, false_negatives=fn, true_negatives=tn
    )

openbtk.eval.groundedness.GroundedExample ¶

Bases: BaseModel

An answer and the source text it should be grounded in.

openbtk.eval.groundedness.LabelledExample ¶

Bases: GroundedExample

A :class:GroundedExample a human judged: grounded is True when every claim in answer is supported by context.

openbtk.eval.groundedness.GroundednessReport ¶

Bases: BaseModel

Claim-level faithfulness over a set of examples. Counts only.

faithfulness `property` ¶

faithfulness: float

Supported claims / all claims. 1.0 when there were no claims at all (nothing to get wrong); consult n_claims to tell that from perfect.

grounded_rate `property` ¶

grounded_rate: float

Fraction of examples whose every claim was supported.

openbtk.eval.groundedness.DetectorReport ¶

Bases: BaseModel

Agreement between the checker and human labels, for the task of detecting ungrounded answers (the positive class is "not grounded").

openbtk.eval.groundedness.groundedness_manifest ¶

groundedness_manifest(report: GroundednessReport) -> EvalManifest

The provenance record for a groundedness run (FR-X-06): when, and the counts -- no answer or context text.

Source code in src/openbtk/eval/groundedness.py

def groundedness_manifest(report: GroundednessReport) -> EvalManifest:
    """The provenance record for a groundedness run (FR-X-06): when, and the
    counts -- no answer or context text."""
    return build_manifest(
        "groundedness",
        report.as_dict(),
        started_at=report.started_at,
        ended_at=report.ended_at,
    )

Summaries: ROUGE and BERTScore¶

openbtk.eval.summarisation.evaluate_summaries ¶

evaluate_summaries(pairs: Iterable[SummaryPair], *, rouge_types: Sequence[str] = ROUGE_TYPES, use_stemmer: bool = True, bertscorer: BertScorer | None = None) -> SummarisationReport

Score candidate summaries against references.

Streams the pairs, holding only the running sums and one small record of scores per example. The report's means are macro averages: every example counts once, however long its summary.

Parameters:

Name	Type	Description	Default
`pairs`	`Iterable[SummaryPair]`	The candidate and reference of each example.	required
`rouge_types`	`Sequence[str]`	ROUGE variants to compute (`()` skips ROUGE).	`ROUGE_TYPES`
`use_stemmer`	`bool`	Stem before ROUGE matching.	`True`
`bertscorer`	`BertScorer \| None`	Also compute BERTScore with this scorer (`None` skips it).	`None`

Example

report = evaluate_summaries( ... [SummaryPair(example_id="a", reference="stable", candidate="stable")] ... ) report.rouge["rouge1"].f1, report.n (1.0, 1)

Source code in src/openbtk/eval/summarisation.py

def evaluate_summaries(
    pairs: Iterable[SummaryPair],
    *,
    rouge_types: Sequence[str] = ROUGE_TYPES,
    use_stemmer: bool = True,
    bertscorer: BertScorer | None = None,
) -> SummarisationReport:
    """Score candidate summaries against references.

    Streams the pairs, holding only the running sums and one small record of scores per
    example. The report's means are macro averages: every example counts once, however
    long its summary.

    Args:
        pairs: The candidate and reference of each example.
        rouge_types: ROUGE variants to compute (``()`` skips ROUGE).
        use_stemmer: Stem before ROUGE matching.
        bertscorer: Also compute BERTScore with this scorer (``None`` skips it).

    Example:
        >>> report = evaluate_summaries(
        ...     [SummaryPair(example_id="a", reference="stable", candidate="stable")]
        ... )
        >>> report.rouge["rouge1"].f1, report.n
        (1.0, 1)
    """
    started = datetime.now(UTC)
    rouge_sums: dict[str, list[float]] = {t: [0.0, 0.0, 0.0] for t in rouge_types}
    bert_sums = [0.0, 0.0, 0.0]
    examples: list[ExampleScores] = []
    for pair in pairs:
        rouge = rouge_scores(
            pair.reference, pair.candidate, types=rouge_types, use_stemmer=use_stemmer
        )
        for name, prf in rouge.items():
            rouge_sums[name][0] += prf.precision
            rouge_sums[name][1] += prf.recall
            rouge_sums[name][2] += prf.f1
        bert = None
        if bertscorer is not None:
            bert = bertscorer.score(pair.reference, pair.candidate)
            bert_sums[0] += bert.precision
            bert_sums[1] += bert.recall
            bert_sums[2] += bert.f1
        examples.append(
            ExampleScores(example_id=pair.example_id, rouge=rouge, bertscore=bert)
        )
    n = len(examples)

    def mean(total: list[float]) -> Prf:
        return Prf(
            precision=total[0] / n if n else 0.0,
            recall=total[1] / n if n else 0.0,
            f1=total[2] / n if n else 0.0,
        )

    return SummarisationReport(
        n=n,
        rouge={name: mean(total) for name, total in rouge_sums.items()} if n else {},
        bertscore=mean(bert_sums) if bertscorer is not None and n else None,
        examples=examples,
        started_at=started,
        ended_at=datetime.now(UTC),
    )

openbtk.eval.summarisation.rouge_scores ¶

rouge_scores(reference: str, candidate: str, *, types: Sequence[str] = ROUGE_TYPES, use_stemmer: bool = True) -> dict[str, Prf]

ROUGE precision, recall and F1 of one candidate against one reference.

Parameters:

Name	Type	Description	Default
`reference`	`str`	The reference summary.	required
`candidate`	`str`	The candidate summary.	required
`types`	`Sequence[str]`	Which variants: `rouge1`, `rouge2`, `rougeL`.	`ROUGE_TYPES`
`use_stemmer`	`bool`	Porter-stem words first (the usual reporting convention).	`True`

Raises:

Type	Description
`ConfigError`	On an unknown variant.
`MissingDependencyError`	If `rouge-score` is not installed (`eval` extra).

Example

scores = rouge_scores("the patient is stable", "the patient is stable") scores["rouge1"].f1 1.0

Source code in src/openbtk/eval/summarisation.py

def rouge_scores(
    reference: str,
    candidate: str,
    *,
    types: Sequence[str] = ROUGE_TYPES,
    use_stemmer: bool = True,
) -> dict[str, Prf]:
    """ROUGE precision, recall and F1 of one candidate against one reference.

    Args:
        reference: The reference summary.
        candidate: The candidate summary.
        types: Which variants: ``rouge1``, ``rouge2``, ``rougeL``.
        use_stemmer: Porter-stem words first (the usual reporting convention).

    Raises:
        ConfigError: On an unknown variant.
        MissingDependencyError: If ``rouge-score`` is not installed (``eval`` extra).

    Example:
        >>> scores = rouge_scores("the patient is stable", "the patient is stable")
        >>> scores["rouge1"].f1
        1.0
    """
    unknown = [t for t in types if t not in ROUGE_TYPES]
    if unknown:
        raise ConfigError(
            f"Unknown ROUGE variant(s) {unknown}; choose from {list(ROUGE_TYPES)}.",
            context={"variants": unknown},
        )
    if not types:
        return {}
    rouge_scorer = require("rouge_score.rouge_scorer", extra="eval")
    scorer = rouge_scorer.RougeScorer(list(types), use_stemmer=use_stemmer)
    raw = scorer.score(reference, candidate)  # (target, prediction)
    return {
        name: Prf(
            precision=float(raw[name].precision),
            recall=float(raw[name].recall),
            f1=float(raw[name].fmeasure),
        )
        for name in types
    }

openbtk.eval.summarisation.BertScorer ¶

BertScorer(*, model: str, revision: str, layer: int, device: str = 'cpu', max_length: int = 512)

BERTScore with a model, revision and layer you choose. See the module docstring.

Parameters:

Name	Type	Description	Default
`model`	`str`	A Hugging Face Hub encoder id (or local path).	required
`revision`	`str`	The pinned commit of `model`. Required: an unpinned download follows a moving branch.	required
`layer`	`int`	Which hidden layer's token embeddings to compare, `1` to the model's number of layers (`0` would be the input embeddings, which carry no context). The BERTScore paper tuned one per model; choose and record yours.	required
`device`	`str`	A torch device string.	`'cpu'`
`max_length`	`int`	Tokens kept per text; a longer text is truncated, and the score covers only what was kept.	`512`

Example

scorer = BertScorer(model="org/encoder", revision="abc1234", layer=8) scorer.provenance().model_identity.revision # nothing is loaded yet 'abc1234'

Source code in src/openbtk/eval/summarisation.py

def __init__(
    self,
    *,
    model: str,
    revision: str,
    layer: int,
    device: str = "cpu",
    max_length: int = 512,
) -> None:
    if not revision:
        raise ConfigError(
            "BERTScore needs a pinned model revision.",
            context={"component": "eval.bertscore"},
        )
    if layer < 1 or max_length < 3:
        raise ConfigError(
            "layer must be at least 1 and max_length at least 3.",
            context={"component": "eval.bertscore"},
        )
    self._model_name = model
    self._revision = revision
    self._layer = layer
    self._device = device
    self._max_length = max_length
    self._loaded: tuple[Any, Any, Any] | None = None

provenance ¶

provenance() -> ComponentProvenance

The model, revision and layer, for a manifest.

Source code in src/openbtk/eval/summarisation.py

def provenance(self) -> ComponentProvenance:
    """The model, revision and layer, for a manifest."""
    return ComponentProvenance(
        registry_key="eval.bertscore",
        class_name=type(self).__name__,
        package_version=_openbtk_version(),
        config={
            "model": self._model_name,
            "layer": self._layer,
            "device": self._device,
            "max_length": self._max_length,
        },
        model_identity=ModelIdentity(
            name=self._model_name, revision=self._revision, source="huggingface"
        ),
    )

score ¶

score(reference: str, candidate: str) -> Prf

Raw BERTScore precision, recall and F1 of candidate vs reference.

As in the reference implementation, the special tokens ([CLS], [SEP]) carry zero weight in the averages but still take part in the matching, so a token can be matched to the other text's special token. An empty (or all-special-token) text scores 0 on all three.

Raises:

Type	Description
`ProcessingError`	If the model fails. The message never contains the text.

Source code in src/openbtk/eval/summarisation.py

def score(self, reference: str, candidate: str) -> Prf:
    """Raw BERTScore precision, recall and F1 of ``candidate`` vs ``reference``.

    As in the reference implementation, the special tokens (``[CLS]``, ``[SEP]``)
    carry zero weight in the averages but still take part in the matching, so a
    token can be matched to the other text's special token. An empty (or
    all-special-token) text scores ``0`` on all three.

    Raises:
        ProcessingError: If the model fails. The message never contains the text.
    """
    try:
        ref, ref_real = self._embed(reference)
        hyp, hyp_real = self._embed(candidate)
        if not bool(ref_real.any()) or not bool(hyp_real.any()):
            return Prf(precision=0.0, recall=0.0, f1=0.0)
        similarity = hyp @ ref.T  # (candidate tokens, reference tokens)
        precision = float(similarity.max(dim=1).values[hyp_real].mean())
        recall = float(similarity.max(dim=0).values[ref_real].mean())
    except (ConfigError, ProcessingError):
        raise
    except Exception as e:  # torch / transformers raise their own hierarchy
        raise ProcessingError(
            "BERTScore failed.",
            context={"component": "eval.bertscore", "model": self._model_name},
        ) from e
    denominator = precision + recall
    f1 = 2 * precision * recall / denominator if denominator else 0.0
    return Prf(precision=precision, recall=recall, f1=f1)

openbtk.eval.summarisation.SummaryPair ¶

Bases: BaseModel

One candidate summary and the reference it is compared with.

Example

SummaryPair( ... example_id="d1", reference="Stable on discharge.", candidate="Stable." ... ).example_id 'd1'

openbtk.eval.summarisation.SummarisationReport ¶

Bases: BaseModel

Mean scores over the examples (a macro average) and each example's own.

Example

report = evaluate_summaries([], rouge_types=()) report.n 0

as_dict ¶

as_dict() -> dict[str, JsonValue]

A flat {name: number} view (rouge1_f1, bertscore_f1, n, ...), the shape EvalManifest.report and ModelCard.with_evaluation read.

Source code in src/openbtk/eval/summarisation.py

def as_dict(self) -> dict[str, JsonValue]:
    """A flat ``{name: number}`` view (``rouge1_f1``, ``bertscore_f1``, ``n``, ...),
    the shape ``EvalManifest.report`` and ``ModelCard.with_evaluation`` read."""
    flat: dict[str, JsonValue] = {"n": self.n}
    for name, prf in self.rouge.items():
        flat[f"{name}_precision"] = prf.precision
        flat[f"{name}_recall"] = prf.recall
        flat[f"{name}_f1"] = prf.f1
    if self.bertscore is not None:
        flat["bertscore_precision"] = self.bertscore.precision
        flat["bertscore_recall"] = self.bertscore.recall
        flat["bertscore_f1"] = self.bertscore.f1
    return flat

openbtk.eval.summarisation.summarisation_manifest ¶

summarisation_manifest(report: SummarisationReport, *, bertscorer: BertScorer | None = None, input_digests: Sequence[DataDigest] = ()) -> EvalManifest

The provenance record for a summarisation run: when, the mean scores and counts, the BERTScore model and revision if one was used, and the digests of the input files. Never the summaries.

Source code in src/openbtk/eval/summarisation.py

def summarisation_manifest(
    report: SummarisationReport,
    *,
    bertscorer: BertScorer | None = None,
    input_digests: Sequence[DataDigest] = (),
) -> EvalManifest:
    """The provenance record for a summarisation run: when, the mean scores and counts,
    the BERTScore model and revision if one was used, and the digests of the input
    files. Never the summaries."""
    return build_manifest(
        "summarisation",
        report.as_dict(),
        started_at=report.started_at,
        ended_at=report.ended_at,
        component=bertscorer.provenance() if bertscorer is not None else None,
        input_digests=input_digests,
    )

Model cards¶

openbtk.eval.model_card.ModelCard ¶

Bases: BaseModel

A model card. Build one with :meth:for_component or :func:cards_from_run.

Example

from openbtk.core.provenance import ModelIdentity card = ModelCard( ... model=ModelIdentity( ... name="org/model", revision="abc1234", source="huggingface" ... ), ... intended_use="Ranking passages for clinical question answering.", ... ) "## Intended use" in card.to_markdown() True "Not provided." in card.to_markdown() # the sections you did not write True

for_component `classmethod` ¶

for_component(component: Component | ComponentProvenance, *, description: str | None = None, intended_use: str | None = None, out_of_scope_use: str | None = None, training_data: str | None = None, limitations: str | None = None, ethical_considerations: str | None = None, contact: str | None = None) -> ModelCard

A card for the model a component wraps.

Parameters:

Name	Type	Description	Default
`component`	`Component \| ComponentProvenance`	A component instance or its provenance. It must carry a model identity (an LLM, an embedding provider, a cross-encoder reranker, ...).	required
`description`	`str \| None`	One or two sentences on what the model is.	`None`
`intended_use`	`str \| None`	What it is meant for.	`None`
`out_of_scope_use`	`str \| None`	What it must not be used for.	`None`
`training_data`	`str \| None`	What it was trained on (for a fine-tune, what you trained it on). Openbtk cannot know this and does not guess.	`None`
`limitations`	`str \| None`	Known failure modes and blind spots.	`None`
`ethical_considerations`	`str \| None`	Risks, populations affected, mitigations.	`None`
`contact`	`str \| None`	Who to ask.	`None`

Raises:

Type	Description
`ConfigError`	If the component has no model identity (it wraps no model).

Source code in src/openbtk/eval/model_card.py

@classmethod
def for_component(
    cls,
    component: Component | ComponentProvenance,
    *,
    description: str | None = None,
    intended_use: str | None = None,
    out_of_scope_use: str | None = None,
    training_data: str | None = None,
    limitations: str | None = None,
    ethical_considerations: str | None = None,
    contact: str | None = None,
) -> ModelCard:
    """A card for the model a component wraps.

    Args:
        component: A component instance or its provenance. It must carry a model
            identity (an LLM, an embedding provider, a cross-encoder reranker, ...).
        description: One or two sentences on what the model is.
        intended_use: What it is meant for.
        out_of_scope_use: What it must not be used for.
        training_data: What it was trained on (for a fine-tune, what *you* trained
            it on). Openbtk cannot know this and does not guess.
        limitations: Known failure modes and blind spots.
        ethical_considerations: Risks, populations affected, mitigations.
        contact: Who to ask.

    Raises:
        ConfigError: If the component has no model identity (it wraps no model).
    """
    provenance = (
        component
        if isinstance(component, ComponentProvenance)
        else component.provenance()
    )
    if provenance.model_identity is None:
        raise ConfigError(
            f"{provenance.registry_key or provenance.class_name} wraps no model, "
            "so there is no model card to write.",
            context={"component": provenance.registry_key},
        )
    return cls(
        model=provenance.model_identity,
        component_key=provenance.registry_key or None,
        openbtk_version=provenance.package_version,
        settings=provenance.config,
        description=description,
        intended_use=intended_use,
        out_of_scope_use=out_of_scope_use,
        training_data=training_data,
        limitations=limitations,
        ethical_considerations=ethical_considerations,
        contact=contact,
    )

to_markdown ¶

to_markdown() -> str

The card as Markdown. Sections you did not provide say so.

Source code in src/openbtk/eval/model_card.py

def to_markdown(self) -> str:
    """The card as Markdown. Sections you did not provide say so."""

    def section(title: str, body: str | None) -> str:
        return f"## {title}\n\n{body or _NOT_PROVIDED}\n"

    identity = [
        f"- **Model:** `{self.model.name}`",
        f"- **Revision:** `{self.model.revision}`",
        f"- **Source:** {self.model.source}",
    ]
    if self.component_key:
        identity.append(f"- **Openbtk component:** `{self.component_key}`")
    if self.openbtk_version:
        identity.append(f"- **Openbtk version:** {self.openbtk_version}")
    parts = [
        f"# Model card: {self.model.name}\n",
        "## Model details\n\n" + "\n".join(identity) + "\n",
        section("Description", self.description),
        section("Intended use", self.intended_use),
        section("Out-of-scope use", self.out_of_scope_use),
        section("Training data", self.training_data),
        self._evaluation_section(),
        section("Limitations", self.limitations),
        section("Ethical considerations", self.ethical_considerations),
        section("Contact", self.contact),
    ]
    if self.settings:
        rows = "\n".join(
            f"- `{k}`: `{v}`" for k, v in sorted(self.settings.items())
        )
        parts.append(f"## Configuration\n\n{rows}\n")
    parts.append(f"---\nGenerated {self.generated_at.date().isoformat()}.\n")
    return "\n".join(parts)

with_evaluation ¶

with_evaluation(manifest: EvalManifest, metric: str, *, name: str | None = None) -> ModelCard

A copy of this card with one score from an evaluation run attached.

Parameters:

Name	Type	Description	Default
`manifest`	`EvalManifest`	The `EvalManifest` the evaluation produced.	required
`metric`	`str`	The key of the score in `manifest.report`, e.g. `"f1"`.	required
`name`	`str \| None`	How to label it on the card (default: `metric`).	`None`

Raises:

Type	Description
`ConfigError`	If the report has no numeric value under `metric`. A card never records a number the harness did not produce.

Source code in src/openbtk/eval/model_card.py

def with_evaluation(
    self, manifest: EvalManifest, metric: str, *, name: str | None = None
) -> ModelCard:
    """A copy of this card with one score from an evaluation run attached.

    Args:
        manifest: The ``EvalManifest`` the evaluation produced.
        metric: The key of the score in ``manifest.report``, e.g. ``"f1"``.
        name: How to label it on the card (default: ``metric``).

    Raises:
        ConfigError: If the report has no numeric value under ``metric``. A card
            never records a number the harness did not produce.
    """
    value = manifest.report.get(metric)
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ConfigError(
            f"The evaluation report has no numeric {metric!r} to put on the card.",
            context={"eval_id": manifest.eval_id, "metric": metric},
        )
    entry = CardEvaluation(
        metric=name or metric,
        value=float(value),
        eval_id=manifest.eval_id,
        kind=manifest.kind,
        inputs=[
            f"{d.uri} (sha256 {d.sha256})" if d.sha256 else f"{d.uri} (not hashed)"
            for d in manifest.input_digests
        ],
    )
    return self.model_copy(update={"evaluations": [*self.evaluations, entry]})

write ¶

write(path: str | Path) -> Path

Write the Markdown card to path (creating parent directories).

Source code in src/openbtk/eval/model_card.py

def write(self, path: str | Path) -> Path:
    """Write the Markdown card to ``path`` (creating parent directories)."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(self.to_markdown(), encoding="utf-8")
    return target

openbtk.eval.model_card.CardEvaluation ¶

Bases: BaseModel

One metric on a card, with the evaluation run it came from.

openbtk.eval.model_card.cards_from_run ¶

cards_from_run(manifest: RunManifest, **text: str | None) -> list[ModelCard]

One card per distinct model a run used, read from its manifest.

Every step whose component recorded a model identity contributes a card; a model used by several steps appears once. Any keyword (intended_use=..., limitations=...) is applied to every card, so pass only what is true of all of them, and write the rest with :meth:ModelCard.for_component.

Source code in src/openbtk/eval/model_card.py

def cards_from_run(manifest: RunManifest, **text: str | None) -> list[ModelCard]:
    """One card per distinct model a run used, read from its manifest.

    Every step whose component recorded a model identity contributes a card; a model
    used by several steps appears once. Any keyword (``intended_use=...``,
    ``limitations=...``) is applied to every card, so pass only what is true of all of
    them, and write the rest with :meth:`ModelCard.for_component`.

    """
    cards: dict[tuple[str, str], ModelCard] = {}
    for step in manifest.steps:
        identity = step.component.model_identity
        if identity is None:
            continue
        key = (identity.name, identity.revision)
        if key not in cards:
            cards[key] = ModelCard.for_component(step.component, **text)
    return list(cards.values())

Manifests¶

openbtk.eval.manifest.EvalManifest ¶

Bases: BaseModel

The audit record of one evaluation run.

Example

m = EvalManifest( ... eval_id="e1", kind="qa", started_at=datetime(2026, 1, 1, tzinfo=UTC), ... ended_at=datetime(2026, 1, 1, tzinfo=UTC), report={"n": 3}, ... ) m.manifest_version '1.0.0'

openbtk.eval.manifest.build_manifest ¶

build_manifest(kind: str, report: dict[str, JsonValue], *, started_at: datetime, ended_at: datetime | None = None, component: ComponentProvenance | None = None, input_digests: Sequence[DataDigest] = ()) -> EvalManifest

Assemble a manifest with a fresh id. ended_at defaults to now.

Source code in src/openbtk/eval/manifest.py

def build_manifest(
    kind: str,
    report: dict[str, JsonValue],
    *,
    started_at: datetime,
    ended_at: datetime | None = None,
    component: ComponentProvenance | None = None,
    input_digests: Sequence[DataDigest] = (),
) -> EvalManifest:
    """Assemble a manifest with a fresh id. ``ended_at`` defaults to now."""
    return EvalManifest(
        eval_id=uuid.uuid4().hex,
        kind=kind,
        started_at=started_at,
        ended_at=ended_at or datetime.now(UTC),
        component=component,
        input_digests=list(input_digests),
        report=report,
    )

openbtk.eval.manifest.file_digest ¶

file_digest(path: str | Path, *, record_count: int) -> DataDigest

A DataDigest for a local file, hashed in 1 MiB chunks (so an evaluation set is never held in memory just to be fingerprinted).

Example

import tempfile, pathlib with tempfile.TemporaryDirectory() as d: ... p = pathlib.Path(d) / "x.jsonl" ... _ = p.write_text("{}") ... digest = file_digest(p, record_count=1) len(digest.sha256 or "") 64

Source code in src/openbtk/eval/manifest.py

def file_digest(path: str | Path, *, record_count: int) -> DataDigest:
    """A ``DataDigest`` for a local file, hashed in 1 MiB chunks (so an
    evaluation set is never held in memory just to be fingerprinted).

    Example:
        >>> import tempfile, pathlib
        >>> with tempfile.TemporaryDirectory() as d:
        ...     p = pathlib.Path(d) / "x.jsonl"
        ...     _ = p.write_text("{}")
        ...     digest = file_digest(p, record_count=1)
        >>> len(digest.sha256 or "")
        64
    """
    h = hashlib.sha256()
    with Path(path).open("rb") as f:
        while chunk := f.read(_HASH_CHUNK):
            h.update(chunk)
    return DataDigest(uri=str(path), sha256=h.hexdigest(), record_count=record_count)

Evaluation¶

De-identification¶

openbtk.eval.deid.evaluate_deid ¶

openbtk.eval.deid.EvaluationResult ¶

overall property ¶

openbtk.eval.deid.CategoryMetrics ¶

openbtk.eval.deid.report_dict ¶

openbtk.eval.deid.format_markdown ¶

openbtk.deid.labelled.LabelledDocument ¶

openbtk.deid.labelled.LabelledSpan ¶

openbtk.data.clinical_text.datasets.N2C2DeidDataset ¶

load ¶

Retrieval¶

openbtk.eval.retrieval.evaluate_retrieval ¶

openbtk.eval.retrieval.retriever_from ¶

openbtk.eval.retrieval.RetrievalQuery ¶

openbtk.eval.retrieval.RetrievalReport ¶

openbtk.eval.retrieval.recall_at_k ¶

openbtk.eval.retrieval.reciprocal_rank ¶

openbtk.eval.retrieval.ndcg_at_k ¶

Clinical QA¶

openbtk.eval.qa.evaluate_qa ¶

openbtk.eval.qa.LLMAnswerer ¶

openbtk.eval.qa.QAReport ¶

accuracy property ¶

accuracy_ci95 property ¶

as_dict ¶

openbtk.eval.qa.MCQItem ¶

openbtk.eval.qa.read_medqa_jsonl ¶

openbtk.eval.qa.read_medmcqa_jsonl ¶

openbtk.eval.qa.parse_choice ¶

openbtk.eval.qa.format_prompt ¶

openbtk.eval.qa.qa_manifest ¶

Groundedness¶

openbtk.eval.groundedness.score_groundedness ¶

openbtk.eval.groundedness.evaluate_detector ¶

openbtk.eval.groundedness.GroundedExample ¶

openbtk.eval.groundedness.LabelledExample ¶

openbtk.eval.groundedness.GroundednessReport ¶

faithfulness property ¶

grounded_rate property ¶

openbtk.eval.groundedness.DetectorReport ¶

openbtk.eval.groundedness.groundedness_manifest ¶

Summaries: ROUGE and BERTScore¶

openbtk.eval.summarisation.evaluate_summaries ¶

openbtk.eval.summarisation.rouge_scores ¶

openbtk.eval.summarisation.BertScorer ¶

provenance ¶

score ¶

openbtk.eval.summarisation.SummaryPair ¶

openbtk.eval.summarisation.SummarisationReport ¶

as_dict ¶

openbtk.eval.summarisation.summarisation_manifest ¶

Model cards¶

openbtk.eval.model_card.ModelCard ¶

for_component classmethod ¶

to_markdown ¶

with_evaluation ¶

write ¶

openbtk.eval.model_card.CardEvaluation ¶

openbtk.eval.model_card.cards_from_run ¶

Manifests¶

openbtk.eval.manifest.EvalManifest ¶

openbtk.eval.manifest.build_manifest ¶

openbtk.eval.manifest.file_digest ¶

overall `property` ¶

accuracy `property` ¶

accuracy_ci95 `property` ¶

faithfulness `property` ¶

grounded_rate `property` ¶

for_component `classmethod` ¶