Skip to content

API reference

This page is generated automatically from the docstrings in cafe-core.

Study & factors

cafe.Study dataclass

A complete evaluation: a system, the factors to vary, the inputs, and how to judge the results.

Fields

system: The black box under test, runnable as run(config, item) -> output — a plain callable (sync or async) or an object with such a method. factors: The axes to vary. Their Cartesian product (full factorial) defines the configurations. dataset: The evaluation set, one element per item. An item may be any value; if it is a mapping it may carry "id" (for resume), "text" (the question shown to the judge), and "reference" (a gold answer for the judge). rubric: The :class:cafe.Rubric the judge scores answers on. Optional — leave None to only generate answers without judging. judge: The :class:cafe.LLMJudge (or any object with a compatible score method). Optional, paired with rubric. replications: How many times each (configuration, input) is executed. This is how CAFE measures the system's run-to-run nondeterminism. judge_replications: How many times the judge re-scores each answer — measures the judge's own nondeterminism, separately from the system's.

Source code in packages/cafe-core/src/cafe/study.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
@dataclass
class Study:
    """A complete evaluation: a system, the factors to vary, the inputs, and how
    to judge the results.

    Fields
    ------
    system:
        The black box under test, runnable as ``run(config, item) -> output`` —
        a plain callable (sync or async) or an object with such a method.
    factors:
        The axes to vary. Their Cartesian product (full factorial) defines the
        configurations.
    dataset:
        The evaluation set, one element per item. An item may be any value; if it
        is a mapping it may carry ``"id"`` (for resume), ``"text"`` (the question
        shown to the judge), and ``"reference"`` (a gold answer for the judge).
    rubric:
        The :class:`cafe.Rubric` the judge scores answers on. Optional —
        leave ``None`` to only generate answers without judging.
    judge:
        The :class:`cafe.LLMJudge` (or any object with a compatible ``score``
        method). Optional, paired with ``rubric``.
    replications:
        How many times each (configuration, input) is executed. This is how CAFE
        measures the *system's* run-to-run nondeterminism.
    judge_replications:
        How many times the judge re-scores each answer — measures the *judge's*
        own nondeterminism, separately from the system's.
    """

    name: str
    system: Any
    factors: list[Factor] = field(default_factory=list)
    dataset: list[Any] = field(default_factory=list)
    rubric: Any = None
    judge: Any = None
    design: str = "full_factorial"
    design_options: dict[str, Any] = field(default_factory=dict)
    replications: int = 1
    judge_replications: int = 1

    def __post_init__(self) -> None:
        names = [f.name for f in self.factors]
        if len(names) != len(set(names)):
            raise ValueError(f"duplicate factor names: {names}")
        if self.replications < 1:
            raise ValueError("replications must be >= 1")
        if self.judge_replications < 1:
            raise ValueError("judge_replications must be >= 1")

    # ── What this study will do, before running it ─────────────────────────────
    def plan(self) -> dict[str, Any]:
        """The run plan as numbers: configs, items, reps, total runs (and judge calls).

        Cheap — it expands the design but runs nothing. Useful for a sanity check or a
        cost estimate before spending tokens.
        """
        from cafe.design import size

        n_configs = size(self)
        n_items = max(1, len(self.dataset))
        runs = n_configs * n_items * self.replications
        judged = self.judge is not None and self.rubric is not None
        plan: dict[str, Any] = {
            "name": self.name,
            "design": self.design,
            "factors": [f.name for f in self.factors],
            "configs": n_configs,
            "items": n_items,
            "replications": self.replications,
            "runs": runs,
            "judged": judged,
        }
        if judged:
            plan["judge_calls"] = runs * self.judge_replications
        return plan

    def check(self) -> list[str]:
        """Advisory warnings about whether the design has enough data for the stats —
        cheap, runs nothing, meant to be read **before** spending tokens. Flags thin
        designs that make the mixed-effects models (linear / CLMM / logistic) unstable.
        Returns a list of human-readable strings (empty = nothing obviously wrong).

        Not a power analysis (mixed-model power needs simulation) — just heuristics on the
        numbers. Surfaced by :meth:`preflight` and emitted once when :func:`cafe.evaluate`
        starts.
        """
        from cafe.design import size

        warns: list[str] = []
        n_items = len(self.dataset)
        try:
            n_configs = size(self)
        except Exception:  # noqa: BLE001
            n_configs = 0
        scale = getattr(getattr(self.rubric, "scale_type", None), "value", None)

        # Enough questions to estimate the per-question random intercept?
        if n_items == 0:
            warns.append("no dataset items — nothing to run")
        elif n_items < 2:
            warns.append(
                f"only {n_items} input — the mixed-effects models need ≥2 questions for the "
                "per-question random intercept; they'll drop to a fixed-effects fit"
            )
        elif n_items < 8:
            warns.append(
                f"{n_items} inputs — the mixed-effects models (linear / CLMM / logistic) estimate a "
                "per-question random intercept; with fewer than ~8 questions those estimates can be "
                "near-singular/unstable. Add inputs for reliable p-values"
            )

        # Thin design: many configurations relative to the number of questions.
        if n_items and n_configs > n_items:
            warns.append(
                f"{n_configs} configurations but only {n_items} inputs — few observations per "
                "configuration; per-factor estimates will be weak. Add inputs, or reduce factors/levels"
            )

        # Binary rubric with few items → a factor may perfectly separate pass/fail.
        if scale == "binary" and 0 < n_items < 10:
            warns.append(
                f"binary rubric with {n_items} inputs — a factor may perfectly separate pass/fail, "
                "making the logistic model degenerate (odds ratios diverge). More inputs reduce this risk"
            )

        # Design shape: can this study actually compare anything?
        multi_level = [f for f in self.factors if len(f.levels) >= 2]
        if len(multi_level) == 0:
            warns.append(
                "no factor varies across ≥2 levels — there is nothing to compare; add a factor "
                "with at least two levels (e.g. an on/off feature to test 'does X help?')"
            )
        elif len(multi_level) == 1:
            warns.append(
                "only one factor varies — you can estimate its main effect but no interactions "
                "(you can't ask whether two techniques help *together*); add a second factor for that"
            )

        # Judge / rubric must both be set to judge; one without the other silently skips judging.
        if (self.judge is None) != (self.rubric is None):
            has, missing = ("judge", "rubric") if self.rubric is None else ("rubric", "judge")
            warns.append(
                f"a {has} is set but no {missing} — judging is skipped unless BOTH are provided "
                "(the study will only generate answers)"
            )

        # Judge replications at temperature 0 measure nothing (a deterministic judge won't vary).
        temp = getattr(self.judge, "temperature", None)
        if self.judge_replications >= 2 and temp == 0.0:
            warns.append(
                f"judge_replications={self.judge_replications} but the judge temperature is 0.0 — "
                "a deterministic judge won't vary between repetitions, so this measures no judge "
                "noise; raise the judge temperature to measure it"
            )

        # FactorType.continuous is advisory only — the stats layer treats every factor as categorical.
        continuous = [f.name for f in self.factors if getattr(f.type, "value", None) == "continuous"]
        if continuous:
            warns.append(
                f"factor(s) {continuous} are declared continuous, but the statistics currently treat "
                "every factor as categorical (each level is a separate group; no trend/slope is fit). "
                "Numeric-covariate modeling is planned — for now FactorType is advisory"
            )
        return warns

    def __repr__(self) -> str:
        try:
            p = self.plan()
        except Exception:  # noqa: BLE001 — repr must never raise
            return f"Study(name={self.name!r}, factors={[f.name for f in self.factors]})"
        head = (
            f"Study({p['name']!r}: {p['configs']} configs × {p['items']} items "
            f"× {p['replications']} reps = {p['runs']} runs"
        )
        parts = []
        if p["design"] != "full_factorial":
            parts.append(p["design"])
        parts.append("factors: " + (", ".join(p["factors"]) or "—"))
        if p["judged"]:
            jr = f" ×{self.judge_replications}" if self.judge_replications > 1 else ""
            parts.append(f"judged by {getattr(self.judge, 'model', 'judge')}{jr}")
        return head + "; " + "; ".join(parts) + ")"

    # ── Running, synchronously, from anywhere ──────────────────────────────────
    #
    # These wrappers work in plain scripts AND inside an already-running event
    # loop (e.g. a Jupyter notebook): if a loop is running we execute in a worker
    # thread so we never hit "asyncio.run() cannot be called from a running event
    # loop". Async-native callers can instead await the module functions
    # (``cafe.evaluate``, ``cafe.run_study``, ``cafe.preflight``).

    def evaluate(self, **kwargs: Any):
        """Run the **complete** evaluation: generate answers, judge them (if a
        rubric + judge are set), and attribute quality to the factors. Returns an
        :class:`cafe.evaluation.Evaluation`."""
        from cafe.evaluation import emit_design_warnings, evaluate

        # Emit design-adequacy warnings here — synchronously, before the event loop — so
        # they point at the caller's code rather than at asyncio internals.
        emit_design_warnings(self)
        return self._run_blocking(lambda: evaluate(self, _warn_design=False, **kwargs))

    def preflight(self, **kwargs: Any):
        """Quick check before a full run: one input through every configuration,
        no replication or judging, plus a cost/time estimate. Returns a
        :class:`cafe.evaluation.Preflight`."""
        from cafe.evaluation import preflight

        return self._run_blocking(lambda: preflight(self, **kwargs))

    def run(self, **kwargs: Any):
        """Lower-level: generate answers only (no judging). Returns
        :class:`cafe.Results`. Most users want :meth:`evaluate`."""
        from cafe.execution import run_study

        return self._run_blocking(lambda: run_study(self, **kwargs))

    def preview_judge_prompt(self, answer: str, item: Any = None) -> str:
        """Return the exact judge prompt for an example ``answer`` — no LLM call.

        Shows the messages exactly as sent, with role labels (``[SYSTEM]`` / ``[USER]``)
        so the system framing and the rubric-derived user prompt are both visible. Uses
        the study's rubric + judge and, by default, the first dataset item. Print it to
        verify the judging before spending any tokens.
        """
        if self.judge is None or self.rubric is None:
            raise ValueError("study has no judge/rubric set to preview")
        if item is None:
            item = self.dataset[0] if self.dataset else ""
        question = item["text"] if isinstance(item, dict) and "text" in item else str(item)
        reference = item.get("reference") if isinstance(item, dict) else None
        if not hasattr(self.judge, "preview"):
            raise ValueError("this judge has no .preview() to show a prompt (use an LLMJudge)")
        return self.judge.preview(self.rubric, question, answer, reference)

    def _run_blocking(self, make_coro):
        from cafe._async import run_blocking

        return run_blocking(make_coro)

plan 

plan()

The run plan as numbers: configs, items, reps, total runs (and judge calls).

Cheap — it expands the design but runs nothing. Useful for a sanity check or a cost estimate before spending tokens.

Source code in packages/cafe-core/src/cafe/study.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def plan(self) -> dict[str, Any]:
    """The run plan as numbers: configs, items, reps, total runs (and judge calls).

    Cheap — it expands the design but runs nothing. Useful for a sanity check or a
    cost estimate before spending tokens.
    """
    from cafe.design import size

    n_configs = size(self)
    n_items = max(1, len(self.dataset))
    runs = n_configs * n_items * self.replications
    judged = self.judge is not None and self.rubric is not None
    plan: dict[str, Any] = {
        "name": self.name,
        "design": self.design,
        "factors": [f.name for f in self.factors],
        "configs": n_configs,
        "items": n_items,
        "replications": self.replications,
        "runs": runs,
        "judged": judged,
    }
    if judged:
        plan["judge_calls"] = runs * self.judge_replications
    return plan

check 

check()

Advisory warnings about whether the design has enough data for the stats — cheap, runs nothing, meant to be read before spending tokens. Flags thin designs that make the mixed-effects models (linear / CLMM / logistic) unstable. Returns a list of human-readable strings (empty = nothing obviously wrong).

Not a power analysis (mixed-model power needs simulation) — just heuristics on the numbers. Surfaced by :meth:preflight and emitted once when :func:cafe.evaluate starts.

Source code in packages/cafe-core/src/cafe/study.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def check(self) -> list[str]:
    """Advisory warnings about whether the design has enough data for the stats —
    cheap, runs nothing, meant to be read **before** spending tokens. Flags thin
    designs that make the mixed-effects models (linear / CLMM / logistic) unstable.
    Returns a list of human-readable strings (empty = nothing obviously wrong).

    Not a power analysis (mixed-model power needs simulation) — just heuristics on the
    numbers. Surfaced by :meth:`preflight` and emitted once when :func:`cafe.evaluate`
    starts.
    """
    from cafe.design import size

    warns: list[str] = []
    n_items = len(self.dataset)
    try:
        n_configs = size(self)
    except Exception:  # noqa: BLE001
        n_configs = 0
    scale = getattr(getattr(self.rubric, "scale_type", None), "value", None)

    # Enough questions to estimate the per-question random intercept?
    if n_items == 0:
        warns.append("no dataset items — nothing to run")
    elif n_items < 2:
        warns.append(
            f"only {n_items} input — the mixed-effects models need ≥2 questions for the "
            "per-question random intercept; they'll drop to a fixed-effects fit"
        )
    elif n_items < 8:
        warns.append(
            f"{n_items} inputs — the mixed-effects models (linear / CLMM / logistic) estimate a "
            "per-question random intercept; with fewer than ~8 questions those estimates can be "
            "near-singular/unstable. Add inputs for reliable p-values"
        )

    # Thin design: many configurations relative to the number of questions.
    if n_items and n_configs > n_items:
        warns.append(
            f"{n_configs} configurations but only {n_items} inputs — few observations per "
            "configuration; per-factor estimates will be weak. Add inputs, or reduce factors/levels"
        )

    # Binary rubric with few items → a factor may perfectly separate pass/fail.
    if scale == "binary" and 0 < n_items < 10:
        warns.append(
            f"binary rubric with {n_items} inputs — a factor may perfectly separate pass/fail, "
            "making the logistic model degenerate (odds ratios diverge). More inputs reduce this risk"
        )

    # Design shape: can this study actually compare anything?
    multi_level = [f for f in self.factors if len(f.levels) >= 2]
    if len(multi_level) == 0:
        warns.append(
            "no factor varies across ≥2 levels — there is nothing to compare; add a factor "
            "with at least two levels (e.g. an on/off feature to test 'does X help?')"
        )
    elif len(multi_level) == 1:
        warns.append(
            "only one factor varies — you can estimate its main effect but no interactions "
            "(you can't ask whether two techniques help *together*); add a second factor for that"
        )

    # Judge / rubric must both be set to judge; one without the other silently skips judging.
    if (self.judge is None) != (self.rubric is None):
        has, missing = ("judge", "rubric") if self.rubric is None else ("rubric", "judge")
        warns.append(
            f"a {has} is set but no {missing} — judging is skipped unless BOTH are provided "
            "(the study will only generate answers)"
        )

    # Judge replications at temperature 0 measure nothing (a deterministic judge won't vary).
    temp = getattr(self.judge, "temperature", None)
    if self.judge_replications >= 2 and temp == 0.0:
        warns.append(
            f"judge_replications={self.judge_replications} but the judge temperature is 0.0 — "
            "a deterministic judge won't vary between repetitions, so this measures no judge "
            "noise; raise the judge temperature to measure it"
        )

    # FactorType.continuous is advisory only — the stats layer treats every factor as categorical.
    continuous = [f.name for f in self.factors if getattr(f.type, "value", None) == "continuous"]
    if continuous:
        warns.append(
            f"factor(s) {continuous} are declared continuous, but the statistics currently treat "
            "every factor as categorical (each level is a separate group; no trend/slope is fit). "
            "Numeric-covariate modeling is planned — for now FactorType is advisory"
        )
    return warns

evaluate 

evaluate(**kwargs)

Run the complete evaluation: generate answers, judge them (if a rubric + judge are set), and attribute quality to the factors. Returns an :class:cafe.evaluation.Evaluation.

Source code in packages/cafe-core/src/cafe/study.py 
244
245
246
247
248
249
250
251
252
253
def evaluate(self, **kwargs: Any):
    """Run the **complete** evaluation: generate answers, judge them (if a
    rubric + judge are set), and attribute quality to the factors. Returns an
    :class:`cafe.evaluation.Evaluation`."""
    from cafe.evaluation import emit_design_warnings, evaluate

    # Emit design-adequacy warnings here — synchronously, before the event loop — so
    # they point at the caller's code rather than at asyncio internals.
    emit_design_warnings(self)
    return self._run_blocking(lambda: evaluate(self, _warn_design=False, **kwargs))

preflight 

preflight(**kwargs)

Quick check before a full run: one input through every configuration, no replication or judging, plus a cost/time estimate. Returns a :class:cafe.evaluation.Preflight.

Source code in packages/cafe-core/src/cafe/study.py 
255
256
257
258
259
260
261
def preflight(self, **kwargs: Any):
    """Quick check before a full run: one input through every configuration,
    no replication or judging, plus a cost/time estimate. Returns a
    :class:`cafe.evaluation.Preflight`."""
    from cafe.evaluation import preflight

    return self._run_blocking(lambda: preflight(self, **kwargs))

run 

run(**kwargs)

Lower-level: generate answers only (no judging). Returns :class:cafe.Results. Most users want :meth:evaluate.

Source code in packages/cafe-core/src/cafe/study.py 
263
264
265
266
267
268
def run(self, **kwargs: Any):
    """Lower-level: generate answers only (no judging). Returns
    :class:`cafe.Results`. Most users want :meth:`evaluate`."""
    from cafe.execution import run_study

    return self._run_blocking(lambda: run_study(self, **kwargs))

preview_judge_prompt 

preview_judge_prompt(answer, item=None)

Return the exact judge prompt for an example answer — no LLM call.

Shows the messages exactly as sent, with role labels ([SYSTEM] / [USER]) so the system framing and the rubric-derived user prompt are both visible. Uses the study's rubric + judge and, by default, the first dataset item. Print it to verify the judging before spending any tokens.

Source code in packages/cafe-core/src/cafe/study.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def preview_judge_prompt(self, answer: str, item: Any = None) -> str:
    """Return the exact judge prompt for an example ``answer`` — no LLM call.

    Shows the messages exactly as sent, with role labels (``[SYSTEM]`` / ``[USER]``)
    so the system framing and the rubric-derived user prompt are both visible. Uses
    the study's rubric + judge and, by default, the first dataset item. Print it to
    verify the judging before spending any tokens.
    """
    if self.judge is None or self.rubric is None:
        raise ValueError("study has no judge/rubric set to preview")
    if item is None:
        item = self.dataset[0] if self.dataset else ""
    question = item["text"] if isinstance(item, dict) and "text" in item else str(item)
    reference = item.get("reference") if isinstance(item, dict) else None
    if not hasattr(self.judge, "preview"):
        raise ValueError("this judge has no .preview() to show a prompt (use an LLMJudge)")
    return self.judge.preview(self.rubric, question, answer, reference)

cafe.Factor dataclass

One axis of the experiment.

levels is the set of values this factor can take. Full factorial visits every combination of every factor's levels — there is no limit on the number of factors or levels (that limit only applies to fractional/screening designs).

