# 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: ```bash compute/platform/.venv/bin/python compute/platform/examples/benchmarks/publish.py text \ --config .local-dev/compute/client.json ``` > **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: ```text Evaluation 0195f3e0-1111-7000-8000-000000000001 {"benchmark":"...","dataset":"...","candidate":"...","evaluation":"...","summary":{"metrics":{"match_rate":1.0,"mismatches":0},...}} ``` ## 2. Write the 3 benchmark functions Every benchmark uses three functions (`compute/platform/examples/benchmarks/jobs.py`): 1. **Candidate (`label`)** — Takes one dataset item (`Text`) and returns a prediction (`Label`). 2. **Evaluator (`score_label`)** — Receives `EvaluationInput` (the candidate output and reference answer) and returns `ItemResult` for that item. 3. **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: ```bash compute/platform/.venv/bin/python compute/platform/examples/benchmarks/publish.py numbers \ --config .local-dev/compute/client.json ``` 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 json import math from 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: ```bash ct-cli benchmark list ct-cli dataset list ct-cli benchmark status EVALUATION_UUID ct-cli benchmark evidence EVALUATION_UUID ``` 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](/test/algorithms/) for the built-in chemistry benchmarks and [Measure throughput and efficiency](/test/throughput/) to track runtime alongside accuracy.