Python Scorers

When no built-in scorer expresses your notion of correctness, write it in Python. There are two types, and the difference is what the function receives:

Type	Called	Use for
`python`	Once per sample	Scoring each row on its own. The common case.
`python_all_samples`	Once, with every sample	Scores that depend on the whole set - deduplication, ranking, relative comparisons.

Prefer a built-in scorer where one fits: built-ins are validated when the task is created, whereas a snippet fails at run time, on the fifth sample, after you have already paid for four model calls.

Per-Sample Scorer

The snippet defines compute_scores and returns a dictionary of score values. Both def and async def work.

tasks/geography/score.py

from __future__ import annotations

from typing import Any

from latticeflow.core.dtypes import RawSample
from latticeflow.core.dtypes import SolverOutput


def compute_scores(sample: RawSample, solver_output: SolverOutput) -> dict[str, Any]:
    prediction = (solver_output.trace.get_last_assistant_text() or "").strip().lower()
    expected = sample["capital"].lower()
    return {
        "is_correct": prediction == expected,
        "prediction": prediction,
    }

tasks/geography/task.yaml

definition:
  scorers:
    - key: capital_city
      type: python
      compute_scores_snippet: !include "./score.py"
      metrics:
        - type: mean
          field: is_correct
          name: "Accuracy"

Accepted Signatures

Argument names are fixed and only these four combinations are accepted:

Signature	Use
`compute_scores(sample)`	Dataset-only tasks - scoring the data itself, no model involved.
`compute_scores(sample, solver_output)`	The common case.
`compute_scores(sample, task_config)`	Dataset-only, with the task’s resolved configuration.
`compute_scores(sample, solver_output, task_config)`	Everything.

Important

The names must match exactly. compute_scores(sample, output) is rejected before the task runs, and asking for solver_output in a task that has no solver fails with an explicit error.

task_config gives you the resolved configuration parameters, including any models referenced by a model config parameter - so a Python scorer can call a judge model itself when the built-in judges are not enough.

Return Shapes

Two shapes are accepted:

Flat - every key becomes a score value:

return {"is_correct": True, "edit_distance": 3}

Structured - separate the numbers from the explanation:

return {
    "scores": {"is_correct": True},
    "metadata": {"reason": "Matched after normalising whitespace."},
}

Use the structured shape when you want a human-readable justification in the evidence without it becoming something metrics aggregate over. metadata is recorded per sample and shown alongside the scores; only scores is available to metrics and to action rule filters.

Note

The two shapes are distinguished structurally: a returned dict whose keys are exactly scores (and optionally metadata) is read as structured, anything else as flat. So a flat scorer must not name one of its score fields scores.

Returning anything other than a dict fails the sample with The Python snippet must return a dict….

Raising Instead of Scoring

Raising an exception marks that sample as failed rather than scoring it zero - which is what you want when the result is undeterminable rather than wrong. A timeout tells you nothing about the model’s correctness, so recording a 0 would quietly bias the metric downwards.

def compute_scores(sample, solver_output):
    outcome = solver_output.trace.items[-1].content
    if outcome.get("exception_type") == "ReadTimeout":
        raise ValueError(
            "The endpoint timed out; cannot determine whether the input was rejected. "
            f"Original error: {outcome['reason']}"
        )
    return {"scores": {"rejected": 1 if outcome["rejected"] else 0}}

Failed samples are reported separately in the evaluation results, so the distinction stays visible.

All-Samples Scorer

python_all_samples receives the full list and must return one score dictionary per sample, in the same order.

tasks/diversity/score_all.py

from __future__ import annotations

from typing import Any

from latticeflow.core.dtypes import RawSample


def compute_scores(samples: list[RawSample]) -> list[dict[str, Any]]:
    seen: set[str] = set()
    results: list[dict[str, Any]] = []
    for sample in samples:
        answer = sample["answer"].strip().lower()
        results.append({"is_novel": answer not in seen})
        seen.add(answer)
    return results

Accepted signatures are compute_scores(samples) and compute_scores(samples, task_config). Returning anything other than a list, or a list of the wrong length, fails the task.

Warning

An all-samples scorer holds the entire evaluation in memory at once and cannot be parallelised across samples. Use it only when the score genuinely depends on other samples - otherwise python is faster and its failures are isolated to one row.

Reading Solver Output

What solver_output looks like depends on the solver:

Solver	Read it as
`single_turn_solver`, `multi_turn_solver`, `pass_through_solver`, `python`	`solver_output.trace` - use `get_last_assistant_text()`, `items`, `turns`, `function_calls`.
`grouped_single_turn_solver`	`solver_output.solver_outputs` - a dict or list of outputs, each with its own `.trace`.

# Grouped solver: compare each perturbation against the baseline.
def compute_scores(sample, solver_output):
    outputs = solver_output.solver_outputs
    baseline = outputs["base"].trace.get_last_assistant_text()
    perturbed = [
        output.trace.get_last_assistant_text()
        for key, output in outputs.items()
        if key.startswith("perturbation_")
    ]
    return {"num_unchanged": sum(1 for text in perturbed if text == baseline)}

Note

Reach for trace.items rather than trace.turns when the conversation may not start with a user message - an imported trace with only a system prompt and an assistant reply keeps everything in the trace preamble, leaving turns and get_last_assistant_text() empty. See Core Types: Traces.

Runtime

Snippets run server-side in a fixed Python runtime with a fixed library set; they cannot install packages. See Python Snippets for what is available.

Configuration parameters are substituted textually before the snippet runs, so keep placeholders inside string literals:

threshold = float("<< config.similarity_threshold >>")

Field-level reference: Python Scorer and Python Batch Scorer.