Source code in packages/cafe-core/src/cafe/study.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass
class Factor:
    """One axis of the experiment.

    ``levels`` is the set of values this factor can take. Full factorial visits
    every combination of every factor's levels — there is no limit on the number
    of factors or levels (that limit only applies to fractional/screening designs).
    """

    name: str
    levels: list[Any]
    type: FactorType = FactorType.categorical

    def __post_init__(self) -> None:
        if not self.name or not isinstance(self.name, str):
            raise ValueError("factor name must be a non-empty string")
        self.levels = list(self.levels)
        if len(self.levels) == 0:
            raise ValueError(f"factor {self.name!r} must have at least one level")
        if isinstance(self.type, str):
            self.type = FactorType(self.type)

cafe.FactorType

Bases: str, Enum

How a factor is treated by design generation and (later) statistics.

Source code in packages/cafe-core/src/cafe/study.py 
20
21
22
23
24
25
class FactorType(str, Enum):
    """How a factor is treated by design generation and (later) statistics."""

    categorical = "categorical"  # unordered: reranker in {none, cross_encoder}
    ordinal = "ordinal"          # ordered: effort in {low, med, high}
    continuous = "continuous"    # numeric knob: temperature, top_k

Running & estimating

cafe.evaluate async 

evaluate(study, *, concurrency=8, checkpoint_path=None, judge_checkpoint_path=None, on_progress=None, progress=True, _warn_design=True)

Generate answers, judge them, and attribute quality — the whole pipeline.

Shows a progress bar by default (one for answers, one for judging); pass progress=False to silence it, or on_progress for custom reporting. checkpoint_path / judge_checkpoint_path make the answer / judging phases crash-safe and resumable.

Source code in packages/cafe-core/src/cafe/evaluation.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
async def evaluate(
    study: "Study",
    *,
    concurrency: int = 8,
    checkpoint_path: str | None = None,
    judge_checkpoint_path: str | None = None,
    on_progress: ProgressFn | None = None,
    progress: bool = True,
    _warn_design: bool = True,
) -> Evaluation:
    """Generate answers, judge them, and attribute quality — the whole pipeline.

    Shows a progress bar by default (one for answers, one for judging); pass
    ``progress=False`` to silence it, or ``on_progress`` for custom reporting.
    ``checkpoint_path`` / ``judge_checkpoint_path`` make the answer / judging phases
    crash-safe and resumable.
    """
    # Advise on thin designs before spending tokens (unless the sync wrapper already did).
    if _warn_design:
        emit_design_warnings(study)

    answers = await run_study(
        study,
        replications=study.replications,
        concurrency=concurrency,
        checkpoint_path=checkpoint_path,
        on_progress=on_progress,
        progress=progress,
    )

    questions, references = _question_and_reference_maps(study)
    ratings: Ratings | None = None
    attribution: Attribution | None = None
    if study.judge is not None and study.rubric is not None:
        ratings = await judge_results(
            answers,
            study.judge,
            study.rubric,
            repetitions=study.judge_replications,
            concurrency=concurrency,
            questions=questions,
            references=references,
            checkpoint_path=judge_checkpoint_path,
            progress=progress,
        )
        attribution = attribute(ratings)

    return Evaluation(
        study_name=study.name,
        answers=answers,
        ratings=ratings,
        attribution=attribution,
        questions=questions,
        references=references,
    )

cafe.run_study async 

run_study(study, *, replications=None, concurrency=8, checkpoint_path=None, resume=True, smoke=False, on_progress=None, progress=False)

Run a study end-to-end and return its :class:Results.

Parameters

replications: Repetitions per (config, input). Defaults to study.replications. Replication is how CAFE measures run-to-run nondeterminism. concurrency: Max cells in flight at once. checkpoint_path: If given, observations are appended here as they complete and a resumed run skips already-done cells. If None, the run is in-memory only. smoke: Preflight: one input through every config, a single replication, no judging — to confirm configs execute and to estimate cost before committing to the full study. on_progress: Called as on_progress(observation, completed, total) after each cell.

Source code in packages/cafe-core/src/cafe/execution/runner.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
async def run_study(
    study: Study,
    *,
    replications: int | None = None,
    concurrency: int = 8,
    checkpoint_path: str | None = None,
    resume: bool = True,
    smoke: bool = False,
    on_progress: ProgressFn | None = None,
    progress: bool = False,
) -> Results:
    """Run a study end-to-end and return its :class:`Results`.

    Parameters
    ----------
    replications:
        Repetitions per (config, input). Defaults to ``study.replications``.
        Replication is how CAFE measures run-to-run nondeterminism.
    concurrency:
        Max cells in flight at once.
    checkpoint_path:
        If given, observations are appended here as they complete and a resumed
        run skips already-done cells. If ``None``, the run is in-memory only.
    smoke:
        Preflight: one input through every config, a single replication, no
        judging — to confirm configs execute and to estimate cost before
        committing to the full study.
    on_progress:
        Called as ``on_progress(observation, completed, total)`` after each cell.
    """
    reps = study.replications if replications is None else replications
    if reps < 1:
        raise ValueError("replications must be >= 1")

    configs = design.generate(study)
    inputs = list(study.dataset)
    if smoke:
        inputs = inputs[:1]
        reps = 1
    if not inputs:
        raise ValueError("study has no dataset to run")

    system = as_system(study.system)

    # Resume: load prior observations and skip their keys.
    checkpoint = None
    done: dict[str, Observation] = {}
    if checkpoint_path is not None:
        from cafe.execution.checkpoint import Checkpoint

        checkpoint = Checkpoint(checkpoint_path)
        if resume:
            done = checkpoint.load()

    # Build the full cell list, then drop the ones already done.
    pending: list[tuple[dict[str, Any], Any, str, int]] = []
    valid_keys: set[str] = set()
    for cfg in configs:
        for idx, item in enumerate(inputs):
            in_id = _input_id(item, idx)
            for rep in range(reps):
                probe = Observation(config=cfg, input_id=in_id, rep=rep)
                valid_keys.add(probe.key())
                if probe.key() not in done:
                    pending.append((cfg, item, in_id, rep))

    # Only carry forward checkpoint rows that belong to the CURRENT design. Resuming an
    # *edited* study (a dropped level, a changed dataset) must not contaminate the results
    # with ghost rows from the old design — they would silently enter the statistics.
    kept = {k: o for k, o in done.items() if k in valid_keys}
    stale = len(done) - len(kept)
    if stale:
        import warnings

        warnings.warn(
            f"checkpoint {checkpoint_path!r}: dropped {stale} stored observation(s) that are "
            "no longer in the study's design (the study changed since the checkpoint was "
            "written); resumed results contain only the current design's cells.",
            stacklevel=2,
        )

    total = len(configs) * len(inputs) * reps
    observations: list[Observation] = list(kept.values())

    sem = asyncio.Semaphore(max(1, concurrency))
    write_lock = asyncio.Lock()
    completed = len(kept)

    from cafe.execution.progress import progress_bar

    with progress_bar(total, f"{study.name}: answers", enabled=progress and on_progress is None) as bar:
        report = on_progress or bar

        async def worker(cfg: dict[str, Any], item: Any, in_id: str, rep: int) -> None:
            nonlocal completed
            async with sem:
                obs = await _run_cell(system, cfg, item, in_id, rep)
            async with write_lock:
                observations.append(obs)
                if checkpoint is not None:
                    checkpoint.append(obs)
                completed += 1
                if report is not None:
                    report(obs, completed, total)

        if pending:
            await asyncio.gather(*(worker(*cell) for cell in pending))

    return Results(
        study_name=study.name,
        factors=[f.name for f in study.factors],
        observations=observations,
    )

cafe.preflight async 

preflight(study, *, concurrency=8, progress=False)

Run one input through every configuration (no reps, no judging) and estimate the full study's cost/time.

Source code in packages/cafe-core/src/cafe/evaluation.py 
500
501
502
503
504
505
506
507
508
509
async def preflight(study: "Study", *, concurrency: int = 8, progress: bool = False) -> Preflight:
    """Run one input through every configuration (no reps, no judging) and
    estimate the full study's cost/time."""
    answers = await run_study(study, smoke=True, concurrency=concurrency, progress=progress)
    total_cells = size(study) * max(1, len(study.dataset)) * study.replications
    plan = study.plan()
    return Preflight(
        answers=answers, estimate=estimate(answers, total_cells),
        warnings=study.check(), judge_calls=plan.get("judge_calls", 0),
    )

cafe.estimate 

estimate(results, total_cells)

Rough full-study time/cost estimate from a smoke/partial run.

Uses mean per-cell time and any cost_usd metadata observed so far, extrapolated to total_cells. Conservative and meant for the preflight.

Source code in packages/cafe-core/src/cafe/execution/runner.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def estimate(results: Results, total_cells: int) -> dict[str, Any]:
    """Rough full-study time/cost estimate from a smoke/partial run.

    Uses mean per-cell time and any ``cost_usd`` metadata observed so far,
    extrapolated to ``total_cells``. Conservative and meant for the preflight.
    """
    oks = [o for o in results.observations if o.ok]
    times = [o.elapsed_s for o in oks if o.elapsed_s is not None]
    costs = [o.metadata.get("cost_usd") for o in oks if "cost_usd" in o.metadata]
    mean_t = sum(times) / len(times) if times else None
    mean_c = sum(costs) / len(costs) if costs else None
    return {
        "sampled_cells": len(oks),
        "total_cells": total_cells,
        "est_total_compute_s": round(mean_t * total_cells, 2) if mean_t else None,
        "est_total_cost_usd": round(mean_c * total_cells, 4) if mean_c else None,
        "labels": [config_label(o.config) for o in oks[:1]],
    }

Results

cafe.Evaluation dataclass

The complete result of evaluating a study: answers, ratings, attribution.

Everything is reachable as plain data: answers.observations (each Observation), ratings.items (each Rating, with prompt / raw_response / value_numeric), attribution / effects / clmm, and the answers.df / ratings.df DataFrames. :meth:records joins it all into one row-per-verdict view.

Source code in packages/cafe-core/src/cafe/evaluation.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
@dataclass
class Evaluation:
    """The complete result of evaluating a study: answers, ratings, attribution.

    Everything is reachable as plain data: ``answers.observations`` (each
    ``Observation``), ``ratings.items`` (each ``Rating``, with ``prompt`` /
    ``raw_response`` / ``value_numeric``), ``attribution`` / ``effects`` / ``clmm``,
    and the ``answers.df`` / ``ratings.df`` DataFrames. :meth:`records` joins it all
    into one row-per-verdict view.
    """

    study_name: str
    answers: Results
    ratings: Ratings | None = None
    attribution: Attribution | None = None
    questions: dict[str, str] = field(default_factory=dict)    # input_id -> question text
    references: dict[str, str] = field(default_factory=dict)   # input_id -> gold answer

    _effects_cache: Any = field(default=None, repr=False, compare=False)

    _clmm_cache: Any = field(default=None, repr=False, compare=False)

    _logistic_cache: Any = field(default=None, repr=False, compare=False)

    def records(self) -> list[dict[str, Any]]:
        """One row per judge verdict, joining **everything** for inspection/export:
        question, reference, the factor levels, the answer, per-answer cost/latency, the
        full judge prompt (system + user), the raw judge response, and the parsed verdict.

        This is the "give me everything" view — ``import pandas as pd;
        pd.DataFrame(ev.records())`` and slice however you like. With no judge it falls
        back to one row per answer.
        """
        by_key = {o.key(): o for o in self.answers.observations}
        sys_prompt = self.ratings.judge_system_prompt if self.ratings else None
        rows: list[dict[str, Any]] = []

        if self.ratings is None or not self.ratings.items:
            for o in self.answers.observations:
                meta = o.metadata or {}
                rows.append({
                    "input_id": o.input_id,
                    "question": self.questions.get(o.input_id),
                    "reference": self.references.get(o.input_id),
                    **o.config, "rep": o.rep, "answer": o.output,
                    "elapsed_s": o.elapsed_s,
                    "cost_usd": meta.get("cost_usd"), "tokens": meta.get("tokens"),
                    "error": o.error,
                })
            return rows

        for r in self.ratings.items:
            obs = by_key.get(r.obs_key)
            meta = (obs.metadata if obs else {}) or {}
            rows.append({
                "input_id": r.input_id,
                "question": self.questions.get(r.input_id),
                "reference": self.references.get(r.input_id),
                **r.config,
                "rep": r.rep,
                "judge_rep": r.judge_rep,
                "answer": obs.output if obs else None,
                "elapsed_s": obs.elapsed_s if obs else None,
                "cost_usd": meta.get("cost_usd"),
                "tokens": meta.get("tokens"),
                "judge_system": sys_prompt,
                "judge_prompt": r.prompt,
                "judge_raw": r.raw_response,
                "verdict": r.value_numeric,
                "reasoning": r.reasoning,
                "error": r.error,
            })
        return rows

    @property
    def overall_mean(self) -> float | None:
        """Mean verdict across all usable answers (judge replications averaged), or
        ``None`` if unjudged. The single-number summary — e.g. a pass rate for a binary
        rubric. Per-configuration / per-factor breakdowns live on ``attribution``."""
        return self.attribution.overall_mean if self.attribution is not None else None

    @property
    def marginal_means(self):
        """Per-factor marginal means as a tidy DataFrame (factor, level, mean, n) — the
        "which level scores higher" table, ready to drop into a paper. ``None`` if
        unjudged. (The same data is in ``attribution.factor_marginals``.)"""
        if self.attribution is None:
            return None
        import pandas as pd

        return pd.DataFrame(self.attribution.factor_marginals)

    @property
    def residuals(self):
        """Per-answer residuals from the inferential fit (like R's ``residuals(fit)``),
        or ``None`` if unjudged. Handy for diagnostics (normality, outliers)."""
        eff = self.effects
        return list(eff.residuals) if eff is not None and eff.residuals else None

    @property
    def fitted(self):
        """Per-answer fitted values from the inferential fit (like R's ``fitted(fit)``),
        or ``None`` if unjudged."""
        eff = self.effects
        return list(eff.fitted) if eff is not None and eff.fitted else None

    @property
    def variance_components(self):
        """The mixed model's ``{random_intercept, residual}`` variances (like R's
        ``VarCorr(fit)``), or ``None`` if unjudged / only a fixed-effects fit was possible."""
        eff = self.effects
        return eff.variance_components if eff is not None else None

    @property
    def effects(self):
        """Inferential statistics (mixed model → F/p, partial η², Cohen's d).

        Computed lazily on first access from the ratings; cached thereafter.
        Returns ``None`` if the study wasn't judged.
        """
        if self.ratings is None:
            return None
        if self._effects_cache is None:
            from cafe.stats import fit_effects

            self._effects_cache = fit_effects(self.ratings)
        return self._effects_cache

    @property
    def clmm(self):
        """Ordinal CLMM (R) for ordinal rubrics. Lazy; returns ``None`` if unjudged.

        The result's ``available`` flag says whether R produced a fit; if not, its
        ``reason`` explains why (e.g. R not installed) — see ``cafe doctor``.
        """
        if self.ratings is None:
            return None
        if self._clmm_cache is None:
            from cafe.stats import fit_clmm

            self._clmm_cache = fit_clmm(self.ratings)
        return self._clmm_cache

    @property
    def logistic(self):
        """Binary logistic model (log-odds / odds ratios) for **binary** pass/fail
        rubrics. Lazy; returns ``None`` if unjudged. ``available=False`` (with a
        ``reason``) for non-binary rubrics or degenerate data."""
        if self.ratings is None:
            return None
        if self._logistic_cache is None:
            from cafe.stats import fit_logistic

            self._logistic_cache = fit_logistic(self.ratings)
        return self._logistic_cache

    def report(self, *, interactions: int = 2) -> str:
        """The **full** statistical picture in one string: a pipeline summary, the
        descriptive layer, then the **model matched to the rubric's scale type** —
        numeric → linear mixed model, ordinal → linear + cumulative-link mixed model,
        binary → logistic. So each scale is analysed with its statistically correct model.

        ``interactions`` is the max interaction order to model (2 = also two-way, the
        default; 1 = main effects only). The cheap ``repr`` shows only the descriptive
        layer (so displaying a result is instant); ``report()`` additionally fits the
        models, so the first call may take a moment. Print it::

            print(result.report())
        """
        from cafe.judging.rubric import ScaleType

        bar = "─" * 60
        parts = [self.show(), "", self._pipeline_line()]
        warns = self._health_warnings()
        if warns:
            parts.append("")
            parts += [f"⚠ {w}" for w in warns]
        if self.attribution is not None:
            parts += ["", bar, "DESCRIPTIVE — means & best configuration", bar,
                      self.attribution.show()]
        if self.ratings is not None:
            from cafe.stats import fit_clmm, fit_effects, fit_logistic

            scale = getattr(getattr(self.ratings, "rubric", None), "scale_type", None)
            if scale == ScaleType.binary:
                log = self.logistic if interactions == 2 else fit_logistic(self.ratings, interactions=interactions)
                parts += ["", bar, "LOGISTIC — binary pass/fail model (log-odds & odds ratios)",
                          bar, log.show()]
            elif scale == ScaleType.numeric:
                eff = self.effects if interactions == 2 else fit_effects(self.ratings, interactions=interactions)
                parts += ["", bar, "LINEAR — Gaussian mixed-effects model (significance & effect sizes)",
                          bar, eff.show()]
            else:  # ordinal (and the default): the linear view + the correct ordinal CLMM
                eff = self.effects if interactions == 2 else fit_effects(self.ratings, interactions=interactions)
                parts += ["", bar, "INFERENTIAL — mixed-effects model (significance & effect sizes)",
                          bar, eff.show()]
                clmm = self.clmm if interactions == 2 else fit_clmm(self.ratings, interactions=interactions)
                parts += ["", bar, "ORDINAL — cumulative-link mixed model (verdicts as ordered categories)",
                          bar, clmm.show()]
        return "\n".join(parts)

    def rejudge(self, judge: Any, *, rubric: Any = None, repetitions: int = 1,
                name: str | None = None, concurrency: int = 8,
                checkpoint_path: str | None = None, progress: bool = True):
        """Score the **same answers** again — with a different judge, ``rubric``, or
        number of ``repetitions`` — returning a new ``Evaluation`` (reusing this one's
        questions/references; nothing is regenerated). This is the "generate once, judge
        many ways" path:

            free = result.rejudge(cafe.LLMJudge(model=m, preset="single_answer"))
            biny = result.rejudge(judge, rubric=cafe.rubrics.CORRECT_PASS_FAIL)
            noise = result.rejudge(judge, repetitions=3)   # the judge's own spread

        ``checkpoint_path`` makes the judging crash-safe/resumable (verdicts appended as
        they land). Also handy for judge↔judge reliability —
        ``cafe.reliability(raters={"a": result, "b": result.rejudge(judge_b)})``.
        """
        from cafe._async import run_blocking
        from cafe.judging import judge_results
        from cafe.stats import attribute

        rubric = rubric or getattr(self.ratings, "rubric", None)
        if rubric is None:
            raise ValueError("rejudge needs a rubric — this evaluation wasn't judged")
        ratings = run_blocking(lambda: judge_results(
            self.answers, judge, rubric, repetitions=repetitions, concurrency=concurrency,
            questions=self.questions, references=self.references,
            checkpoint_path=checkpoint_path, progress=progress,
        ))
        return Evaluation(
            study_name=name or f"{self.study_name} ({getattr(judge, 'model', 'judge')})",
            answers=self.answers, ratings=ratings, attribution=attribute(ratings),
            questions=self.questions, references=self.references,
        )

    def judge_stability(self):
        """How consistently the judge scored the **same answer** across its repetitions —
        the judge analogue of system ``replications``. Returns a
        :class:`cafe.stats.JudgeStability` (per-answer std dev + a summary). Only
        informative when the judge scored answers more than once (``judge_replications``
        on the study, or ``rejudge(..., repetitions=k)``)::

            noisy = result.rejudge(judge, repetitions=3)
            print(noisy.judge_stability().show())
        """
        from cafe.stats import judge_stability as _judge_stability

        if self.ratings is None:
            raise ValueError("no ratings to measure — this evaluation wasn't judged")
        return _judge_stability(self.ratings)

    def pipeline(self):
        """The composed pipeline (stage order + levels), observed from the run's traces.
        Only meaningful for a composed system; returns an empty pipeline otherwise."""
        from cafe.techniques.composed import pipeline as _pipeline

        return _pipeline(self)

    def plot(self, kind: str | None = None, **kwargs):
        """Plots for this evaluation. No argument → a **dashboard** of the key plots; or
        pass a ``kind``: ``"marginals"``, ``"interaction"``, ``"configs"``,
        ``"distribution"``, ``"effects"``, ``"pareto"``. Returns the matplotlib Figure
        (dashboard) or Axes (single plot) — ``.savefig(...)`` or tweak it freely::

            result.plot()                 # everything
            result.plot("interaction")    # just one
        """
        from cafe.stats.plots import plot as _plot

        return _plot(self, kind, **kwargs)

    def _pipeline_line(self) -> str:
        """A funnel: answers attempted → judged → usable, so dropped cells are visible."""
        n_ans, n_failed = len(self.answers), len(self.answers.errors)
        seg = f"pipeline: {n_ans} answers"
        if n_failed:
            seg += f" ({n_failed} failed to generate)"
        if self.ratings is not None:
            n_bad = len(self.ratings.errors)
            seg += f"  →  {len(self.ratings)} judged"
            if n_bad:
                seg += f" ({n_bad} unparseable)"
            seg += f"  →  {len(self.ratings) - n_bad} usable verdict(s)"
        return seg

    def _health_warnings(self) -> list[str]:
        """Problems worth flagging: failed generations (which unbalance the design) and
        unparseable verdicts — distinguished, since they happen at different stages."""
        from collections import Counter

        from cafe.execution.results import config_label

        out: list[str] = []
        failed = self.answers.errors
        if failed:
            by = Counter(config_label(o.config) for o in failed)
            top = ", ".join(f"{c} ({n})" for c, n in by.most_common(2))
            out.append(
                f"{len(failed)} of {len(self.answers)} answers failed to generate "
                f"(mostly {top}) — the design is unbalanced; inspect result.answers.errors"
            )
        if self.ratings is not None and self.ratings.errors:
            errs = self.ratings.errors
            unjudgeable = [r for r in errs if (r.error or "").startswith("unjudgeable")]
            unparseable = [r for r in errs if r not in unjudgeable]
            if unjudgeable:
                out.append(
                    f"{len(unjudgeable)} answer(s) produced no output (None) and could not be "
                    "judged — they are recorded as errors; inspect result.ratings.failures()"
                )
            if unparseable:
                out.append(
                    f"{len(unparseable)} verdict(s) were unparseable — "
                    "inspect result.ratings.failures()"
                )
        return out

    def summary(self) -> dict[str, Any]:
        out = self.answers.summary()
        if self.ratings is not None:
            out["n_ratings"] = len(self.ratings)
            out["n_rating_errors"] = len(self.ratings.errors)
        if self.attribution is not None and self.attribution.best_config is not None:
            out["best_config"] = self.attribution.best_config["config"]
        return out

    def show(self) -> str:
        s = self.summary()
        line = f"{s['n_answers']} answers"
        if s.get("n_errors"):
            line += f" ({s['n_errors']} failed)"
        line += f" · {s['n_configs']} configs · {s['n_inputs']} inputs"
        if "n_ratings" in s:
            line += f" · {s['n_ratings']} ratings"
        best = s.get("best_config")
        if best:
            line += " · best: " + "·".join(f"{k}={best[k]}" for k in sorted(best))
        return line

    def __repr__(self) -> str:
        return f"Evaluation({self.show()})"

    def _repr_html_(self) -> str:  # pragma: no cover - notebook display
        parts = [f"<b>Evaluation</b> — {self.show()}"]
        if self.attribution is not None:
            parts.append(f"<pre>{self.attribution.show()}</pre>")
        return "".join(parts)

