Skip to content

Benchmark API

decision_bench.benchmark.Benchmark

A named collection of tasks whose datasets remain independently pinned.

Source code in src/decision_bench/benchmark.py 
 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
class Benchmark:
    """A named collection of tasks whose datasets remain independently pinned."""

    def __init__(
        self,
        *,
        spec: BenchmarkSpec,
        tasks: tuple[DecisionTask, ...],
        project_root: Path,
        spec_path: Path | None = None,
    ) -> None:
        self.spec = spec
        self.tasks = tasks
        self.project_root = project_root
        self.spec_path = spec_path

    @cached_property
    def examples(self) -> tuple[DecisionExample, ...]:
        """Load each immutable dataset once, then materialize its member tasks."""

        grouped: dict[
            tuple[str, str, str | None, str | None, str],
            list[DecisionTask],
        ] = defaultdict(list)
        dataset_by_key: dict[
            tuple[str, str, str | None, str | None, str],
            DatasetSpec,
        ] = {}
        for task in self.tasks:
            dataset = task.metadata.dataset
            grouped[dataset.cache_key].append(task)
            dataset_by_key[dataset.cache_key] = dataset

        examples: list[DecisionExample] = []
        for key, member_tasks in grouped.items():
            rows = tuple(
                dict(row)
                for row in load_rows(dataset_by_key[key], project_root=self.project_root)
            )
            for task in member_tasks:
                examples.extend(task.load_data(rows=rows))

        row_ids = [example.row_id for example in examples]
        if len(row_ids) != len(set(row_ids)):
            raise ValueError(f"benchmark {self.spec.name!r} contains duplicate row IDs")
        return tuple(examples)

    @property
    def datasets(self) -> tuple[DatasetSpec, ...]:
        """Return the benchmark's unique immutable datasets in task order."""

        unique: dict[tuple[str, str, str | None, str | None, str], DatasetSpec] = {}
        for task in self.tasks:
            dataset = task.metadata.dataset
            unique.setdefault(dataset.cache_key, dataset)
        return tuple(unique.values())

    def select(
        self,
        *,
        task_name: str | None = None,
        family: str | None = None,
        domain: str | None = None,
        primitive: Primitive | str | None = None,
    ) -> Self:
        """Return a task-filtered benchmark without changing task identity."""

        normalized_primitive = Primitive(primitive) if primitive is not None else None
        selected = tuple(
            task
            for task in self.tasks
            if (task_name is None or task.metadata.name == task_name)
            and (family is None or task.metadata.family == family)
            and (domain is None or task.metadata.domain == domain)
            and (
                normalized_primitive is None
                or task.metadata.primitive is normalized_primitive
            )
        )
        return type(self)(
            spec=self.spec,
            tasks=selected,
            project_root=self.project_root,
            spec_path=self.spec_path,
        )

datasets property

Return the benchmark's unique immutable datasets in task order.

examples cached property

Load each immutable dataset once, then materialize its member tasks.

select(*, task_name=None, family=None, domain=None, primitive=None)

Return a task-filtered benchmark without changing task identity.

Source code in src/decision_bench/benchmark.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
128
129
130
def select(
    self,
    *,
    task_name: str | None = None,
    family: str | None = None,
    domain: str | None = None,
    primitive: Primitive | str | None = None,
) -> Self:
    """Return a task-filtered benchmark without changing task identity."""

    normalized_primitive = Primitive(primitive) if primitive is not None else None
    selected = tuple(
        task
        for task in self.tasks
        if (task_name is None or task.metadata.name == task_name)
        and (family is None or task.metadata.family == family)
        and (domain is None or task.metadata.domain == domain)
        and (
            normalized_primitive is None
            or task.metadata.primitive is normalized_primitive
        )
    )
    return type(self)(
        spec=self.spec,
        tasks=selected,
        project_root=self.project_root,
        spec_path=self.spec_path,
    )

decision_bench.benchmark.BenchmarkSpec

Bases: BaseModel

Describe a named, immutable collection of registered tasks.

