Create an algorithm benchmark

Define a dataset, score candidate outputs against reference answers and rank algorithm versions.

A benchmark freezes a dataset of test inputs and a scoring rule so you can compare algorithm versions fairly. Every evaluation runs three ordinary Program stages—candidate, evaluator, and reducer—on your existing workers.

1. Run the example benchmark

Run compute/platform/examples/benchmarks/publish.py with the text example, which runs on any CPU worker without a model checkpoint:

Tip: Using a deployed account login instead of the local dev stack? Omit --config .local-dev/compute/client.json.

The script publishes the Programs, dataset, and benchmark definition, runs the evaluation, and prints the summary JSON:

2. Write the 3 benchmark functions

Every benchmark uses three functions (compute/platform/examples/benchmarks/jobs.py):

Candidate (label) — Takes one dataset item (Text) and returns a prediction (Label).
Evaluator (score_label) — Receives EvaluationInput (the candidate output and reference answer) and returns ItemResult for that item.
Reducer (reduce_labels) — Receives ReductionInput across all items via outcomes(request) and returns the aggregate Summary.

python

class Text(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    text: str

class Label(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    label: str

def label(request: Text) -> Label:    return Label(label=request.text[:1].lower())

def score_label(request: EvaluationInput) -> ItemResult:    correct = (        request.candidate.state == "succeeded"        and request.candidate.value["label"] == request.item.reference["label"]    )    return ItemResult(        outcome="answered" if request.candidate.state == "succeeded" else "candidate_failed",        metrics={"correct": int(correct)},        evidence={"correct": correct},    )

def reduce_labels(request: ReductionInput) -> Summary:    items = outcomes(request)    correct = sum(int(item.result.metrics["correct"]) for item in items)    return Summary(        metrics={"match_rate": correct / len(items), "mismatches": len(items) - correct},        annotations={},        evidence={"items": [item.id for item in items]},    )

Because candidate execution and scoring are separate Programs, your candidate can run on a GPU worker while score_label and reduce_labels run on a cheap CPU pool.

3. Score candidate failures explicitly

What happens when a candidate crashes on a difficult input? Run the numbers benchmark to see how candidate failures are scored:

In compute/platform/examples/benchmarks/jobs.py, the second item intentionally raises ValueError("synthetic candidate failure"):

score_number checks request.candidate.state == "failed", assigns the configured failure_penalty (100), and marks the item outcome="candidate_failed".
reduce_numbers counts failed_items alongside rmse so a candidate that crashes on hard cases cannot hide behind a low error on easy cases.

python

"""Two small benchmark families using identical platform execution and publication paths."""
from __future__ import annotations
import jsonimport mathfrom pathlib import Path
from pydantic import BaseModel, ConfigDict
from chemistry_compute.benchmarks import (    EvaluationInput,    ItemResult,    ReductionInput,    Summary,    outcomes,)

class Number(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    value: float    fail: bool = False

class Prediction(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    prediction: float

class Text(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    text: str

class Label(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    label: str

def predict(request: Number) -> Prediction:    if request.fail:        raise ValueError("synthetic candidate failure")    return Prediction(prediction=request.value * 2)

def label(request: Text) -> Label:    return Label(label=request.text[:1].lower())

def score_number(request: EvaluationInput) -> ItemResult:    failed = request.candidate.state == "failed"    error = (        float(request.context.protocol["failure_penalty"])        if failed        else (            float(request.candidate.value["prediction"]) - float(request.item.reference["target"])        )        ** 2    )    Path("results/item.json").write_text(json.dumps({"squared_error": error}))    return ItemResult(        outcome="candidate_failed" if failed else "answered",        metrics={"squared_error": error},        evidence={"squared_error": error},    )

def reduce_numbers(request: ReductionInput) -> Summary:    items = outcomes(request)    if any(        "results/item.json" not in item.artifacts or not item.artifacts["results/item.json"].is_file()        for item in items    ):        raise ValueError("reducer did not receive every accepted scoring artifact")    squared = sum(float(item.result.metrics["squared_error"]) for item in items)    return Summary(        metrics={            "rmse": math.sqrt(squared / len(items)),            "failed_items": sum(item.result.outcome != "answered" for item in items),        },        annotations={},        evidence={"items": [item.id for item in items]},    )

def score_label(request: EvaluationInput) -> ItemResult:    correct = (        request.candidate.state == "succeeded"        and request.candidate.value["label"] == request.item.reference["label"]    )    return ItemResult(        outcome="answered" if request.candidate.state == "succeeded" else "candidate_failed",        metrics={"correct": int(correct)},        evidence={"correct": correct},    )

def reduce_labels(request: ReductionInput) -> Summary:    items = outcomes(request)    correct = sum(int(item.result.metrics["correct"]) for item in items)    return Summary(        metrics={"match_rate": correct / len(items), "mismatches": len(items) - correct},        annotations={},        evidence={"items": [item.id for item in items]},    )

4. Inspect results and compare versions

Use ct-cli benchmark to list benchmarks, check an evaluation's status, or inspect item-level artifacts:

When you modify your algorithm, publish a new candidate version (--version 2) and run it against the same dataset and benchmark version to compare metrics directly. See Test parsers and TS strategies for the built-in chemistry benchmarks and Measure throughput and efficiency to track runtime alongside accuracy.