overall_mean property 

overall_mean

Mean verdict across all usable answers (judge replications averaged), or None if unjudged. The single-number summary — e.g. a pass rate for a binary rubric. Per-configuration / per-factor breakdowns live on attribution.

marginal_means property 

marginal_means

Per-factor marginal means as a tidy DataFrame (factor, level, mean, n) — the "which level scores higher" table, ready to drop into a paper. None if unjudged. (The same data is in attribution.factor_marginals.)

residuals property 

residuals

Per-answer residuals from the inferential fit (like R's residuals(fit)), or None if unjudged. Handy for diagnostics (normality, outliers).

fitted property 

fitted

Per-answer fitted values from the inferential fit (like R's fitted(fit)), or None if unjudged.

variance_components property 

variance_components

The mixed model's {random_intercept, residual} variances (like R's VarCorr(fit)), or None if unjudged / only a fixed-effects fit was possible.

effects property 

effects

Inferential statistics (mixed model → F/p, partial η², Cohen's d).

Computed lazily on first access from the ratings; cached thereafter. Returns None if the study wasn't judged.

clmm property 

clmm

Ordinal CLMM (R) for ordinal rubrics. Lazy; returns None if unjudged.

The result's available flag says whether R produced a fit; if not, its reason explains why (e.g. R not installed) — see cafe doctor.

logistic property 

logistic

Binary logistic model (log-odds / odds ratios) for binary pass/fail rubrics. Lazy; returns None if unjudged. available=False (with a reason) for non-binary rubrics or degenerate data.

records 

records()

One row per judge verdict, joining everything for inspection/export: question, reference, the factor levels, the answer, per-answer cost/latency, the full judge prompt (system + user), the raw judge response, and the parsed verdict.

This is the "give me everything" view — import pandas as pd; pd.DataFrame(ev.records()) and slice however you like. With no judge it falls back to one row per answer.

Source code in packages/cafe-core/src/cafe/evaluation.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def records(self) -> list[dict[str, Any]]:
    """One row per judge verdict, joining **everything** for inspection/export:
    question, reference, the factor levels, the answer, per-answer cost/latency, the
    full judge prompt (system + user), the raw judge response, and the parsed verdict.

    This is the "give me everything" view — ``import pandas as pd;
    pd.DataFrame(ev.records())`` and slice however you like. With no judge it falls
    back to one row per answer.
    """
    by_key = {o.key(): o for o in self.answers.observations}
    sys_prompt = self.ratings.judge_system_prompt if self.ratings else None
    rows: list[dict[str, Any]] = []

    if self.ratings is None or not self.ratings.items:
        for o in self.answers.observations:
            meta = o.metadata or {}
            rows.append({
                "input_id": o.input_id,
                "question": self.questions.get(o.input_id),
                "reference": self.references.get(o.input_id),
                **o.config, "rep": o.rep, "answer": o.output,
                "elapsed_s": o.elapsed_s,
                "cost_usd": meta.get("cost_usd"), "tokens": meta.get("tokens"),
                "error": o.error,
            })
        return rows

    for r in self.ratings.items:
        obs = by_key.get(r.obs_key)
        meta = (obs.metadata if obs else {}) or {}
        rows.append({
            "input_id": r.input_id,
            "question": self.questions.get(r.input_id),
            "reference": self.references.get(r.input_id),
            **r.config,
            "rep": r.rep,
            "judge_rep": r.judge_rep,
            "answer": obs.output if obs else None,
            "elapsed_s": obs.elapsed_s if obs else None,
            "cost_usd": meta.get("cost_usd"),
            "tokens": meta.get("tokens"),
            "judge_system": sys_prompt,
            "judge_prompt": r.prompt,
            "judge_raw": r.raw_response,
            "verdict": r.value_numeric,
            "reasoning": r.reasoning,
            "error": r.error,
        })
    return rows

report 

report(*, interactions=2)

The full statistical picture in one string: a pipeline summary, the descriptive layer, then the model matched to the rubric's scale type — numeric → linear mixed model, ordinal → linear + cumulative-link mixed model, binary → logistic. So each scale is analysed with its statistically correct model.

interactions is the max interaction order to model (2 = also two-way, the default; 1 = main effects only). The cheap repr shows only the descriptive layer (so displaying a result is instant); report() additionally fits the models, so the first call may take a moment. Print it::

print(result.report())
Source code in packages/cafe-core/src/cafe/evaluation.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def report(self, *, interactions: int = 2) -> str:
    """The **full** statistical picture in one string: a pipeline summary, the
    descriptive layer, then the **model matched to the rubric's scale type** —
    numeric → linear mixed model, ordinal → linear + cumulative-link mixed model,
    binary → logistic. So each scale is analysed with its statistically correct model.

    ``interactions`` is the max interaction order to model (2 = also two-way, the
    default; 1 = main effects only). The cheap ``repr`` shows only the descriptive
    layer (so displaying a result is instant); ``report()`` additionally fits the
    models, so the first call may take a moment. Print it::

        print(result.report())
    """
    from cafe.judging.rubric import ScaleType

    bar = "─" * 60
    parts = [self.show(), "", self._pipeline_line()]
    warns = self._health_warnings()
    if warns:
        parts.append("")
        parts += [f"⚠ {w}" for w in warns]
    if self.attribution is not None:
        parts += ["", bar, "DESCRIPTIVE — means & best configuration", bar,
                  self.attribution.show()]
    if self.ratings is not None:
        from cafe.stats import fit_clmm, fit_effects, fit_logistic

        scale = getattr(getattr(self.ratings, "rubric", None), "scale_type", None)
        if scale == ScaleType.binary:
            log = self.logistic if interactions == 2 else fit_logistic(self.ratings, interactions=interactions)
            parts += ["", bar, "LOGISTIC — binary pass/fail model (log-odds & odds ratios)",
                      bar, log.show()]
        elif scale == ScaleType.numeric:
            eff = self.effects if interactions == 2 else fit_effects(self.ratings, interactions=interactions)
            parts += ["", bar, "LINEAR — Gaussian mixed-effects model (significance & effect sizes)",
                      bar, eff.show()]
        else:  # ordinal (and the default): the linear view + the correct ordinal CLMM
            eff = self.effects if interactions == 2 else fit_effects(self.ratings, interactions=interactions)
            parts += ["", bar, "INFERENTIAL — mixed-effects model (significance & effect sizes)",
                      bar, eff.show()]
            clmm = self.clmm if interactions == 2 else fit_clmm(self.ratings, interactions=interactions)
            parts += ["", bar, "ORDINAL — cumulative-link mixed model (verdicts as ordered categories)",
                      bar, clmm.show()]
    return "\n".join(parts)

rejudge 

rejudge(judge, *, rubric=None, repetitions=1, name=None, concurrency=8, checkpoint_path=None, progress=True)

Score the same answers again — with a different judge, rubric, or number of repetitions — returning a new Evaluation (reusing this one's questions/references; nothing is regenerated). This is the "generate once, judge many ways" path:

free = result.rejudge(cafe.LLMJudge(model=m, preset="single_answer"))
biny = result.rejudge(judge, rubric=cafe.rubrics.CORRECT_PASS_FAIL)
noise = result.rejudge(judge, repetitions=3)   # the judge's own spread

checkpoint_path makes the judging crash-safe/resumable (verdicts appended as they land). Also handy for judge↔judge reliability — cafe.reliability(raters={"a": result, "b": result.rejudge(judge_b)}).

Source code in packages/cafe-core/src/cafe/evaluation.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def rejudge(self, judge: Any, *, rubric: Any = None, repetitions: int = 1,
            name: str | None = None, concurrency: int = 8,
            checkpoint_path: str | None = None, progress: bool = True):
    """Score the **same answers** again — with a different judge, ``rubric``, or
    number of ``repetitions`` — returning a new ``Evaluation`` (reusing this one's
    questions/references; nothing is regenerated). This is the "generate once, judge
    many ways" path:

        free = result.rejudge(cafe.LLMJudge(model=m, preset="single_answer"))
        biny = result.rejudge(judge, rubric=cafe.rubrics.CORRECT_PASS_FAIL)
        noise = result.rejudge(judge, repetitions=3)   # the judge's own spread

    ``checkpoint_path`` makes the judging crash-safe/resumable (verdicts appended as
    they land). Also handy for judge↔judge reliability —
    ``cafe.reliability(raters={"a": result, "b": result.rejudge(judge_b)})``.
    """
    from cafe._async import run_blocking
    from cafe.judging import judge_results
    from cafe.stats import attribute

    rubric = rubric or getattr(self.ratings, "rubric", None)
    if rubric is None:
        raise ValueError("rejudge needs a rubric — this evaluation wasn't judged")
    ratings = run_blocking(lambda: judge_results(
        self.answers, judge, rubric, repetitions=repetitions, concurrency=concurrency,
        questions=self.questions, references=self.references,
        checkpoint_path=checkpoint_path, progress=progress,
    ))
    return Evaluation(
        study_name=name or f"{self.study_name} ({getattr(judge, 'model', 'judge')})",
        answers=self.answers, ratings=ratings, attribution=attribute(ratings),
        questions=self.questions, references=self.references,
    )

judge_stability 

judge_stability()

How consistently the judge scored the same answer across its repetitions — the judge analogue of system replications. Returns a :class:cafe.stats.JudgeStability (per-answer std dev + a summary). Only informative when the judge scored answers more than once (judge_replications on the study, or rejudge(..., repetitions=k))::

noisy = result.rejudge(judge, repetitions=3)
print(noisy.judge_stability().show())
Source code in packages/cafe-core/src/cafe/evaluation.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def judge_stability(self):
    """How consistently the judge scored the **same answer** across its repetitions —
    the judge analogue of system ``replications``. Returns a
    :class:`cafe.stats.JudgeStability` (per-answer std dev + a summary). Only
    informative when the judge scored answers more than once (``judge_replications``
    on the study, or ``rejudge(..., repetitions=k)``)::

        noisy = result.rejudge(judge, repetitions=3)
        print(noisy.judge_stability().show())
    """
    from cafe.stats import judge_stability as _judge_stability

    if self.ratings is None:
        raise ValueError("no ratings to measure — this evaluation wasn't judged")
    return _judge_stability(self.ratings)

pipeline 

pipeline()

The composed pipeline (stage order + levels), observed from the run's traces. Only meaningful for a composed system; returns an empty pipeline otherwise.

Source code in packages/cafe-core/src/cafe/evaluation.py 
282
283
284
285
286
287
def pipeline(self):
    """The composed pipeline (stage order + levels), observed from the run's traces.
    Only meaningful for a composed system; returns an empty pipeline otherwise."""
    from cafe.techniques.composed import pipeline as _pipeline

    return _pipeline(self)

plot 

plot(kind=None, **kwargs)

Plots for this evaluation. No argument → a dashboard of the key plots; or pass a kind: "marginals", "interaction", "configs", "distribution", "effects", "pareto". Returns the matplotlib Figure (dashboard) or Axes (single plot) — .savefig(...) or tweak it freely::

result.plot()                 # everything
result.plot("interaction")    # just one
Source code in packages/cafe-core/src/cafe/evaluation.py 
289
290
291
292
293
294
295
296
297
298
299
300
def plot(self, kind: str | None = None, **kwargs):
    """Plots for this evaluation. No argument → a **dashboard** of the key plots; or
    pass a ``kind``: ``"marginals"``, ``"interaction"``, ``"configs"``,
    ``"distribution"``, ``"effects"``, ``"pareto"``. Returns the matplotlib Figure
    (dashboard) or Axes (single plot) — ``.savefig(...)`` or tweak it freely::

        result.plot()                 # everything
        result.plot("interaction")    # just one
    """
    from cafe.stats.plots import plot as _plot

    return _plot(self, kind, **kwargs)

cafe.Results dataclass

All observations from one study run, plus the factor list for context.

Source code in packages/cafe-core/src/cafe/execution/results.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@dataclass
class Results:
    """All observations from one study run, plus the factor list for context."""

    study_name: str
    factors: list[str]
    observations: list[Observation] = field(default_factory=list)

    def __len__(self) -> int:
        return len(self.observations)

    def __iter__(self):
        return iter(self.observations)

    @property
    def errors(self) -> list[Observation]:
        return [o for o in self.observations if not o.ok]

    def summary(self) -> dict[str, Any]:
        configs = {o.config_id for o in self.observations}
        inputs = {o.input_id for o in self.observations}
        oks = [o for o in self.observations if o.ok]
        timed = [o.elapsed_s for o in oks if o.elapsed_s is not None]
        return {
            "study": self.study_name,
            "n_answers": len(self.observations),
            "n_configs": len(configs),
            "n_inputs": len(inputs),
            "n_errors": len(self.errors),
            "total_compute_s": round(sum(timed), 4) if timed else 0.0,
            "mean_cell_s": round(sum(timed) / len(timed), 4) if timed else None,
        }

    def __repr__(self) -> str:
        s = self.summary()
        return (
            f"Results({self.study_name}: {s['n_answers']} answers, "
            f"{s['n_configs']} configs, {s['n_errors']} errors)"
        )

    @property
    def df(self):
        """A tidy pandas DataFrame of the answers (needs the ``stats`` extra)."""
        import pandas as pd

        return pd.DataFrame(self.to_records())

    def _repr_html_(self) -> str:  # pragma: no cover - notebook display
        try:
            df = self.df
            note = "" if len(df) <= 20 else f"<i>showing 20 of {len(df)} rows</i>"
            factors = ", ".join(self.factors) or "—"
            return (
                f"<b>{self!r}</b><br><small>factor columns: <b>{factors}</b></small>"
                f"{df.head(20).to_html(index=False)}{note}"
            )
        except Exception:
            return f"<b>{self!r}</b>"

    # Column names that the frame reserves for itself; a factor or metadata key
    # colliding with one of these is suffixed to keep both visible.
    _RESERVED = ("input_id", "rep", "output", "elapsed_s", "error", "config_id")

    def to_records(self) -> list[dict[str, Any]]:
        """Flat rows for a DataFrame/CSV: one column per factor (by its own name),
        then the answer and timing, then any metadata the system returned.

        Columns: ``<each factor>``, ``input_id``, ``rep``, ``output``,
        ``elapsed_s``, ``error``, ``<each metadata key>``.
        """
        rows: list[dict[str, Any]] = []
        for o in self.observations:
            row: dict[str, Any] = {}
            for name, value in o.config.items():  # factors first, by their own names
                row[name if name not in self._RESERVED else f"{name}_factor"] = level_label(value)
            row["input_id"] = o.input_id
            row["rep"] = o.rep
            row["output"] = o.output
            row["elapsed_s"] = o.elapsed_s
            row["error"] = o.error
            for key, value in o.metadata.items():
                row[key if key not in row else f"{key}_meta"] = value
            rows.append(row)
        return rows

    def to_jsonl(self, path: str) -> None:
        with open(path, "w", encoding="utf-8") as fh:
            for o in self.observations:
                fh.write(json.dumps(o.to_dict(), default=str) + "\n")

    def to_df(self):  # pragma: no cover - thin pandas adapter
        """Return a pandas DataFrame (requires the ``stats`` extra)."""
        try:
            import pandas as pd
        except ImportError as exc:  # pragma: no cover
            raise ImportError(
                "Results.to_df() needs pandas; install with: pip install 'cafe-core[stats]'"
            ) from exc
        return pd.DataFrame(self.to_records())

df property 

df

A tidy pandas DataFrame of the answers (needs the stats extra).

to_records 

to_records()

Flat rows for a DataFrame/CSV: one column per factor (by its own name), then the answer and timing, then any metadata the system returned.

Columns: <each factor>, input_id, rep, output, elapsed_s, error, <each metadata key>.

Source code in packages/cafe-core/src/cafe/execution/results.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def to_records(self) -> list[dict[str, Any]]:
    """Flat rows for a DataFrame/CSV: one column per factor (by its own name),
    then the answer and timing, then any metadata the system returned.

    Columns: ``<each factor>``, ``input_id``, ``rep``, ``output``,
    ``elapsed_s``, ``error``, ``<each metadata key>``.
    """
    rows: list[dict[str, Any]] = []
    for o in self.observations:
        row: dict[str, Any] = {}
        for name, value in o.config.items():  # factors first, by their own names
            row[name if name not in self._RESERVED else f"{name}_factor"] = level_label(value)
        row["input_id"] = o.input_id
        row["rep"] = o.rep
        row["output"] = o.output
        row["elapsed_s"] = o.elapsed_s
        row["error"] = o.error
        for key, value in o.metadata.items():
            row[key if key not in row else f"{key}_meta"] = value
        rows.append(row)
    return rows

to_df 

to_df()

Return a pandas DataFrame (requires the stats extra).

Source code in packages/cafe-core/src/cafe/execution/results.py 
162
163
164
165
166
167
168
169
170
def to_df(self):  # pragma: no cover - thin pandas adapter
    """Return a pandas DataFrame (requires the ``stats`` extra)."""
    try:
        import pandas as pd
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "Results.to_df() needs pandas; install with: pip install 'cafe-core[stats]'"
        ) from exc
    return pd.DataFrame(self.to_records())

cafe.Observation dataclass

One executed (config x input x replication) cell.

Source code in packages/cafe-core/src/cafe/execution/results.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass
class Observation:
    """One executed (config x input x replication) cell."""

    config: dict[str, Any]
    input_id: str
    rep: int
    output: Any = None
    error: str | None = None
    elapsed_s: float | None = None
    started_at: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    @property
    def config_id(self) -> str:
        return config_id(self.config)

    @property
    def ok(self) -> bool:
        return self.error is None

    def key(self) -> str:
        """Idempotency key: (config, input, replication)."""
        return f"{self.config_id}::{self.input_id}::{self.rep}"

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Observation":
        known = {f for f in cls.__dataclass_fields__}  # type: ignore[attr-defined]
        return cls(**{k: v for k, v in d.items() if k in known})

key 

key()

Idempotency key: (config, input, replication).

Source code in packages/cafe-core/src/cafe/execution/results.py 
59
60
61
def key(self) -> str:
    """Idempotency key: (config, input, replication)."""
    return f"{self.config_id}::{self.input_id}::{self.rep}"

Judging

cafe.LLMJudge

Scores answers with an LLM (any provider, via a LiteLLM model string).

Source code in packages/cafe-core/src/cafe/judging/judge.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class LLMJudge:
    """Scores answers with an LLM (any provider, via a LiteLLM model string)."""

    #: System message framing the judge; override per instance if you like.
    default_system = "You are a strict, fair, impartial evaluator."

    def __init__(
        self,
        model: str,
        *,
        temperature: float = 0.0,
        preset: str = "reference_qa",
        system_prompt: str | None = None,
        prompt_template: str | None = None,
        structured: str | bool = "auto",
    ) -> None:
        if preset not in JUDGE_PRESETS:
            raise ValueError(f"unknown preset {preset!r}; choose from {sorted(JUDGE_PRESETS)}")
        if structured not in (True, False, "auto"):
            raise ValueError(f"structured must be True, False, or 'auto'; got {structured!r}")
        self.model = model
        self.temperature = temperature
        self.preset = preset
        self.system_prompt = system_prompt or self.default_system
        #: Full override of the user prompt (placeholders: {instruction} {question}
        #: {answer} {reference} {scale} {grade} {min} {max}). Wins over the rubric's
        #: template and the preset. Must still elicit a final ``GRADE: <int>`` line.
        self.prompt_template = prompt_template
        if prompt_template is not None:
            from cafe.judging.prompts import check_template_placeholders

            check_template_placeholders(prompt_template, where="judge prompt_template")
        #: Ask for a JSON verdict instead of a GRADE: line — ``True``/``False`` to force,
        #: ``"auto"`` to use it where the model supports it (static map, else one probe).
        self.structured = structured
        self._use_json: bool | None = None  # resolved once by prepare()/score()

    def render_messages(
        self, rubric: Rubric, question: str, answer: str, reference: str | None = None
    ) -> list[dict[str, str]]:
        """The **exact** message list sent to the judge — the system framing plus the
        rubric-derived user prompt. This is the single source of truth shared with
        :meth:`score`, so a preview can never drift from what's actually sent. For a
        human-readable view, use :meth:`preview`."""
        user = build_judge_prompt(
            rubric, question, answer, reference,
            preset=self.preset, template=self.prompt_template,
        )
        return [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user},
        ]

    def preview(
        self, rubric: Rubric, question: str, answer: str, reference: str | None = None
    ) -> str:
        """The **full** judge input exactly as sent — the system message *and* the user
        prompt, labelled ``[SYSTEM]`` / ``[USER]`` (no LLM call). This is the way to see
        everything the judge receives."""
        return "\n\n".join(
            f"[{m['role'].upper()}]\n{m['content']}"
            for m in self.render_messages(rubric, question, answer, reference)
        )

    async def prepare(self) -> bool:
        """Resolve one-time setup before scoring — here, whether to use structured (JSON)
        output for this model. Idempotent and cached; the judging runner calls it **once**
        before the loop so the capability probe never runs per-call."""
        if self._use_json is None:
            if self.structured in (True, False):
                self._use_json = bool(self.structured)
            else:  # "auto"
                from cafe.judging.structured import supports_structured

                self._use_json = await supports_structured(self.model)
        return self._use_json

    async def score(
        self, rubric: Rubric, question: str, answer: str, reference: str | None = None
    ) -> JudgeOutput:
        messages = self.render_messages(rubric, question, answer, reference)
        prompt = messages[-1]["content"]

        if await self.prepare():
            out = await self._score_structured(rubric, messages, prompt)
            if out is not None:  # None ⇒ provider rejected JSON; fall back to a plain call
                return out

        try:
            raw = await complete(self.model, messages, temperature=self.temperature)
        except LLMError as exc:
            return JudgeOutput(None, None, f"judge call failed: {exc}", prompt, None)
        value, numeric, reasoning = parse_verdict(raw, rubric)
        return JudgeOutput(value, numeric, reasoning, prompt, raw)

    async def _score_structured(
        self, rubric: Rubric, messages: list[dict[str, str]], prompt: str
    ) -> JudgeOutput | None:
        """Score by asking for a JSON verdict. Returns ``None`` only if the provider
        rejects ``response_format`` (so the caller falls back to a plain call); a parsed
        result — even an unparseable one — is returned with its raw response for audit."""
        from cafe.judging.structured import JSON_INSTRUCTION, parse_json_verdict

        json_prompt = prompt + JSON_INSTRUCTION.format(grade=rubric.grade_hint())
        json_messages = [messages[0], {"role": "user", "content": json_prompt}]
        try:
            raw = await complete(
                self.model, json_messages,
                temperature=self.temperature, response_format={"type": "json_object"},
            )
        except LLMError:
            return None
        value, numeric, reasoning = parse_json_verdict(raw, rubric)
        if numeric is None:  # valid call but not JSON-with-grade → try the GRADE regex on it
            value, numeric, reasoning = parse_verdict(raw, rubric)
        return JudgeOutput(value, numeric, reasoning, json_prompt, raw)

render_messages 

render_messages(rubric, question, answer, reference=None)

The exact message list sent to the judge — the system framing plus the rubric-derived user prompt. This is the single source of truth shared with :meth:score, so a preview can never drift from what's actually sent. For a human-readable view, use :meth:preview.

Source code in packages/cafe-core/src/cafe/judging/judge.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def render_messages(
    self, rubric: Rubric, question: str, answer: str, reference: str | None = None
) -> list[dict[str, str]]:
    """The **exact** message list sent to the judge — the system framing plus the
    rubric-derived user prompt. This is the single source of truth shared with
    :meth:`score`, so a preview can never drift from what's actually sent. For a
    human-readable view, use :meth:`preview`."""
    user = build_judge_prompt(
        rubric, question, answer, reference,
        preset=self.preset, template=self.prompt_template,
    )
    return [
        {"role": "system", "content": self.system_prompt},
        {"role": "user", "content": user},
    ]

preview 

preview(rubric, question, answer, reference=None)

The full judge input exactly as sent — the system message and the user prompt, labelled [SYSTEM] / [USER] (no LLM call). This is the way to see everything the judge receives.

Source code in packages/cafe-core/src/cafe/judging/judge.py 
82
83
84
85
86
87
88
89
90
91
def preview(
    self, rubric: Rubric, question: str, answer: str, reference: str | None = None
) -> str:
    """The **full** judge input exactly as sent — the system message *and* the user
    prompt, labelled ``[SYSTEM]`` / ``[USER]`` (no LLM call). This is the way to see
    everything the judge receives."""
    return "\n\n".join(
        f"[{m['role'].upper()}]\n{m['content']}"
        for m in self.render_messages(rubric, question, answer, reference)
    )

prepare async 

prepare()

Resolve one-time setup before scoring — here, whether to use structured (JSON) output for this model. Idempotent and cached; the judging runner calls it once before the loop so the capability probe never runs per-call.

Source code in packages/cafe-core/src/cafe/judging/judge.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def prepare(self) -> bool:
    """Resolve one-time setup before scoring — here, whether to use structured (JSON)
    output for this model. Idempotent and cached; the judging runner calls it **once**
    before the loop so the capability probe never runs per-call."""
    if self._use_json is None:
        if self.structured in (True, False):
            self._use_json = bool(self.structured)
        else:  # "auto"
            from cafe.judging.structured import supports_structured

            self._use_json = await supports_structured(self.model)
    return self._use_json

cafe.Judge

Bases: Protocol

Anything that can score an answer. Implement this for a custom/non-LLM judge.

Source code in packages/cafe-core/src/cafe/judging/judge.py 
18
19
20
21
22
23
24
25
26
@runtime_checkable
class Judge(Protocol):
    """Anything that can score an answer. Implement this for a custom/non-LLM judge."""

    model: str

    async def score(
        self, rubric: Rubric, question: str, answer: str, reference: str | None = ...
    ) -> JudgeOutput: ...

cafe.Rubric dataclass

An ordered quality scale + how the judge is prompted.

Set prompt_template to take full control of the judge prompt (placeholders: {instruction}, {question}, {answer}, {reference}, {scale}, {min}, {max}, {grade} — the scale-aware "allowed final grade" hint). Leave it None to use the cited default in :mod:cafe.judging.

Source code in packages/cafe-core/src/cafe/judging/rubric.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@dataclass
class Rubric:
    """An ordered quality scale + how the judge is prompted.

    Set ``prompt_template`` to take full control of the judge prompt (placeholders:
    ``{instruction}``, ``{question}``, ``{answer}``, ``{reference}``, ``{scale}``,
    ``{min}``, ``{max}``, ``{grade}`` — the scale-aware "allowed final grade" hint).
    Leave it ``None`` to use the cited default in :mod:`cafe.judging`.
    """

    name: str
    levels: list[Level]
    scale_type: ScaleType = ScaleType.ordinal
    instruction: str = "Judge how well the ANSWER responds to the QUESTION."
    prompt_template: str | None = None

    def __post_init__(self) -> None:
        if isinstance(self.scale_type, str):
            self.scale_type = ScaleType(self.scale_type)
        if not self.levels:
            raise ValueError("a rubric needs at least two levels")
        if self.prompt_template is not None:
            from cafe.judging.prompts import check_template_placeholders

            check_template_placeholders(self.prompt_template, where=f"rubric {self.name!r} prompt_template")

    def numeric(self, value: object) -> int | None:
        """Map a raw verdict value onto its integer scale point (or None).

        For ``numeric`` scales any integer within ``[min_value, max_value]`` is valid
        (the levels are just anchors); for ordinal/binary the value must match a defined
        level exactly."""
        try:
            iv = int(value)  # type: ignore[arg-type]
        except (TypeError, ValueError):
            return None
        if self.scale_type == ScaleType.numeric:
            return iv if self.min_value <= iv <= self.max_value else None
        return iv if any(lvl.value == iv for lvl in self.levels) else None

    def scale_text(self) -> str:
        """A compact, judge-readable description of the scale."""
        return "\n".join(f"  {lvl.value} = {lvl.label}: {lvl.description}" for lvl in self.levels)

    def grade_hint(self) -> str:
        """How to constrain the judge's final grade, matched to the scale type: a range
        for ``numeric`` (any integer in it is valid — the levels are anchors), or the
        **exact allowed values** for ``ordinal``/``binary`` (only the defined levels are
        valid, and they may be non-contiguous, e.g. 1, 3, 5)."""
        if self.scale_type == ScaleType.numeric:
            return f"an integer from {self.min_value} to {self.max_value}"
        return "exactly one of: " + ", ".join(str(lvl.value) for lvl in self.levels)

    @property
    def min_value(self) -> int:
        return min(lvl.value for lvl in self.levels)

    @property
    def max_value(self) -> int:
        return max(lvl.value for lvl in self.levels)

numeric 

numeric(value)

Map a raw verdict value onto its integer scale point (or None).

For numeric scales any integer within [min_value, max_value] is valid (the levels are just anchors); for ordinal/binary the value must match a defined level exactly.

Source code in packages/cafe-core/src/cafe/judging/rubric.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def numeric(self, value: object) -> int | None:
    """Map a raw verdict value onto its integer scale point (or None).

    For ``numeric`` scales any integer within ``[min_value, max_value]`` is valid
    (the levels are just anchors); for ordinal/binary the value must match a defined
    level exactly."""
    try:
        iv = int(value)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None
    if self.scale_type == ScaleType.numeric:
        return iv if self.min_value <= iv <= self.max_value else None
    return iv if any(lvl.value == iv for lvl in self.levels) else None

scale_text 

scale_text()

A compact, judge-readable description of the scale.

Source code in packages/cafe-core/src/cafe/judging/rubric.py 
77
78
79
def scale_text(self) -> str:
    """A compact, judge-readable description of the scale."""
    return "\n".join(f"  {lvl.value} = {lvl.label}: {lvl.description}" for lvl in self.levels)

grade_hint 

grade_hint()

How to constrain the judge's final grade, matched to the scale type: a range for numeric (any integer in it is valid — the levels are anchors), or the exact allowed values for ordinal/binary (only the defined levels are valid, and they may be non-contiguous, e.g. 1, 3, 5).

Source code in packages/cafe-core/src/cafe/judging/rubric.py 
81
82
83
84
85
86
87
88
def grade_hint(self) -> str:
    """How to constrain the judge's final grade, matched to the scale type: a range
    for ``numeric`` (any integer in it is valid — the levels are anchors), or the
    **exact allowed values** for ``ordinal``/``binary`` (only the defined levels are
    valid, and they may be non-contiguous, e.g. 1, 3, 5)."""
    if self.scale_type == ScaleType.numeric:
        return f"an integer from {self.min_value} to {self.max_value}"
    return "exactly one of: " + ", ".join(str(lvl.value) for lvl in self.levels)

cafe.ScaleType

Bases: str, Enum

How the rubric's numbers should be interpreted (drives the statistics).

Source code in packages/cafe-core/src/cafe/judging/rubric.py 
20
21
22
23
24
25
class ScaleType(str, Enum):
    """How the rubric's numbers should be interpreted (drives the statistics)."""

    ordinal = "ordinal"    # ordered categories, unequal spacing (e.g. 1..5 quality)
    numeric = "numeric"    # interval/ratio score (e.g. 0..10 treated as continuous)
    binary = "binary"      # two outcomes (pass/fail, yes/no)

cafe.Level dataclass

One point on a rubric's ordered scale.

Source code in packages/cafe-core/src/cafe/judging/rubric.py 
28
29
30
31
32
33
34
@dataclass(frozen=True)
class Level:
    """One point on a rubric's ordered scale."""

    value: int
    label: str
    description: str

cafe.rubrics

A small library of ready-made rubrics — grab one, or copy it as a template.

Each is a plain :class:cafe.Rubric; nothing here is special. Use them directly::

import cafe
study = cafe.Study(..., rubric=cafe.rubrics.FAITHFULNESS_1_5, judge=...)

or copy one and edit the levels/instruction for your own criterion. Pick the scale_type deliberately — it decides which statistical model runs (ordinal → CLMM, numeric → linear, binary → logistic).

cafe.judge_results async 

judge_results(results, judge, rubric, *, repetitions=1, concurrency=8, references=None, questions=None, checkpoint_path=None, resume=True, on_progress=None, progress=False)

Judge every successful answer repetitions times.

questions / references map input_id -> text so the judge sees the original question and any gold answer.

checkpoint_path makes the (often expensive) judging phase crash-safe: each verdict is appended as it lands and a resumed run skips already-scored (answer, judge_rep) pairs — the same guarantee :func:cafe.run_study gives the answer phase.

Source code in packages/cafe-core/src/cafe/judging/runner.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def judge_results(
    results: Results,
    judge: Judge,
    rubric: Rubric,
    *,
    repetitions: int = 1,
    concurrency: int = 8,
    references: dict[str, str] | None = None,
    questions: dict[str, str] | None = None,
    checkpoint_path: str | None = None,
    resume: bool = True,
    on_progress: ProgressFn | None = None,
    progress: bool = False,
) -> Ratings:
    """Judge every successful answer ``repetitions`` times.

    ``questions`` / ``references`` map ``input_id`` -> text so the judge sees the
    original question and any gold answer.

    ``checkpoint_path`` makes the (often expensive) judging phase crash-safe: each verdict
    is appended as it lands and a resumed run skips already-scored ``(answer, judge_rep)``
    pairs — the same guarantee :func:`cafe.run_study` gives the answer phase.
    """
    questions = questions or {}
    references = references or {}
    # Resolve one-time judge setup (e.g. structured-output capability) once, before the
    # loop — so a capability probe runs at most once, never per judged answer.
    if hasattr(judge, "prepare"):
        await judge.prepare()

    # An empty-string answer is a valid (bad) answer — judge it (it will score low). Only a
    # genuinely absent output (None) is unjudgeable; record that as an explicit error rating
    # so it is visible in ratings.failures() rather than silently dropped.
    targets = [o for o in results.observations if o.ok and o.output is not None]
    unjudgeable = [o for o in results.observations if o.ok and o.output is None]

    # Resume from a ratings checkpoint if given.
    ckpt = None
    prior: dict[str, Rating] = {}
    if checkpoint_path is not None:
        from cafe.execution.checkpoint import RatingsCheckpoint

        ckpt = RatingsCheckpoint(checkpoint_path)
        if resume:
            prior = ckpt.load()

    def _key(obs_key: str, judge_rep: int) -> str:
        return f"{obs_key}::jr{judge_rep}"

    todo = [(o, jr) for o in targets for jr in range(repetitions)
            if _key(o.key(), jr) not in prior]
    total = len(targets) * repetitions
    ratings: list[Rating] = list(prior.values())
    sem = asyncio.Semaphore(max(1, concurrency))
    lock = asyncio.Lock()
    done = len(prior)

    with progress_bar(total, "judging", enabled=progress and on_progress is None) as bar:
        report = on_progress or bar

        async def one(obs, judge_rep: int) -> None:
            nonlocal done
            question = questions.get(obs.input_id, "")
            reference = references.get(obs.input_id)
            answer = obs.output if isinstance(obs.output, str) else str(obs.output)
            async with sem:
                out = await judge.score(rubric, question, answer, reference)
            rating = Rating(
                obs_key=obs.key(),
                config=dict(obs.config),
                input_id=obs.input_id,
                rep=obs.rep,
                judge_rep=judge_rep,
                value=out.value,
                value_numeric=out.value_numeric,
                reasoning=out.reasoning,
                error=None if out.value_numeric is not None else (out.reasoning or "unparseable"),
                prompt=out.prompt,
                raw_response=out.raw_response,
            )
            async with lock:
                ratings.append(rating)
                if ckpt is not None:
                    ckpt.append(rating)
                done += 1
                if report is not None:
                    report(rating, done, total)

        if todo:
            await asyncio.gather(*(one(o, jr) for o, jr in todo))

    # Explicit error rows for answers that produced no output at all.
    for obs in unjudgeable:
        ratings.append(Rating(
            obs_key=obs.key(), config=dict(obs.config), input_id=obs.input_id,
            rep=obs.rep, judge_rep=0, value=None, value_numeric=None,
            error="unjudgeable: the system produced no output (None)",
        ))

    return Ratings(
        rubric=rubric,
        judge_model=getattr(judge, "model", "judge"),
        factors=list(results.factors),
        items=ratings,
        judge_system_prompt=getattr(judge, "system_prompt", None),
    )

Statistics

cafe.attribute 

attribute(ratings)

Compute per-config and per-factor mean quality from judge ratings.

Judge replications of the same answer are averaged first (see :func:cafe.stats._frame.analysis_frame), so each answer counts once.

Source code in packages/cafe-core/src/cafe/stats/descriptive.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def attribute(ratings: Ratings) -> Attribution:
    """Compute per-config and per-factor mean quality from judge ratings.

    Judge replications of the same answer are averaged first (see
    :func:`cafe.stats._frame.analysis_frame`), so each answer counts once.
    """
    from cafe.stats._frame import analysis_frame

    result = Attribution(n_ratings=len(ratings.items))
    df = analysis_frame(ratings)  # one row per answer (judge replications averaged)
    if df.empty or "verdict" not in df.columns:
        return result
    result.n_usable = len(df)
    df["verdict"] = df["verdict"].astype(float)
    result.overall_mean = float(df["verdict"].mean())

    # Factor columns are exactly the study's factor names (carried on Ratings).
    factor_cols = [f for f in ratings.factors if f in df.columns]
    result.factors = list(factor_cols)

    # Per-configuration means.
    if factor_cols:
        for keys, sub in df.groupby(factor_cols, dropna=False):
            keys = keys if isinstance(keys, tuple) else (keys,)
            config = {col: k for col, k in zip(factor_cols, keys)}
            result.config_means.append(
                {"config": config, "mean": float(sub["verdict"].mean()), "n": int(len(sub))}
            )
        result.best_config = max(result.config_means, key=lambda r: r["mean"])

    # Per-factor marginal means.
    for col in factor_cols:
        for level, sub in df.groupby(col, dropna=True):
            result.factor_marginals.append(
                {
                    "factor": col,
                    "level": str(level),
                    "mean": float(sub["verdict"].mean()),
                    "n": int(len(sub)),
                }
            )
    return result

cafe.fit_effects 

fit_effects(ratings, *, alpha=0.05, interactions=2)

Fit the mixed-effects model and return per-term significance + η².

interactions is the maximum interaction order to include: 1 = main effects only, 2 = also two-way (e.g. model×prompt — the default, since two-way captures most), 3 = up to three-way. It's auto-capped at the number of factors, and falls back to main-effects-only if the interaction model can't be estimated on the data.

Source code in packages/cafe-core/src/cafe/stats/inferential.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def fit_effects(ratings: "Ratings", *, alpha: float = 0.05, interactions: int = 2) -> Effects:
    """Fit the mixed-effects model and return per-term significance + η².

    ``interactions`` is the maximum interaction order to include: ``1`` = main effects
    only, ``2`` = also two-way (e.g. model×prompt — the default, since two-way captures
    most), ``3`` = up to three-way. It's auto-capped at the number of factors, and falls
    back to main-effects-only if the interaction model can't be estimated on the data.
    """
    import numpy as np  # noqa: F401

    from cafe.stats._frame import analysis_frame

    res = Effects(alpha=alpha)
    df = analysis_frame(ratings)  # one row per answer (judge replications averaged)
    if df.empty or "verdict" not in df.columns:
        res.model = "skipped (no verdicts)"
        return res
    df["verdict"] = df["verdict"].astype(float)
    res.n_obs = len(df)

    if df["verdict"].nunique() < 2:
        # Every answer scored identically — there is no variance for any factor to
        # explain (a degenerate fit would otherwise report a spurious effect).
        res.model = "skipped (no variance in verdict — every answer scored the same)"
        res.warnings.append("all verdicts identical; no effect to estimate")
        return res

    usable = [
        c for c in ratings.factors
        if c in df.columns and df[c].astype("string").dropna().nunique() >= 2
    ]
    if not usable:
        res.model = "skipped (no factor varies across ≥2 levels)"
        res.warnings.append("need a factor with at least two observed levels")
        return res

    res.pairwise_d = _pairwise_d(df, usable)

    try:
        import statsmodels.api as sm
        import statsmodels.formula.api as smf
    except ImportError:
        res.model = "one-way ANOVA (statsmodels unavailable)"
        res.terms = _oneway(df, usable, res.warnings, alpha)
        return res

    order = max(1, min(interactions, len(usable)))
    # Fit on collision-proof column names so a factor named like a patsy builtin (C/Q/I)
    # can't shadow the formula function; the display formula + term labels keep real names.
    dfs, safe, _back = _safe_factor_frame(df, usable)
    main_terms = " + ".join(f"C(Q('{safe[c]}'))" for c in usable)

    def _formula(o: int) -> str:
        return f"verdict ~ ({main_terms})**{o}" if o >= 2 and len(usable) >= 2 else f"verdict ~ {main_terms}"

    def _display_formula(o: int) -> str:
        fixed = f"({' + '.join(usable)})^{o}" if o >= 2 and len(usable) >= 2 else " + ".join(usable)
        return f"verdict ~ (1 | input_id) + {fixed}"  # random effect first, human-readable

    res.formula = _display_formula(order)

    # Mixed model needs ≥2 questions to estimate the random intercept.
    if not ("input_id" in df.columns and df["input_id"].nunique() >= 2):
        res.warnings.append("only one question group; using one-way ANOVA")
        res.model = "one-way ANOVA (need ≥2 questions for a random effect)"
        res.terms = _oneway(df, usable, res.warnings, alpha)
        return res

    # statsmodels emits a flurry of singular/convergence warnings on small or
    # near-degenerate data. Capture them quietly and translate to ONE clear note,
    # rather than leaking the raw stack-trace-like spam to the user.
    import warnings as _warnings

    with _warnings.catch_warnings(record=True) as caught:
        _warnings.simplefilter("always")
        # The main-effects mixedlm just confirms the random intercept is identifiable.
        # If it isn't, we keep the interaction formula but drop to fixed-effects ANOVA
        # (rather than one-way, which would silently lose the interaction terms).
        mixed_ok = True
        try:
            mfit = smf.mixedlm(_formula(1), dfs, groups=dfs["input_id"]).fit(
                method="lbfgs", reml=False, disp=False)
            # Random-intercept + residual variance components (paper-reportable).
            try:
                re_var = float(mfit.cov_re.iloc[0, 0])
            except Exception:  # noqa: BLE001
                re_var = float(np.asarray(mfit.cov_re).ravel()[0])
            res.variance_components = {"random_intercept": re_var, "residual": float(mfit.scale)}
        except Exception:  # noqa: BLE001
            mixed_ok = False
            res.warnings.append("random intercept not estimable; using fixed-effects ANOVA")

        # Per-term F / p / partial η² via Type-II ANOVA on the OLS counterpart, at the
        # requested interaction order — falling back to main effects if it won't estimate.
        order_used, table = order, None
        try:
            fit = smf.ols(_formula(order), dfs).fit()
            # A fractional factorial (or any aliased design) makes interaction columns
            # collinear — e.g. at resolution IV model×sabotage IS verbosity×hedge. statsmodels
            # would silently split their variance into meaningless separate terms, so detect
            # the rank deficiency and drop to main effects rather than report bogus interactions.
            if order >= 2:
                exog = fit.model.exog
                if np.linalg.matrix_rank(exog) < exog.shape[1]:
                    order_used = 1
                    res.formula = _display_formula(1)
                    res.warnings.append(
                        "interactions are aliased in this design (e.g. a fractional factorial "
                        "below resolution V) — they can't be separated, so only main effects are "
                        "fit; read the design's alias structure to interpret them"
                    )
                    fit = smf.ols(_formula(1), dfs).fit()
            table = sm.stats.anova_lm(fit, typ=2)
        except Exception:  # noqa: BLE001
            if order_used >= 2:
                order_used = 1
                res.formula = _display_formula(1)
                res.warnings.append("interactions not estimable on this data — main effects only")
                try:
                    table = sm.stats.anova_lm(smf.ols(_formula(1), dfs).fit(), typ=2)
                except Exception as exc:  # noqa: BLE001
                    res.warnings.append(f"ANOVA failed ({exc})")
            else:
                res.warnings.append("ANOVA failed")

        if table is None:
            res.model = "ANOVA failed"
            res.terms = _oneway(df, usable, res.warnings, alpha)
        else:
            # Per-answer residuals / fitted (aligned to the analysis frame rows).
            try:
                res.residuals = [float(x) for x in np.asarray(fit.resid)]
                res.fitted = [float(x) for x in np.asarray(fit.fittedvalues)]
            except Exception:  # noqa: BLE001
                pass
            base = "MixedLM (random intercept: question)" if mixed_ok else "fixed-effects model"
            res.model = (f"{base} + Type-II ANOVA"
                         + (f", up to {order_used}-way" if order_used >= 2 else ""))
            ss_resid = float(table.loc["Residual", "sum_sq"])
            for name in table.index:
                if name in ("Residual", "Intercept"):
                    continue
                cols = [c for c in usable if f"Q('{safe[c]}')" in name]
                if not cols:
                    continue
                ss = float(table.loc[name, "sum_sq"])
                F = float(table.loc[name, "F"])
                p = float(table.loc[name, "PR(>F)"])
                eta = ss / (ss + ss_resid) if (ss + ss_resid) > 0 else None
                res.terms.append(
                    {
                        "factor": " × ".join(cols),
                        "interaction": len(cols) > 1,
                        "df": float(table.loc[name, "df"]),
                        "F": F if F == F else None,
                        "p": p if p == p else None,
                        "partial_eta_sq": eta,
                        "significant": (p < alpha) if p == p else False,
                    }
                )

    if any(_degenerate(w) for w in caught):
        res.warnings.append(
            "model was near-singular (little between-question variance) — treat p-values as "
            "unstable; the effect sizes (Cohen's d) are the more reliable signal here"
        )
    return res

cafe.fit_clmm 

fit_clmm(ratings, *, alpha=0.05, interactions=2, timeout=120)

Fit an ordinal CLMM (verdict ~ factors + (1|question)) via R.

interactions is the maximum interaction order (1 = main effects, 2 = also two-way, …); R falls back to main effects if the interaction model doesn't converge.

Source code in packages/cafe-core/src/cafe/stats/ordinal.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def fit_clmm(
    ratings: "Ratings", *, alpha: float = 0.05, interactions: int = 2, timeout: int = 120
) -> CLMMResult:
    """Fit an ordinal CLMM (``verdict ~ factors + (1|question)``) via R.

    ``interactions`` is the maximum interaction order (1 = main effects, 2 = also two-way,
    …); R falls back to main effects if the interaction model doesn't converge.
    """
    import json
    import os
    import tempfile
    from importlib.resources import files

    from cafe.stats._frame import analysis_frame

    res = CLMMResult(alpha=alpha)

    rubric = getattr(ratings, "rubric", None)
    scale = getattr(getattr(rubric, "scale_type", None), "value", None)
    if scale is not None and scale != "ordinal":
        res.reason = f"CLMM is for ordinal scales; this rubric's scale_type is {scale!r}."
        return res

    df = analysis_frame(ratings)  # one row per answer (judge replications averaged)
    if df.empty or "verdict" not in df.columns:
        res.reason = "no verdicts to fit"
        return res
    df["verdict"] = df["verdict"].astype(int)

    factors = [
        f for f in ratings.factors
        if f in df.columns and df[f].astype("string").dropna().nunique() >= 2
    ]
    if not factors:
        res.reason = "no factor varies across at least two levels"
        return res

    ok, msg = check_r()
    if not ok:
        res.reason = msg
        return res

    keep = ["verdict", "input_id", *factors]
    tmp = tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False)
    try:
        df[keep].to_csv(tmp.name, index=False)
        tmp.close()
        r_script = files("cafe.stats").joinpath("clmm.R")
        order = max(1, min(interactions, len(factors)))
        # Cap the order to what the design can estimate: an aliased fractional factorial is
        # rank-deficient above main effects, so its 2FIs can't be separated (matches the
        # Gaussian layer, so the two reports agree).
        from cafe.stats.inferential import _design_rank_deficient

        if order >= 2 and _design_rank_deficient(df, factors, order):
            order = 1
            res.warnings.append(
                "interactions are aliased in this design (e.g. a fractional factorial below "
                "resolution V) — they can't be separated, so only main effects are fit"
            )
        try:
            proc = subprocess.run(
                ["Rscript", str(r_script), tmp.name, ",".join(factors), str(order)],
                capture_output=True, text=True, timeout=timeout,
            )
        except subprocess.TimeoutExpired:
            res.reason = f"R timed out after {timeout}s"
            return res
    finally:
        try:
            os.unlink(tmp.name)
        except OSError:
            pass

    if proc.returncode != 0:
        res.reason = f"R exited {proc.returncode}: {proc.stderr.strip()[:200]}"
        return res
    try:
        payload = json.loads(proc.stdout)
    except json.JSONDecodeError:
        res.reason = f"R produced invalid JSON: {proc.stdout[:200]!r}"
        return res

    if not payload.get("available"):
        res.error = res.reason = payload.get("error", "R reported the model unavailable")
        return res

    res.available = True
    res.n_obs = payload.get("n_obs")
    res.formula = payload.get("formula")
    res.log_lik = payload.get("logLik")
    res.thresholds = payload.get("thresholds", [])
    coeffs = payload.get("coefficients", [])
    for c in coeffs:
        term = str(c.get("term", ""))
        c["interaction"] = ":" in term  # R names interaction coefficients a:b
        c["factor"] = next((f for f in factors if term.startswith(f)), None)
        c["label"] = _readable_term(term, factors)  # "retrieve=keyword × rerank=none"
        c["significant"] = c.get("p") is not None and c["p"] < alpha
    res.coefficients = coeffs
    return res