Source code in src/decision_bench/benchmark.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class BenchmarkSpec(BaseModel):
    """Describe a named, immutable collection of registered tasks."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    name: str = Field(min_length=1)
    version: str = Field(min_length=1)
    description: str = Field(min_length=1)
    languages: tuple[str, ...] = Field(min_length=1)
    tasks: tuple[str, ...] = Field(min_length=1)
    reference: str | None = None
    citation: str | None = None

    @field_validator("tasks")
    @classmethod
    def task_names_are_unique(cls, names: tuple[str, ...]) -> tuple[str, ...]:
        if len(names) != len(set(names)):
            raise ValueError("benchmark task names must be unique")
        return names

decision_bench.benchmark.get_benchmark(benchmark='DecisionBench', *, project_root=None)

Resolve a registered name, TOML specification, or BenchmarkSpec.

Source code in src/decision_bench/benchmark.py 
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
def get_benchmark(
    benchmark: str | Path | BenchmarkSpec = "DecisionBench",
    *,
    project_root: Path | None = None,
) -> Benchmark:
    """Resolve a registered name, TOML specification, or BenchmarkSpec."""

    spec_path: Path | None = None
    if isinstance(benchmark, BenchmarkSpec):
        spec = benchmark
    elif isinstance(benchmark, Path) or Path(benchmark).suffix == ".toml":
        spec_path = Path(benchmark)
        spec = load_benchmark_spec(spec_path)
    else:
        try:
            spec = _BENCHMARK_REGISTRY[benchmark]
        except KeyError as error:
            available = ", ".join(sorted(_BENCHMARK_REGISTRY))
            raise KeyError(
                f"unknown benchmark {benchmark!r}; available benchmarks: {available}"
            ) from error

    root = project_root if project_root is not None else Path.cwd()
    return Benchmark(
        spec=spec,
        tasks=tuple(get_tasks(spec.tasks)),
        project_root=root,
        spec_path=spec_path,
    )

decision_bench.task_spec.DecisionTask

One independently versioned dataset-backed evaluation task.

Contributors normally subclass this class, declare metadata, and only override :meth:dataset_transform when their source dataset does not already use the :class:DecisionExample schema.

Source code in src/decision_bench/task_spec.py 
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
class DecisionTask:
    """One independently versioned dataset-backed evaluation task.

    Contributors normally subclass this class, declare ``metadata``, and only
    override :meth:`dataset_transform` when their source dataset does not
    already use the :class:`DecisionExample` schema.
    """

    metadata: TaskMetadata

    def dataset_transform(self, row: Mapping[str, Any]) -> Mapping[str, Any]:
        """Transform one source row into the normalized DecisionExample shape."""

        return decode_storage_row(dict(row))

    def includes(self, example: DecisionExample) -> bool:
        """Return whether a normalized row belongs to this task."""

        return (
            example.task_name == (self.metadata.source_task_name or self.metadata.name)
            and example.primitive is self.metadata.primitive
            and example.family == self.metadata.family
            and example.domain == self.metadata.domain
        )

    def load_data(
        self,
        *,
        project_root: Path | None = None,
        rows: Iterable[Mapping[str, Any] | object] | None = None,
    ) -> tuple[DecisionExample, ...]:
        """Load, transform, filter, and validate this task's evaluation rows."""

        source_rows = (
            load_rows(self.metadata.dataset, project_root=project_root)
            if rows is None
            else rows
        )
        examples: list[DecisionExample] = []
        for row in source_rows:
            if not isinstance(row, Mapping):
                raise TypeError(f"dataset row is not a mapping: {type(row)!r}")
            transformed = self.dataset_transform(row)
            example = DecisionExample.model_validate(transformed)
            if self.includes(example):
                examples.append(example)
        if not examples:
            raise ValueError(f"task {self.metadata.name!r} selected no rows")
        return tuple(examples)

dataset_transform(row)

Transform one source row into the normalized DecisionExample shape.

Source code in src/decision_bench/task_spec.py 
52
53
54
55
def dataset_transform(self, row: Mapping[str, Any]) -> Mapping[str, Any]:
    """Transform one source row into the normalized DecisionExample shape."""

    return decode_storage_row(dict(row))

includes(example)

Return whether a normalized row belongs to this task.