cafe.fit_logistic 

fit_logistic(ratings, *, alpha=0.05, interactions=2, backend='auto', timeout=120)

Fit a logistic model of pass/fail (verdict ∈ {0,1}) on the factors.

backend selects the engine: "glmer" (logistic GLMM via R lme4), "statsmodels" (self-contained GEE / logistic), or "auto" (default — GLMM when R+lme4 are present, else statsmodels). interactions is the max interaction order (1 = main effects, 2 = also two-way). Returns a :class:Logistic; available=False with a reason when the rubric isn't binary, there's no variance, or no engine is available.

Source code in packages/cafe-core/src/cafe/stats/logistic.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def fit_logistic(
    ratings: "Ratings", *, alpha: float = 0.05, interactions: int = 2,
    backend: str = "auto", timeout: int = 120,
) -> Logistic:
    """Fit a logistic model of pass/fail (``verdict`` ∈ {0,1}) on the factors.

    ``backend`` selects the engine: ``"glmer"`` (logistic GLMM via R lme4), ``"statsmodels"``
    (self-contained GEE / logistic), or ``"auto"`` (default — GLMM when R+lme4 are present,
    else statsmodels). ``interactions`` is the max interaction order (1 = main effects,
    2 = also two-way). Returns a :class:`Logistic`; ``available=False`` with a ``reason``
    when the rubric isn't binary, there's no variance, or no engine is available.
    """
    from cafe.stats._frame import analysis_frame

    res = Logistic(alpha=alpha)

    rubric = getattr(ratings, "rubric", None)
    scale = getattr(getattr(rubric, "scale_type", None), "value", None)
    if scale is not None and scale != "binary":
        res.reason = f"logistic model is for binary scales; this rubric's scale_type is {scale!r}."
        return res

    df = analysis_frame(ratings)
    if df.empty or "verdict" not in df.columns:
        res.reason = "no verdicts to fit"
        return res
    df["verdict"] = df["verdict"].astype(int)
    res.n_obs = len(df)

    values = set(df["verdict"].unique().tolist())
    if not values <= {0, 1}:
        res.reason = "binary logistic needs 0/1 verdicts"
        return res
    if len(values) < 2:
        only = "pass" if 1 in values else "fail"
        res.reason = f"every answer scored the same ({only}) — no variance to model"
        return res

    usable = [
        f for f in ratings.factors
        if f in df.columns and df[f].astype("string").dropna().nunique() >= 2
    ]
    if not usable:
        res.reason = "no factor varies across at least two levels"
        return res

    # Backend selection: prefer the GLMM (glmer) when available.
    want_glmer = backend in ("auto", "glmer")
    fallback_note = None
    if want_glmer:
        ok, msg = check_glmer()
        if ok:
            glmm = _fit_glmer(df, usable, alpha=alpha, interactions=interactions, timeout=timeout)
            if glmm is not None:
                return glmm
            fallback_note = "R GLMM did not converge; used the statsmodels logistic instead"
        elif backend == "glmer":
            res.reason = f"logistic GLMM (R lme4::glmer) unavailable: {msg}"
            return res
        else:  # auto, R/lme4 missing
            fallback_note = f"logistic GLMM (R lme4) unavailable ({msg}); used the statsmodels logistic"

    res = _fit_statsmodels(df, usable, alpha=alpha, interactions=interactions)
    if fallback_note and res.available:
        res.warnings.insert(0, fallback_note)
    return res

cafe.judge_stability 

judge_stability(ratings)

Measure how consistently a judge scores the same answer across its repetitions.

Groups verdicts by answer (obs_key) and, for each answer with ≥2 usable verdicts, computes the population std dev and the value list. Returns a :class:JudgeStability with per-answer rows and a summary (mean/max sd, and the fraction of answers the judge scored identically every time). Meaningful only when the judge scored answers more than once (judge_replications / repetitions).

Source code in packages/cafe-core/src/cafe/stats/stability.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def judge_stability(ratings: Ratings) -> JudgeStability:
    """Measure how consistently a judge scores the **same answer** across its repetitions.

    Groups verdicts by answer (``obs_key``) and, for each answer with ≥2 usable
    verdicts, computes the population std dev and the value list. Returns a
    :class:`JudgeStability` with per-answer rows and a summary (mean/max sd, and the
    fraction of answers the judge scored identically every time). Meaningful only when
    the judge scored answers more than once (``judge_replications`` / ``repetitions``).
    """
    result = JudgeStability(judge_model=getattr(ratings, "judge_model", "judge"))

    by_answer: dict[str, dict[str, Any]] = {}
    for r in ratings.items:
        if not r.ok:
            continue
        slot = by_answer.setdefault(
            r.obs_key, {"input_id": r.input_id, "config": dict(r.config), "reps": []}
        )
        slot["reps"].append((r.judge_rep, r.value_numeric))

    per_answer = []
    for slot in by_answer.values():
        verdicts = [v for _, v in sorted(slot["reps"])]
        result.judge_reps = max(result.judge_reps, len(verdicts))
        if len(verdicts) < 2:
            continue  # a single verdict has no spread
        sd = statistics.pstdev(verdicts)
        per_answer.append({
            "input_id": slot["input_id"],
            "config": slot["config"],
            "verdicts": verdicts,
            "sd": sd,
            "range": max(verdicts) - min(verdicts),
        })

    result.per_answer = per_answer
    result.n_answers = len(per_answer)
    if per_answer:
        sds = [r["sd"] for r in per_answer]
        result.mean_sd = statistics.fmean(sds)
        result.max_sd = max(sds)
        result.unanimous_frac = sum(1 for s in sds if s == 0) / len(sds)
    return result