Source code in src/decision_bench/task_spec.py 
57
58
59
60
61
62
63
64
65
def includes(self, example: DecisionExample) -> bool:
    """Return whether a normalized row belongs to this task."""

    return (
        example.task_name == (self.metadata.source_task_name or self.metadata.name)
        and example.primitive is self.metadata.primitive
        and example.family == self.metadata.family
        and example.domain == self.metadata.domain
    )

load_data(*, project_root=None, rows=None)

Load, transform, filter, and validate this task's evaluation rows.

Source code in src/decision_bench/task_spec.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def load_data(
    self,
    *,
    project_root: Path | None = None,
    rows: Iterable[Mapping[str, Any] | object] | None = None,
) -> tuple[DecisionExample, ...]:
    """Load, transform, filter, and validate this task's evaluation rows."""

    source_rows = (
        load_rows(self.metadata.dataset, project_root=project_root)
        if rows is None
        else rows
    )
    examples: list[DecisionExample] = []
    for row in source_rows:
        if not isinstance(row, Mapping):
            raise TypeError(f"dataset row is not a mapping: {type(row)!r}")
        transformed = self.dataset_transform(row)
        example = DecisionExample.model_validate(transformed)
        if self.includes(example):
            examples.append(example)
    if not examples:
        raise ValueError(f"task {self.metadata.name!r} selected no rows")
    return tuple(examples)

decision_bench.task_spec.TaskMetadata

Bases: BaseModel

Describe one task, its taxonomy, and its pinned dataset.

Like MTEB's TaskMetadata, this metadata belongs to the task rather than to a benchmark. Many tasks may point at the same dataset revision, but no benchmark owns or republishes their rows.

Source code in src/decision_bench/task_spec.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class TaskMetadata(BaseModel):
    """Describe one task, its taxonomy, and its pinned dataset.

    Like MTEB's ``TaskMetadata``, this metadata belongs to the task rather than
    to a benchmark. Many tasks may point at the same dataset revision, but no
    benchmark owns or republishes their rows.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    name: str = Field(min_length=1)
    description: str = Field(min_length=1)
    dataset: DatasetSpec
    license: str = Field(min_length=1)
    languages: tuple[str, ...] = Field(min_length=1)
    primitive: Primitive
    family: str = Field(min_length=1)
    domain: str = Field(min_length=1)
    source_task_name: str | None = None
    reference: str | None = None
    citation: str | None = None
    is_public: bool = True
    superseded_by: str | None = None

decision_bench.task_spec.get_task(name)

Return one freshly initialized registered task.

Source code in src/decision_bench/task_spec.py 
107
108
109
110
111
112
113
114
115
def get_task(name: str) -> DecisionTask:
    """Return one freshly initialized registered task."""

    try:
        factory = _TASK_REGISTRY[name]
    except KeyError as error:
        available = ", ".join(sorted(_TASK_REGISTRY))
        raise KeyError(f"unknown task {name!r}; available tasks: {available}") from error
    return factory()

decision_bench.task_spec.get_tasks(task_names=None, *, languages=None, families=None, domains=None, primitives=None, exclude_superseded=True)

Return registered tasks selected by metadata, following MTEB's shape.

Source code in src/decision_bench/task_spec.py 
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
def get_tasks(
    task_names: Sequence[str] | None = None,
    *,
    languages: Sequence[str] | None = None,
    families: Sequence[str] | None = None,
    domains: Sequence[str] | None = None,
    primitives: Sequence[Primitive | str] | None = None,
    exclude_superseded: bool = True,
) -> list[DecisionTask]:
    """Return registered tasks selected by metadata, following MTEB's shape."""

    if task_names is not None:
        return [get_task(name) for name in task_names]

    normalized_primitives = (
        {Primitive(primitive) for primitive in primitives} if primitives is not None else None
    )
    selected: list[DecisionTask] = []
    for name in sorted(_TASK_REGISTRY):
        task = get_task(name)
        metadata = task.metadata
        if exclude_superseded and metadata.superseded_by is not None:
            continue
        if languages is not None and not set(languages).intersection(metadata.languages):
            continue
        if families is not None and metadata.family not in families:
            continue
        if domains is not None and metadata.domain not in domains:
            continue
        if normalized_primitives is not None and metadata.primitive not in normalized_primitives:
            continue
        selected.append(task)
    return selected