cafe.pareto

Multi-objective view: the quality–cost–latency Pareto frontier.

"Best configuration" is rarely one-dimensional — the top-quality config may be the slowest and most expensive. :func:pareto aggregates each configuration's mean quality (from the judge), latency (wall-clock), cost (USD), and tokens, then keeps the non-dominated set: the configs you cannot improve on one objective without losing on another. Everything else is strictly beaten.

Quality is maximized; cost / latency / tokens are minimized. An objective that doesn't vary across configs (e.g. cost is 0 for a local model) is dropped from the dominance test automatically, so the frontier reflects only what actually trades off.

Needs the stats extra (pandas); :meth:ParetoResult.plot needs matplotlib.

ParetoResult dataclass

Per-configuration objective values and which configs are Pareto-optimal.

Source code in packages/cafe-core/src/cafe/stats/pareto.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@dataclass
class ParetoResult:
    """Per-configuration objective values and which configs are Pareto-optimal."""

    objectives: list[str]
    rows: list[dict[str, Any]] = field(default_factory=list)  # config, label, <objectives>, pareto_optimal

    @property
    def frontier(self) -> list[dict[str, Any]]:
        """The non-dominated configurations (the trade-off front)."""
        return [r for r in self.rows if r["pareto_optimal"]]

    def show(self) -> str:
        cols = self.objectives
        head = f"  {'':2}{'configuration':<34}" + "".join(f"{c:>12}" for c in cols)
        lines = [
            f"Pareto frontier over {', '.join(self.objectives)} "
            f"({len(self.frontier)} of {len(self.rows)} configs optimal):",
            "",
            head,
        ]
        for r in sorted(self.rows, key=lambda x: x.get("quality", 0), reverse=True):
            mark = "★ " if r["pareto_optimal"] else "  "
            vals = "".join(f"{r[c]:>12.4g}" for c in cols)
            lines.append(f"  {mark}{r['label']:<34}{vals}")
        lines.append("")
        lines.append("★ = Pareto-optimal (not dominated on every objective by another config)")
        return "\n".join(lines)

    def plot(self, x: str = "cost", y: str = "quality", *, ax: Any = None, annotate: bool = True):
        """Scatter all configs on two objectives, with the frontier highlighted.

        Defaults to quality (y) vs cost (x). Returns the matplotlib Axes.
        """
        import matplotlib.pyplot as plt

        if ax is None:
            _, ax = plt.subplots(figsize=(6, 4.5))

        dom = [r for r in self.rows if not r["pareto_optimal"]]
        front = sorted(self.frontier, key=lambda r: r[x])

        if dom:
            ax.scatter([r[x] for r in dom], [r[y] for r in dom],
                       c="#9aa0a6", s=45, label="dominated", zorder=2)
        ax.plot([r[x] for r in front], [r[y] for r in front],
                "-", color="#f5a623", alpha=0.6, zorder=3)
        ax.scatter([r[x] for r in front], [r[y] for r in front],
                   c="#f5a623", s=90, edgecolors="black", linewidths=0.6,
                   label="Pareto-optimal", zorder=4)

        if annotate:  # only the frontier — annotating every point is unreadable
            from cafe.execution.results import level_label

            keys = list(self.rows[0]["config"]) if self.rows else []
            varying = [k for k in keys if len({str(r["config"].get(k)) for r in self.rows}) > 1]
            for r in front:
                short = "·".join(str(level_label(r["config"][k])).split("/")[-1] for k in varying) or r["label"]
                ax.annotate(short, (r[x], r[y]), fontsize=7,
                            xytext=(4, 4), textcoords="offset points", color="#444")

        ax.set_xlabel(x + ("  (lower is better)" if _DIRECTION.get(x, 1) < 0 else ""))
        ax.set_ylabel(y + ("  (higher is better)" if _DIRECTION.get(y, 1) > 0 else ""))
        ax.set_title(f"Quality vs {x}: Pareto frontier")
        ax.legend(loc="best", fontsize=8)
        ax.grid(True, alpha=0.25)
        return ax

    def _repr_html_(self) -> str:  # pragma: no cover - notebook display
        return f"<pre>{self.show()}</pre>"

    def __repr__(self) -> str:
        return f"ParetoResult({len(self.frontier)}/{len(self.rows)} optimal over {self.objectives})"

frontier property 

frontier

The non-dominated configurations (the trade-off front).

plot 

plot(x='cost', y='quality', *, ax=None, annotate=True)

Scatter all configs on two objectives, with the frontier highlighted.

Defaults to quality (y) vs cost (x). Returns the matplotlib Axes.

Source code in packages/cafe-core/src/cafe/stats/pareto.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def plot(self, x: str = "cost", y: str = "quality", *, ax: Any = None, annotate: bool = True):
    """Scatter all configs on two objectives, with the frontier highlighted.

    Defaults to quality (y) vs cost (x). Returns the matplotlib Axes.
    """
    import matplotlib.pyplot as plt

    if ax is None:
        _, ax = plt.subplots(figsize=(6, 4.5))

    dom = [r for r in self.rows if not r["pareto_optimal"]]
    front = sorted(self.frontier, key=lambda r: r[x])

    if dom:
        ax.scatter([r[x] for r in dom], [r[y] for r in dom],
                   c="#9aa0a6", s=45, label="dominated", zorder=2)
    ax.plot([r[x] for r in front], [r[y] for r in front],
            "-", color="#f5a623", alpha=0.6, zorder=3)
    ax.scatter([r[x] for r in front], [r[y] for r in front],
               c="#f5a623", s=90, edgecolors="black", linewidths=0.6,
               label="Pareto-optimal", zorder=4)

    if annotate:  # only the frontier — annotating every point is unreadable
        from cafe.execution.results import level_label

        keys = list(self.rows[0]["config"]) if self.rows else []
        varying = [k for k in keys if len({str(r["config"].get(k)) for r in self.rows}) > 1]
        for r in front:
            short = "·".join(str(level_label(r["config"][k])).split("/")[-1] for k in varying) or r["label"]
            ax.annotate(short, (r[x], r[y]), fontsize=7,
                        xytext=(4, 4), textcoords="offset points", color="#444")

    ax.set_xlabel(x + ("  (lower is better)" if _DIRECTION.get(x, 1) < 0 else ""))
    ax.set_ylabel(y + ("  (higher is better)" if _DIRECTION.get(y, 1) > 0 else ""))
    ax.set_title(f"Quality vs {x}: Pareto frontier")
    ax.legend(loc="best", fontsize=8)
    ax.grid(True, alpha=0.25)
    return ax

pareto 

pareto(result, *, objectives=None)

Compute the quality–cost–latency Pareto frontier over a study's configurations.

Pass the :class:~cafe.evaluation.Evaluation returned by cafe.evaluate. By default the objectives are quality plus whichever of cost / latency / tokens actually vary across configurations.

Source code in packages/cafe-core/src/cafe/stats/pareto.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def pareto(
    result: "Evaluation",
    *,
    objectives: list[str] | None = None,
) -> ParetoResult:
    """Compute the quality–cost–latency Pareto frontier over a study's configurations.

    Pass the :class:`~cafe.evaluation.Evaluation` returned by ``cafe.evaluate``.
    By default the objectives are quality plus whichever of cost / latency / tokens
    actually vary across configurations.
    """
    import statistics

    ratings = getattr(result, "ratings", None)
    answers = getattr(result, "answers", None)
    if ratings is None or answers is None:
        raise ValueError("pareto() needs an evaluated study (answers + ratings)")

    quality, configs = _quality_by_config(ratings)
    resources = _resources_by_config(answers)

    rows: list[dict[str, Any]] = []
    for label, cfg in configs.items():
        r = resources.get(label, {"latency": [], "cost": [], "tokens": []})
        rows.append({
            "config": cfg,
            "label": label,
            "quality": round(statistics.fmean(quality[label]), 4) if quality[label] else 0.0,
            "latency": round(statistics.fmean(r["latency"]), 4) if r["latency"] else 0.0,
            "cost": round(statistics.fmean(r["cost"]), 6) if r["cost"] else 0.0,
            "tokens": round(statistics.fmean(r["tokens"]), 1) if r["tokens"] else 0.0,
        })

    # Choose objectives: quality always; resource objectives only if they vary.
    if objectives is None:
        objectives = ["quality"]
        for obj in ("cost", "latency", "tokens"):
            values = {r[obj] for r in rows}
            if len(values) > 1:
                objectives.append(obj)

    for r in rows:
        r["pareto_optimal"] = not any(
            _dominates(other, r, objectives) for other in rows if other is not r
        )

    return ParetoResult(objectives=objectives, rows=rows)

Reliability & human ratings

cafe.reliability

Inter-rater reliability — is the judge actually measuring quality?

An LLM judge is only worth trusting if it agrees with humans (and with itself, and other judges). This module computes Krippendorff's α — the standard reliability coefficient that handles any number of raters, missing ratings, and ordinal scales — between any set of raters: judge↔human, human↔human, judge↔judge.

Workflow
  1. sheet = cafe.answer_sheet(evaluation) → one row per answer with a stable answer_id (plus the output to read). Hand it to your experts.
  2. Collect their scores and load them: cafe.human_ratings([{answer_id, rater, score}, ...]).
  3. cafe.reliability(evaluation, human=...) → α overall and per rater pair.

α ≥ 0.80 is conventionally reliable; 0.67–0.80 tentative; below that, the judge and humans don't agree enough to trust the judge's verdicts. Needs the stats extra.

HumanRatings dataclass

Human (or external) scores keyed by answer_id — the human counterpart of judge :class:~cafe.judging.ratings.Ratings.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@dataclass
class HumanRatings:
    """Human (or external) scores keyed by ``answer_id`` — the human counterpart of
    judge :class:`~cafe.judging.ratings.Ratings`."""

    records: list[dict[str, Any]] = field(default_factory=list)  # {answer_id, rater, score}

    def raters(self) -> list[str]:
        return sorted({r["rater"] for r in self.records})

    def by_rater(self) -> dict[str, dict[Any, Any]]:
        out: dict[str, dict[Any, Any]] = defaultdict(dict)
        for r in self.records:
            out[r["rater"]][r["answer_id"]] = r["score"]
        return dict(out)

    def __len__(self) -> int:
        return len(self.records)

Reliability dataclass

Krippendorff's α overall and for each pair of raters.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@dataclass
class Reliability:
    """Krippendorff's α overall and for each pair of raters."""

    alpha: float
    metric: str
    raters: list[str]
    n_units: int
    pairwise: list[dict[str, Any]] = field(default_factory=list)  # {a, b, alpha, n_common}
    note: str = ""

    @staticmethod
    def interpret(alpha: float) -> str:
        if alpha != alpha:  # NaN
            return "n/a"
        if alpha >= 0.80:
            return "reliable"
        if alpha >= 0.667:
            return "tentative"
        return "unreliable"

    def show(self) -> str:
        a0 = "n/a" if self.alpha != self.alpha else f"{self.alpha:.3f}"
        undefined = self.n_units < 2 or self.alpha != self.alpha
        interp = "undefined" if undefined else self.interpret(self.alpha)
        lines = [
            f"inter-rater reliability — Krippendorff's α ({self.metric})",
            f"  raters ({len(self.raters)}): {', '.join(self.raters)}"
            f"      answers rated by ≥2: {self.n_units}",
            "",
            f"  overall α = {a0}   ({interp})",
        ]
        if undefined:
            lines.append(
                f"  note: α needs ≥2 answers each rated by at least two raters "
                f"(got {self.n_units}); the raters here don't share enough scored answers."
            )
        if len(self.pairwise) > 1:  # only interesting with ≥3 raters
            lines.append("")
            lines.append("  pairwise:")
            w = max((len(p["a"]) for p in self.pairwise), default=6)
            for p in self.pairwise:
                a = p["alpha"]
                astr = "n/a" if a != a else f"{a:+.3f}"
                lines.append(f"    {p['a']:>{w}} ↔ {p['b']:<{w}}   α = {astr}   (n={p['n_common']})")
        lines.append("")
        lines.append("  α ≥ 0.80 reliable · 0.667–0.80 tentative · < 0.667 unreliable")
        if self.note:
            lines.append(f"  note: {self.note}")
        return "\n".join(lines)

    def _repr_html_(self) -> str:  # pragma: no cover - notebook display
        return f"<pre>{self.show()}</pre>"

    def __repr__(self) -> str:
        return f"Reliability(α={self.alpha:.3f} [{self.interpret(self.alpha)}], {len(self.raters)} raters)"

krippendorff_alpha 

krippendorff_alpha(table, metric='ordinal')

Krippendorff's α for {rater: {unit: value}}. Missing ratings are fine.

metric is the disagreement metric: "ordinal" (default, for rating scales), "nominal" (categories), or "interval" (numeric distance). Returns NaN if there's nothing pairable.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def krippendorff_alpha(table: dict[str, dict[Any, Any]], metric: str = "ordinal") -> float:
    """Krippendorff's α for ``{rater: {unit: value}}``. Missing ratings are fine.

    ``metric`` is the disagreement metric: ``"ordinal"`` (default, for rating
    scales), ``"nominal"`` (categories), or ``"interval"`` (numeric distance).
    Returns NaN if there's nothing pairable.
    """
    if metric not in _METRICS:
        raise ValueError(f"metric must be one of {_METRICS}; got {metric!r}")

    units: dict[Any, list[Any]] = defaultdict(list)
    for scores in table.values():
        for unit, val in scores.items():
            if val is not None:
                units[unit].append(val)
    pairable = [vs for vs in units.values() if len(vs) >= 2]
    if not pairable:
        return float("nan")

    values = sorted({v for vs in pairable for v in vs})
    idx = {v: i for i, v in enumerate(values)}
    V = len(values)

    # Coincidence matrix: each unit contributes its rating pairs, weighted 1/(m-1).
    o = [[0.0] * V for _ in range(V)]
    for vs in pairable:
        m = len(vs)
        w = 1.0 / (m - 1)
        for a in range(m):
            ia = idx[vs[a]]
            for b in range(m):
                if a != b:
                    o[ia][idx[vs[b]]] += w

    n_c = [sum(row) for row in o]
    n = sum(n_c)
    if n < 2 or V < 2:
        return 1.0  # everyone agrees on a single value → perfectly reliable

    def delta2(a: int, b: int) -> float:
        if metric == "nominal":
            return 0.0 if a == b else 1.0
        if metric == "interval":
            return (values[a] - values[b]) ** 2
        lo, hi = (a, b) if a <= b else (b, a)  # ordinal
        between = sum(n_c[g] for g in range(lo, hi + 1))
        return (between - (n_c[a] + n_c[b]) / 2.0) ** 2

    num = den = 0.0
    for a in range(V):
        for b in range(V):
            d = delta2(a, b)
            num += o[a][b] * d
            den += n_c[a] * n_c[b] * d
    if den == 0:
        return 1.0
    return 1.0 - (n - 1) * num / den

human_ratings 

human_ratings(records)

Build :class:HumanRatings from rows of {answer_id, rater, score}.

Accepts a list of dicts, a pandas DataFrame, or a path to a filled-in CSV (e.g. the sheet from :func:answer_sheet). Rows with a blank score are skipped, so a partially-rated sheet is fine; a row whose score can't be read as a number is skipped with a warning (naming the row) rather than aborting the whole import — a stray header row or a typo like "5?" in one cell shouldn't lose the sheet.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def human_ratings(records: Any) -> HumanRatings:
    """Build :class:`HumanRatings` from rows of ``{answer_id, rater, score}``.

    Accepts a list of dicts, a pandas DataFrame, or a **path to a filled-in CSV**
    (e.g. the sheet from :func:`answer_sheet`). Rows with a blank score are skipped, so a
    partially-rated sheet is fine; a row whose score can't be read as a number is skipped
    with a warning (naming the row) rather than aborting the whole import — a stray header
    row or a typo like ``"5?"`` in one cell shouldn't lose the sheet.
    """
    if isinstance(records, str):  # a CSV path (a filled-in answer sheet)
        import pandas as pd

        records = pd.read_csv(records)
    if hasattr(records, "to_dict"):  # a DataFrame
        records = records.to_dict("records")
    out = []
    for i, rec in enumerate(records):
        missing = {"answer_id", "rater", "score"} - set(rec)
        if missing:
            raise ValueError(f"human rating row {i} is missing {sorted(missing)}: {rec}")
        score = rec["score"]
        if score is None or score == "" or (isinstance(score, float) and score != score):
            continue  # unrated row (blank in the sheet)
        try:
            fscore = float(score)
        except (ValueError, TypeError):
            import warnings

            warnings.warn(
                f"human rating row {i} has a non-numeric score {score!r} "
                f"(answer_id={rec.get('answer_id')!r}, rater={rec.get('rater')!r}) — skipping it.",
                stacklevel=2,
            )
            continue
        out.append({
            "answer_id": rec["answer_id"],
            "rater": str(rec["rater"]),
            "score": int(fscore) if fscore.is_integer() else fscore,
        })
    return HumanRatings(records=out)

answer_sheet 

answer_sheet(evaluation, path=None, *, raters=('expert_1',), questions=None)

A rating sheet: one row per (answer × rater) with a blank score to fill in.

Give it to your experts to rate. Pass path="sheet.csv" to write a CSV they can open in Excel/Sheets, fill the score column, and save; load it back with cafe.human_ratings("sheet.csv"). raters names the columns of experts (one row each). questions maps input_id → question text (defaults to the evaluation's).

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def answer_sheet(
    evaluation: "Evaluation",
    path: str | None = None,
    *,
    raters: tuple[str, ...] = ("expert_1",),
    questions: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
    """A rating sheet: one row per (answer × rater) with a blank ``score`` to fill in.

    Give it to your experts to rate. Pass ``path="sheet.csv"`` to **write a CSV** they can
    open in Excel/Sheets, fill the ``score`` column, and save; load it back with
    ``cafe.human_ratings("sheet.csv")``. ``raters`` names the columns of experts (one row
    each). ``questions`` maps ``input_id`` → question text (defaults to the evaluation's).
    """
    from cafe.execution.results import config_label

    questions = questions if questions is not None else getattr(evaluation, "questions", {}) or {}
    references = getattr(evaluation, "references", {}) or {}
    rows = []
    for o in evaluation.answers.observations:
        if not o.ok:
            continue
        for rater in raters:
            # Column order = what a human reads left→right, then fills `score`:
            # question, the gold reference (same one the judge saw), the answer, then score.
            rows.append({
                "answer_id": o.key(),
                "rater": rater,
                "question": questions.get(o.input_id, ""),
                "reference": references.get(o.input_id, ""),
                "output": o.output,
                "score": "",
                "config": config_label(o.config),
            })
    if path is not None:
        import os

        import pandas as pd

        parent = os.path.dirname(os.path.abspath(path))
        os.makedirs(parent, exist_ok=True)  # so "data/sheet.csv" works even if data/ is new
        pd.DataFrame(rows).to_csv(path, index=False)
    return rows

reliability 

reliability(evaluation=None, human=None, *, raters=None, table=None, metric=None, judge_label='judge')

Inter-rater reliability across judges and/or human raters.

Common uses::

cafe.reliability(evaluation, human=expert_scores)   # judge ↔ humans
cafe.reliability(raters={"120b": result, "20b": result.rejudge(judge_20b)})  # judge ↔ judge
cafe.reliability(human=expert_scores)               # humans ↔ each other

raters maps a display name to a rater source — an Evaluation (its judge's scores) or a {answer_id: score} dict. human is a :class:HumanRatings or rows for :func:human_ratings. The disagreement metric defaults to the rubric's scale.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def reliability(
    evaluation: "Evaluation" | None = None,
    human: Any = None,
    *,
    raters: dict[str, Any] | None = None,
    table: dict[str, dict[Any, Any]] | None = None,
    metric: str | None = None,
    judge_label: str = "judge",
) -> Reliability:
    """Inter-rater reliability across judges and/or human raters.

    Common uses::

        cafe.reliability(evaluation, human=expert_scores)   # judge ↔ humans
        cafe.reliability(raters={"120b": result, "20b": result.rejudge(judge_20b)})  # judge ↔ judge
        cafe.reliability(human=expert_scores)               # humans ↔ each other

    ``raters`` maps a display name to a rater source — an ``Evaluation`` (its judge's
    scores) or a ``{answer_id: score}`` dict. ``human`` is a :class:`HumanRatings` or rows
    for :func:`human_ratings`. The disagreement ``metric`` defaults to the rubric's scale.
    """
    if metric is None:
        metric = _metric_for(evaluation or next((s for s in (raters or {}).values()
                                                 if getattr(s, "answers", None) is not None), None))
    if table is None:
        table = {}
        if raters:
            for name, src in raters.items():
                table[name] = _scores_of(src, judge_label)
        if evaluation is not None and not raters:
            js = _judge_scores(evaluation, judge_label)
            if js:
                table[judge_label] = js
        if human is not None:
            hr = human if isinstance(human, HumanRatings) else human_ratings(human)
            table.update(hr.by_rater())

    if len(table) < 2:
        raise ValueError(
            "reliability needs at least two raters (e.g. the judge + one human, "
            "or two judges); got: " + (", ".join(table) or "none")
        )

    units: dict[Any, int] = defaultdict(int)
    for scores in table.values():
        for u, v in scores.items():
            if v is not None:
                units[u] += 1
    n_units = sum(1 for c in units.values() if c >= 2)

    raters = list(table)
    pairwise = []
    for i in range(len(raters)):
        for j in range(i + 1, len(raters)):
            a, b = raters[i], raters[j]
            common = sum(
                1 for u in set(table[a]) & set(table[b])
                if table[a][u] is not None and table[b][u] is not None
            )
            pairwise.append({
                "a": a, "b": b,
                "alpha": krippendorff_alpha({a: table[a], b: table[b]}, metric),
                "n_common": common,
            })

    return Reliability(
        alpha=krippendorff_alpha(table, metric),
        metric=metric,
        raters=raters,
        n_units=n_units,
        pairwise=pairwise,
    )

ratings_from_human 

ratings_from_human(evaluation, human, *, rubric=None)

Turn human scores into a :class:~cafe.judging.ratings.Ratings, so the whole stats stack (attribute / fit_effects / fit_clmm / report) runs on humans instead of the judge. Each rater's score becomes a verdict; several raters on one answer act like judge replications (averaged before the factor models).

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
def ratings_from_human(evaluation: "Evaluation", human: Any, *, rubric: Any = None):
    """Turn human scores into a :class:`~cafe.judging.ratings.Ratings`, so the whole stats
    stack (``attribute`` / ``fit_effects`` / ``fit_clmm`` / ``report``) runs on **humans**
    instead of the judge. Each rater's score becomes a verdict; several raters on one
    answer act like judge replications (averaged before the factor models)."""
    from cafe.judging.ratings import Rating, Ratings

    hr = human if isinstance(human, HumanRatings) else human_ratings(human)
    by_key = {o.key(): o for o in evaluation.answers.observations}
    if rubric is None:
        rubric = getattr(getattr(evaluation, "ratings", None), "rubric", None)
    if rubric is None:
        raise ValueError("no rubric available (this evaluation has no judge ratings) — pass rubric=")

    grouped: dict[Any, list[tuple[str, Any]]] = defaultdict(list)
    for rec in hr.records:
        grouped[rec["answer_id"]].append((rec["rater"], rec["score"]))

    items = []
    for aid, rs in grouped.items():
        obs = by_key.get(aid)
        if obs is None:
            continue
        for jr, (rater, score) in enumerate(sorted(rs)):
            items.append(Rating(
                obs_key=aid, config=dict(obs.config), input_id=obs.input_id, rep=obs.rep,
                judge_rep=jr, value=score, value_numeric=score, reasoning=f"human:{rater}",
            ))
    return Ratings(rubric=rubric, judge_model="human",
                   factors=list(evaluation.answers.factors), items=items)

human_evaluation 

human_evaluation(evaluation, human, *, rubric=None)

An :class:~cafe.evaluation.Evaluation backed by human ratings — so .report() / .plot() / .effects / .clmm all describe what the humans found, letting you compare it against the judge-backed evaluation side by side.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
def human_evaluation(evaluation: "Evaluation", human: Any, *, rubric: Any = None):
    """An :class:`~cafe.evaluation.Evaluation` backed by **human** ratings — so
    ``.report()`` / ``.plot()`` / ``.effects`` / ``.clmm`` all describe what the *humans*
    found, letting you compare it against the judge-backed evaluation side by side."""
    from cafe.evaluation import Evaluation
    from cafe.stats.descriptive import attribute

    rt = ratings_from_human(evaluation, human, rubric=rubric)
    return Evaluation(
        study_name=f"{evaluation.study_name} (human-rated)",
        answers=evaluation.answers, ratings=rt, attribution=attribute(rt),
        questions=getattr(evaluation, "questions", {}), references=getattr(evaluation, "references", {}),
    )

cafe.human_evaluation 

human_evaluation(evaluation, human, *, rubric=None)

An :class:~cafe.evaluation.Evaluation backed by human ratings — so .report() / .plot() / .effects / .clmm all describe what the humans found, letting you compare it against the judge-backed evaluation side by side.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
def human_evaluation(evaluation: "Evaluation", human: Any, *, rubric: Any = None):
    """An :class:`~cafe.evaluation.Evaluation` backed by **human** ratings — so
    ``.report()`` / ``.plot()`` / ``.effects`` / ``.clmm`` all describe what the *humans*
    found, letting you compare it against the judge-backed evaluation side by side."""
    from cafe.evaluation import Evaluation
    from cafe.stats.descriptive import attribute

    rt = ratings_from_human(evaluation, human, rubric=rubric)
    return Evaluation(
        study_name=f"{evaluation.study_name} (human-rated)",
        answers=evaluation.answers, ratings=rt, attribution=attribute(rt),
        questions=getattr(evaluation, "questions", {}), references=getattr(evaluation, "references", {}),
    )

cafe.human_ratings 

human_ratings(records)

Build :class:HumanRatings from rows of {answer_id, rater, score}.

Accepts a list of dicts, a pandas DataFrame, or a path to a filled-in CSV (e.g. the sheet from :func:answer_sheet). Rows with a blank score are skipped, so a partially-rated sheet is fine; a row whose score can't be read as a number is skipped with a warning (naming the row) rather than aborting the whole import — a stray header row or a typo like "5?" in one cell shouldn't lose the sheet.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def human_ratings(records: Any) -> HumanRatings:
    """Build :class:`HumanRatings` from rows of ``{answer_id, rater, score}``.

    Accepts a list of dicts, a pandas DataFrame, or a **path to a filled-in CSV**
    (e.g. the sheet from :func:`answer_sheet`). Rows with a blank score are skipped, so a
    partially-rated sheet is fine; a row whose score can't be read as a number is skipped
    with a warning (naming the row) rather than aborting the whole import — a stray header
    row or a typo like ``"5?"`` in one cell shouldn't lose the sheet.
    """
    if isinstance(records, str):  # a CSV path (a filled-in answer sheet)
        import pandas as pd

        records = pd.read_csv(records)
    if hasattr(records, "to_dict"):  # a DataFrame
        records = records.to_dict("records")
    out = []
    for i, rec in enumerate(records):
        missing = {"answer_id", "rater", "score"} - set(rec)
        if missing:
            raise ValueError(f"human rating row {i} is missing {sorted(missing)}: {rec}")
        score = rec["score"]
        if score is None or score == "" or (isinstance(score, float) and score != score):
            continue  # unrated row (blank in the sheet)
        try:
            fscore = float(score)
        except (ValueError, TypeError):
            import warnings

            warnings.warn(
                f"human rating row {i} has a non-numeric score {score!r} "
                f"(answer_id={rec.get('answer_id')!r}, rater={rec.get('rater')!r}) — skipping it.",
                stacklevel=2,
            )
            continue
        out.append({
            "answer_id": rec["answer_id"],
            "rater": str(rec["rater"]),
            "score": int(fscore) if fscore.is_integer() else fscore,
        })
    return HumanRatings(records=out)

cafe.answer_sheet 

answer_sheet(evaluation, path=None, *, raters=('expert_1',), questions=None)

A rating sheet: one row per (answer × rater) with a blank score to fill in.

Give it to your experts to rate. Pass path="sheet.csv" to write a CSV they can open in Excel/Sheets, fill the score column, and save; load it back with cafe.human_ratings("sheet.csv"). raters names the columns of experts (one row each). questions maps input_id → question text (defaults to the evaluation's).

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def answer_sheet(
    evaluation: "Evaluation",
    path: str | None = None,
    *,
    raters: tuple[str, ...] = ("expert_1",),
    questions: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
    """A rating sheet: one row per (answer × rater) with a blank ``score`` to fill in.

    Give it to your experts to rate. Pass ``path="sheet.csv"`` to **write a CSV** they can
    open in Excel/Sheets, fill the ``score`` column, and save; load it back with
    ``cafe.human_ratings("sheet.csv")``. ``raters`` names the columns of experts (one row
    each). ``questions`` maps ``input_id`` → question text (defaults to the evaluation's).
    """
    from cafe.execution.results import config_label

    questions = questions if questions is not None else getattr(evaluation, "questions", {}) or {}
    references = getattr(evaluation, "references", {}) or {}
    rows = []
    for o in evaluation.answers.observations:
        if not o.ok:
            continue
        for rater in raters:
            # Column order = what a human reads left→right, then fills `score`:
            # question, the gold reference (same one the judge saw), the answer, then score.
            rows.append({
                "answer_id": o.key(),
                "rater": rater,
                "question": questions.get(o.input_id, ""),
                "reference": references.get(o.input_id, ""),
                "output": o.output,
                "score": "",
                "config": config_label(o.config),
            })
    if path is not None:
        import os

        import pandas as pd

        parent = os.path.dirname(os.path.abspath(path))
        os.makedirs(parent, exist_ok=True)  # so "data/sheet.csv" works even if data/ is new
        pd.DataFrame(rows).to_csv(path, index=False)
    return rows

cafe.krippendorff_alpha 

krippendorff_alpha(table, metric='ordinal')

Krippendorff's α for {rater: {unit: value}}. Missing ratings are fine.

metric is the disagreement metric: "ordinal" (default, for rating scales), "nominal" (categories), or "interval" (numeric distance). Returns NaN if there's nothing pairable.

Source code in packages/cafe-core/src/cafe/stats/reliability.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def krippendorff_alpha(table: dict[str, dict[Any, Any]], metric: str = "ordinal") -> float:
    """Krippendorff's α for ``{rater: {unit: value}}``. Missing ratings are fine.

    ``metric`` is the disagreement metric: ``"ordinal"`` (default, for rating
    scales), ``"nominal"`` (categories), or ``"interval"`` (numeric distance).
    Returns NaN if there's nothing pairable.
    """
    if metric not in _METRICS:
        raise ValueError(f"metric must be one of {_METRICS}; got {metric!r}")

    units: dict[Any, list[Any]] = defaultdict(list)
    for scores in table.values():
        for unit, val in scores.items():
            if val is not None:
                units[unit].append(val)
    pairable = [vs for vs in units.values() if len(vs) >= 2]
    if not pairable:
        return float("nan")

    values = sorted({v for vs in pairable for v in vs})
    idx = {v: i for i, v in enumerate(values)}
    V = len(values)

    # Coincidence matrix: each unit contributes its rating pairs, weighted 1/(m-1).
    o = [[0.0] * V for _ in range(V)]
    for vs in pairable:
        m = len(vs)
        w = 1.0 / (m - 1)
        for a in range(m):
            ia = idx[vs[a]]
            for b in range(m):
                if a != b:
                    o[ia][idx[vs[b]]] += w

    n_c = [sum(row) for row in o]
    n = sum(n_c)
    if n < 2 or V < 2:
        return 1.0  # everyone agrees on a single value → perfectly reliable

    def delta2(a: int, b: int) -> float:
        if metric == "nominal":
            return 0.0 if a == b else 1.0
        if metric == "interval":
            return (values[a] - values[b]) ** 2
        lo, hi = (a, b) if a <= b else (b, a)  # ordinal
        between = sum(n_c[g] for g in range(lo, hi + 1))
        return (between - (n_c[a] + n_c[b]) / 2.0) ** 2

    num = den = 0.0
    for a in range(V):
        for b in range(V):
            d = delta2(a, b)
            num += o[a][b] * d
            den += n_c[a] * n_c[b] * d
    if den == 0:
        return 1.0
    return 1.0 - (n - 1) * num / den

Designs

cafe.full_factorial 

full_factorial(factors)

Every combination of every factor's levels.

Source code in packages/cafe-core/src/cafe/design/factorial.py 
23
24
25
26
27
28
29
def full_factorial(factors: list[Factor]) -> list[Config]:
    """Every combination of every factor's levels."""
    if not factors:
        return [{}]
    names = [f.name for f in factors]
    level_lists = [f.levels for f in factors]
    return [dict(zip(names, combo)) for combo in product(*level_lists)]

cafe.fractional_factorial_design 

fractional_factorial_design(factors, *, runs=None, generators=None)

Build a resolution-maximizing 2^(k-p) fractional factorial for 2-level factors.

runs (a power of two) sets the fraction; default is the smallest design that can place every factor (a saturated, resolution-III screening design). Pass generators ({added_factor: [basic_factor, ...]}) to specify the aliasing yourself instead of letting CAFE choose.

Source code in packages/cafe-core/src/cafe/design/fractional.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def fractional_factorial_design(
    factors: list[Factor],
    *,
    runs: int | None = None,
    generators: dict[str, list[str]] | None = None,
) -> FractionalDesign:
    """Build a resolution-maximizing 2^(k-p) fractional factorial for 2-level factors.

    ``runs`` (a power of two) sets the fraction; default is the smallest design that
    can place every factor (a saturated, resolution-III screening design). Pass
    ``generators`` (``{added_factor: [basic_factor, ...]}``) to specify the aliasing
    yourself instead of letting CAFE choose.
    """
    bad = [f.name for f in factors if len(f.levels) != 2]
    if bad:
        raise ValueError(
            "fractional factorial supports only two-level factors; these are not: "
            f"{bad}. Use the full factorial, or give each factor exactly two levels."
        )
    k = len(factors)
    if k < 3:
        raise ValueError("fractional factorial needs at least 3 factors; use full_factorial")
    full_runs = 2 ** k

    if runs is None:
        b = next(bb for bb in range(2, k + 1) if (2 ** bb) - 1 >= k)
        runs = 2 ** b
    else:
        if runs & (runs - 1) or runs < 4:
            raise ValueError(f"runs must be a power of two ≥ 4; got {runs}")
        b = int(round(math.log2(runs)))
    if runs >= full_runs:
        raise ValueError(
            f"runs={runs} is not a fraction of the full {full_runs}; use full_factorial"
        )

    names = [f.name for f in factors]
    basic, added = names[:b], names[b:]

    # Assign each added factor a generator (a subset of basic factors), largest first
    # to push the defining words as long as possible (higher resolution).
    if generators is None:
        pool = [
            combo for size in range(b, 1, -1)
            for combo in itertools.combinations(range(b), size)
        ]
        if len(added) > len(pool):
            raise ValueError(f"cannot place {k} factors in {runs} runs; increase runs")
        gen_idx = {added[t]: pool[t] for t in range(len(added))}
    else:
        name_to_basic = {n: i for i, n in enumerate(basic)}
        gen_idx = {}
        for f in added:
            if f not in generators:
                raise ValueError(f"no generator given for added factor {f!r}")
            gen_idx[f] = tuple(name_to_basic[g] for g in generators[f])

    fidx = {n: i for i, n in enumerate(names)}

    # Build the design matrix: full factorial over basic factors in ±1, added
    # columns are the products of their generator's basic columns.
    configs: list[Config] = []
    for row in itertools.product((-1, 1), repeat=b):
        signs = {basic[i]: row[i] for i in range(b)}
        for f, gen in gen_idx.items():
            s = 1
            for i in gen:
                s *= row[i]
            signs[f] = s
        cfg = {f.name: f.levels[0 if signs[f.name] < 0 else 1] for f in factors}
        configs.append(cfg)

    # Defining relation + resolution + main-effect aliases.
    words = [frozenset((fidx[f],)) | frozenset(gen_idx[f]) for f in added]
    relation = _full_relation(words)
    resolution = min((len(w) for w in relation), default=None)
    defining_words = [sorted(names[i] for i in w) for w in sorted(relation, key=len)]

    aliases: dict[str, list[str]] = {}
    for f in names:
        fset = frozenset((fidx[f],))
        al = []
        for w in relation:
            term = fset ^ w
            if 1 <= len(term) <= 2:  # report aliasing with main effects / 2FIs
                al.append("".join(sorted(names[i] for i in term)))
        aliases[f] = sorted(set(al))

    return FractionalDesign(
        configs=configs,
        factor_names=names,
        runs=runs,
        full_runs=full_runs,
        resolution=resolution,
        generators={f: [basic[i] for i in gen_idx[f]] for f in added},
        defining_words=defining_words,
        aliases=aliases,
    )

cafe.size 

size(study)

Number of configurations the design will produce (without running it).

Source code in packages/cafe-core/src/cafe/design/factorial.py 
74
75
76
def size(study: Study) -> int:
    """Number of configurations the design will produce (without running it)."""
    return len(generate(study))

Composed pipelines (techniques)

cafe.Pipeline

A scoped collection of techniques + a compose function, usable as a Study system.

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class Pipeline:
    """A scoped collection of techniques + a compose function, usable as a Study ``system``."""

    def __init__(self, compose: ComposeFn | None = None) -> None:
        self._techniques: dict[tuple[str, str], TechniqueSpec] = {}
        self._compose: ComposeFn | None = compose
        #: A label for this system (shown in results); set from the compose fn's name.
        self.name = getattr(compose, "__name__", "pipeline")

    # ── registering techniques ─────────────────────────────────────────────────
    def _spec(self, stage: str, name: str, fn: Callable, description: str, cost_usd: float) -> None:
        params = {
            p.name: p.default
            for p in inspect.signature(fn).parameters.values()
            if p.default is not inspect.Parameter.empty
        }
        # Redefining replaces — like re-`def`-ing a function; makes notebook re-runs painless.
        self._techniques[(stage, name)] = TechniqueSpec(stage, name, fn, params, description, cost_usd)

    def technique(
        self, stage: str, name: str, *, description: str = "", cost_usd: float = 0.0
    ) -> Callable:
        """Register ``fn`` as the technique ``name`` for ``stage`` (decorator).

        ``cost_usd`` is a fixed cost charged every time this technique runs — for a non-LLM
        component whose price CAFE can't see (a paid reranker, a web-search API, a human
        step). It's added to any LLM cost tracked inside the function and shows up in
        ``stage_report`` / Pareto / ``report()``.
        """
        def deco(fn: Callable[..., Any]) -> Callable[..., Any]:
            self._spec(stage, name, fn, description, cost_usd)
            return fn

        return deco

    def add(
        self, stage: str, name: str, fn: Callable, *, description: str = "", cost_usd: float = 0.0
    ) -> "Pipeline":
        """Register a technique programmatically (not as a decorator) — e.g. a deployed
        catalog assembling a pipeline per run. Returns ``self`` for chaining."""
        self._spec(stage, name, fn, description, cost_usd)
        return self

    def compose(self, fn: ComposeFn) -> ComposeFn:
        """Mark ``fn(config, item, ctx)`` as this pipeline's system function (decorator)."""
        self._compose = fn
        self.name = getattr(fn, "__name__", "pipeline")
        return fn

    # ── introspection ──────────────────────────────────────────────────────────
    def names_for(self, stage: str) -> list[str]:
        """Technique names registered under ``stage`` (registration order)."""
        return [n for (s, n) in self._techniques if s == stage]

    def stages(self) -> list[str]:
        """Stages that have at least one registered technique (registration order)."""
        seen: list[str] = []
        for s, _ in self._techniques:
            if s not in seen:
                seen.append(s)
        return seen

    def get(self, stage: str, name: str) -> TechniqueSpec:
        try:
            return self._techniques[(stage, name)]
        except KeyError:
            raise KeyError(
                f"no technique {name!r} for stage {stage!r}; have: {self.names_for(stage)}"
            ) from None

    # ── build a factor from this pipeline's techniques ─────────────────────────
    def factor(
        self,
        stage: str,
        names: list[str] | None = None,
        *,
        none: Any = _NONE_UNSET,
        none_name: str = "none",
        **factor_kwargs: Any,
    ) -> Factor:
        """A categorical factor whose levels are ``stage``'s registered techniques.

        ``pipe.factor("retrieve")`` → ``Factor("retrieve", ["dense", "rerank"])``.

        Pass ``none=`` to add a "skip this stage" level without a no-op technique — the honest
        "does this stage help at all?" contrast (only valid when the stage passes its input
        through, i.e. output type == input type):

        - ``none="chunks"`` → the skip level returns the ``chunks`` input unchanged;
        - ``none=None``     → the skip level contributes nothing / returns ``None``.
        """
        levels = list(names if names is not None else self.names_for(stage))
        if none is not _NONE_UNSET:
            self._register_passthrough(stage, none_name, None if none is None else str(none))
            if none_name not in levels:
                levels.append(none_name)
        if not levels:
            raise ValueError(f"no techniques registered for stage {stage!r} on this pipeline")
        return Factor(stage, levels, **factor_kwargs)

    def _register_passthrough(self, stage: str, name: str, returns: str | None) -> None:
        if (stage, name) in self._techniques:
            return

        async def _passthrough(ctx, **inputs):  # noqa: ANN001, ANN003
            if returns is None:
                return None
            if returns not in inputs:
                raise KeyError(
                    f"skip level {stage!r}/{name!r} passes through {returns!r}, but "
                    f"ctx.run({stage!r}, ...) got inputs {sorted(inputs)}"
                )
            return inputs[returns]

        self._techniques[(stage, name)] = TechniqueSpec(
            stage, name, _passthrough, {}, description=f"skip {stage} (passthrough {returns})"
        )

    # ── act as the Study's system (the System protocol: async run(config, item)) ─
    async def run(self, config: dict[str, Any], item: Any) -> dict[str, Any]:
        """Execute the compose function for one (config, item), returning the answer plus
        per-stage trace + total cost/tokens as metadata (same contract as any system)."""
        if self._compose is None:
            raise ValueError("this Pipeline has no compose function — decorate one with @pipe.compose")
        ctx = Context(self._techniques, config)
        result = await self._compose(config, item, ctx)
        output = result["output"] if isinstance(result, dict) and "output" in result else result
        return {
            "output": output,
            "cost_usd": round(ctx.total_cost, 6),
            "tokens": ctx.total_tokens,
            "trace": ctx.trace,
        }

technique 

technique(stage, name, *, description='', cost_usd=0.0)

Register fn as the technique name for stage (decorator).

cost_usd is a fixed cost charged every time this technique runs — for a non-LLM component whose price CAFE can't see (a paid reranker, a web-search API, a human step). It's added to any LLM cost tracked inside the function and shows up in stage_report / Pareto / report().

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def technique(
    self, stage: str, name: str, *, description: str = "", cost_usd: float = 0.0
) -> Callable:
    """Register ``fn`` as the technique ``name`` for ``stage`` (decorator).

    ``cost_usd`` is a fixed cost charged every time this technique runs — for a non-LLM
    component whose price CAFE can't see (a paid reranker, a web-search API, a human
    step). It's added to any LLM cost tracked inside the function and shows up in
    ``stage_report`` / Pareto / ``report()``.
    """
    def deco(fn: Callable[..., Any]) -> Callable[..., Any]:
        self._spec(stage, name, fn, description, cost_usd)
        return fn

    return deco

add 

add(stage, name, fn, *, description='', cost_usd=0.0)

Register a technique programmatically (not as a decorator) — e.g. a deployed catalog assembling a pipeline per run. Returns self for chaining.

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
73
74
75
76
77
78
79
def add(
    self, stage: str, name: str, fn: Callable, *, description: str = "", cost_usd: float = 0.0
) -> "Pipeline":
    """Register a technique programmatically (not as a decorator) — e.g. a deployed
    catalog assembling a pipeline per run. Returns ``self`` for chaining."""
    self._spec(stage, name, fn, description, cost_usd)
    return self

compose 

compose(fn)

Mark fn(config, item, ctx) as this pipeline's system function (decorator).

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
81
82
83
84
85
def compose(self, fn: ComposeFn) -> ComposeFn:
    """Mark ``fn(config, item, ctx)`` as this pipeline's system function (decorator)."""
    self._compose = fn
    self.name = getattr(fn, "__name__", "pipeline")
    return fn

names_for 

names_for(stage)

Technique names registered under stage (registration order).

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
88
89
90
def names_for(self, stage: str) -> list[str]:
    """Technique names registered under ``stage`` (registration order)."""
    return [n for (s, n) in self._techniques if s == stage]

stages 

stages()

Stages that have at least one registered technique (registration order).

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
92
93
94
95
96
97
98
def stages(self) -> list[str]:
    """Stages that have at least one registered technique (registration order)."""
    seen: list[str] = []
    for s, _ in self._techniques:
        if s not in seen:
            seen.append(s)
    return seen

factor 

factor(stage, names=None, *, none=_NONE_UNSET, none_name='none', **factor_kwargs)

A categorical factor whose levels are stage's registered techniques.

pipe.factor("retrieve") → Factor("retrieve", ["dense", "rerank"]).

Pass none= to add a "skip this stage" level without a no-op technique — the honest "does this stage help at all?" contrast (only valid when the stage passes its input through, i.e. output type == input type):

  • none="chunks" → the skip level returns the chunks input unchanged;
  • none=None → the skip level contributes nothing / returns None.
Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def factor(
    self,
    stage: str,
    names: list[str] | None = None,
    *,
    none: Any = _NONE_UNSET,
    none_name: str = "none",
    **factor_kwargs: Any,
) -> Factor:
    """A categorical factor whose levels are ``stage``'s registered techniques.

    ``pipe.factor("retrieve")`` → ``Factor("retrieve", ["dense", "rerank"])``.

    Pass ``none=`` to add a "skip this stage" level without a no-op technique — the honest
    "does this stage help at all?" contrast (only valid when the stage passes its input
    through, i.e. output type == input type):

    - ``none="chunks"`` → the skip level returns the ``chunks`` input unchanged;
    - ``none=None``     → the skip level contributes nothing / returns ``None``.
    """
    levels = list(names if names is not None else self.names_for(stage))
    if none is not _NONE_UNSET:
        self._register_passthrough(stage, none_name, None if none is None else str(none))
        if none_name not in levels:
            levels.append(none_name)
    if not levels:
        raise ValueError(f"no techniques registered for stage {stage!r} on this pipeline")
    return Factor(stage, levels, **factor_kwargs)

run async 

run(config, item)

Execute the compose function for one (config, item), returning the answer plus per-stage trace + total cost/tokens as metadata (same contract as any system).

Source code in packages/cafe-core/src/cafe/techniques/pipe.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def run(self, config: dict[str, Any], item: Any) -> dict[str, Any]:
    """Execute the compose function for one (config, item), returning the answer plus
    per-stage trace + total cost/tokens as metadata (same contract as any system)."""
    if self._compose is None:
        raise ValueError("this Pipeline has no compose function — decorate one with @pipe.compose")
    ctx = Context(self._techniques, config)
    result = await self._compose(config, item, ctx)
    output = result["output"] if isinstance(result, dict) and "output" in result else result
    return {
        "output": output,
        "cost_usd": round(ctx.total_cost, 6),
        "tokens": ctx.total_tokens,
        "trace": ctx.trace,
    }

cafe.pipeline 

pipeline(source)

Derive the :class:PipelineView (stage order + levels) from a run.

source may be an :class:~cafe.evaluation.Evaluation, a Results, or a Study. Given a study, CAFE runs a single configuration once (cheap) to observe the topology — the order comes from that trace; the levels come from the study's factors. Given an already-run result, no new run happens.

Source code in packages/cafe-core/src/cafe/techniques/composed.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def pipeline(source: Any) -> PipelineView:
    """Derive the :class:`PipelineView` (stage order + levels) from a run.

    ``source`` may be an :class:`~cafe.evaluation.Evaluation`, a ``Results``, or a
    ``Study``. Given a study, CAFE runs a **single configuration once** (cheap) to observe
    the topology — the order comes from that trace; the levels come from the study's
    factors. Given an already-run result, no new run happens.
    """
    from cafe.study import Study

    study = source if isinstance(source, Study) else None
    results = getattr(source, "answers", source)  # Evaluation.answers, or a Results

    traces: list[list[dict[str, Any]]]
    if study is not None:
        # One run (first config × first item) is enough to see the order — no full smoke.
        from cafe.design import generate as _generate
        from cafe.system import as_system, normalize_output

        configs = _generate(study)
        item = study.dataset[0] if study.dataset else None
        if not configs or item is None:
            return PipelineView([])
        raw = study._run_blocking(lambda: as_system(study.system).run(dict(configs[0]), item))
        _, meta = normalize_output(raw)
        traces = [meta.get("trace", [])]
    else:
        traces = [(obs.metadata or {}).get("trace", []) for obs in getattr(results, "observations", [])]

    # Observed order: first appearance of each stage across the traces.
    order: list[str] = []
    observed: dict[str, list[str]] = {}
    for trace in traces:
        for step in trace:
            st = step["stage"]
            if st not in order:
                order.append(st)
                observed[st] = []
            tech = step["technique"]
            if tech not in observed[st]:
                observed[st].append(tech)

    # Levels: prefer the study's declared factor levels for the stage, else the pipeline's
    # own techniques (study.system is a Pipeline), else whatever was observed.
    factor_levels: dict[str, list[Any]] = {}
    pipe = study.system if study is not None else None
    if study is not None:
        for f in study.factors:
            factor_levels[f.name] = list(f.levels)

    stages_out = []
    for st in order:
        pipe_names = pipe.names_for(st) if hasattr(pipe, "names_for") else []
        levels = factor_levels.get(st) or pipe_names or observed.get(st, [])
        stages_out.append({"stage": st, "levels": levels, "techniques_observed": observed.get(st, [])})
    return PipelineView(stages_out)

cafe.stage_report 

stage_report(results)

Aggregate the traces in a Results object into per-stage timing + cost.

Returns one row per (stage, technique): mean elapsed, mean cost, count — the basis for per-stage statistics ("which stage is slowest / most expensive").

Source code in packages/cafe-core/src/cafe/techniques/composed.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def stage_report(results: Any) -> list[dict[str, Any]]:
    """Aggregate the traces in a Results object into per-stage timing + cost.

    Returns one row per (stage, technique): mean elapsed, mean cost, count — the
    basis for per-stage statistics ("which stage is slowest / most expensive").
    """
    from collections import defaultdict

    agg: dict[tuple[str, str], dict[str, float]] = defaultdict(
        lambda: {"elapsed": 0.0, "cost": 0.0, "tokens": 0.0, "n": 0}
    )
    for obs in getattr(results, "observations", []):
        for step in (obs.metadata or {}).get("trace", []):
            if step.get("cached"):
                continue
            a = agg[(step["stage"], step["technique"])]
            a["elapsed"] += step.get("elapsed_s", 0.0)
            a["cost"] += step.get("cost_usd", 0.0)
            a["tokens"] += step.get("tokens", 0)
            a["n"] += 1
    rows = []
    for (stage, tech), a in agg.items():
        n = a["n"] or 1
        rows.append({
            "stage": stage,
            "technique": tech,
            "calls": a["n"],
            "mean_elapsed_s": round(a["elapsed"] / n, 4),
            "mean_cost_usd": round(a["cost"] / n, 6),
            "mean_tokens": round(a["tokens"] / n, 1),
        })
    return sorted(rows, key=lambda r: r["mean_elapsed_s"], reverse=True)

LLM & datasets

cafe.complete async 

complete(model, messages, *, temperature=0.0, timeout=120.0, **kwargs)

Run one chat completion and return the assistant's text content.

Source code in packages/cafe-core/src/cafe/llm.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def complete(
    model: str,
    messages: list[dict[str, str]],
    *,
    temperature: float = 0.0,
    timeout: float = 120.0,
    **kwargs: Any,
) -> str:
    """Run one chat completion and return the assistant's text content."""
    load_env()
    try:
        import litellm
    except ImportError as exc:  # pragma: no cover
        raise LLMError(
            "LLM calls need litellm; install with: pip install 'cafe-core[llm]'"
        ) from exc

    litellm.suppress_debug_info = True
    call = _route(model, dict(messages=messages, temperature=temperature, timeout=timeout, **kwargs))
    try:
        resp = await litellm.acompletion(**call)
    except Exception as exc:  # noqa: BLE001 — surface any provider error uniformly
        raise LLMError(f"{type(exc).__name__}: {exc}") from exc
    _emit_usage(model, resp)
    try:
        return resp.choices[0].message.content or ""
    except (AttributeError, IndexError) as exc:
        raise LLMError(f"unexpected completion payload: {resp!r}") from exc

cafe.set_model_cost 

set_model_cost(model, *, per_1k_tokens=None, per_1k_input=None, per_1k_output=None)

Override the price CAFE uses for model (USD per 1,000 tokens).

Use per_1k_tokens for a single blended rate, or per_1k_input / per_1k_output for separate prompt/completion rates. This takes priority over LiteLLM's automatic pricing — set it for subscriptions, negotiated/enterprise rates, self-hosted models, or providers LiteLLM doesn't price. Pass all-None to clear the override.

Source code in packages/cafe-core/src/cafe/llm.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def set_model_cost(
    model: str,
    *,
    per_1k_tokens: float | None = None,
    per_1k_input: float | None = None,
    per_1k_output: float | None = None,
) -> None:
    """Override the price CAFE uses for ``model`` (USD per 1,000 tokens).

    Use ``per_1k_tokens`` for a single blended rate, or ``per_1k_input`` /
    ``per_1k_output`` for separate prompt/completion rates. This takes priority over
    LiteLLM's automatic pricing — set it for subscriptions, negotiated/enterprise
    rates, self-hosted models, or providers LiteLLM doesn't price. Pass all-None to
    clear the override.
    """
    if per_1k_tokens is None and per_1k_input is None and per_1k_output is None:
        _model_prices.pop(model, None)
        return
    for name, val in (("per_1k_tokens", per_1k_tokens), ("per_1k_input", per_1k_input),
                      ("per_1k_output", per_1k_output)):
        if val is not None and val < 0:
            raise ValueError(f"{name} must be non-negative; got {val}")
    inp = per_1k_input if per_1k_input is not None else (per_1k_tokens or 0.0)
    out = per_1k_output if per_1k_output is not None else (per_1k_tokens or 0.0)
    _model_prices[model] = {"input": inp, "output": out}

cafe.datasets

Load public benchmarks as CAFE datasets.

Each loader returns a list of items shaped for Study(dataset=...): {"id", "text", "reference", ...} — so the reference-guided judge has a gold answer to score against. Needs the datasets extra (HuggingFace datasets): pip install 'cafe-core[datasets]'.

load_truthfulqa 

load_truthfulqa(n=20, *, split='train', categories=None, shuffle=True, seed=0)

TruthfulQA — adversarial questions that elicit common misconceptions.

Reference-based: each item carries the gold Best Answer. Returns items {id, text, reference, category}. categories filters by Category (e.g. ["Misconceptions", "Law"]).

Source code in packages/cafe-core/src/cafe/datasets.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def load_truthfulqa(
    n: int | None = 20,
    *,
    split: str = "train",
    categories: list[str] | None = None,
    shuffle: bool = True,
    seed: int = 0,
) -> list[dict[str, Any]]:
    """TruthfulQA — adversarial questions that elicit common misconceptions.

    Reference-based: each item carries the gold ``Best Answer``. Returns items
    ``{id, text, reference, category}``. ``categories`` filters by Category
    (e.g. ``["Misconceptions", "Law"]``).
    """
    ds = _load_hf("domenicrosati/TruthfulQA", split=split)
    rows = list(ds)
    if categories:
        wanted = {c.lower() for c in categories}
        rows = [r for r in rows if str(r.get("Category", "")).lower() in wanted]
    rows = _sample(rows, n, shuffle=shuffle, seed=seed)
    return [
        {
            "id": f"tq{i}",
            "text": r["Question"],
            "reference": r["Best Answer"],
            "category": r.get("Category"),
        }
        for i, r in enumerate(rows)
    ]

load_gsm8k 

load_gsm8k(n=20, *, split='test', shuffle=True, seed=0)

GSM8K — grade-school math word problems with an exact numeric reference.

Source code in packages/cafe-core/src/cafe/datasets.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def load_gsm8k(
    n: int | None = 20,
    *,
    split: str = "test",
    shuffle: bool = True,
    seed: int = 0,
) -> list[dict[str, Any]]:
    """GSM8K — grade-school math word problems with an exact numeric reference."""
    ds = _load_hf("openai/gsm8k", "main", split=split)
    rows = _sample(list(ds), n, shuffle=shuffle, seed=seed)
    return [
        {"id": f"gsm{i}", "text": r["question"], "reference": r["answer"].split("####")[-1].strip()}
        for i, r in enumerate(rows)
    ]