import math

import statistics

from collections import Counter

from typing import Any, Callable, cast

from inspect_ai.scorer._metric import Score, Value, ValueToFloat, value_to_float

from .registry import score_reducer

from .types import ScoreReducer

@score_reducer(name="mode")

def mode_score() -> ScoreReducer:

r"""Take the mode from a list of scores."""

def reduce(scores: list[Score]) -> Score:

r"""A utility function for the most common score in a list of scores.

Args:

scores: a list of Scores.

"""

def most_common(

counts: Counter[str | int | float | bool],

) -> str | int | float | bool:

return counts.most_common(1)[0][0]

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

return _count_dict(scores, most_common)

elif isinstance(representative.value, list):

return _count_list(scores, most_common)

else:

return _count_scalar(scores, most_common)

return reduce

@score_reducer(name="majority")

def majority_score() -> ScoreReducer:

r"""Take the strict majority of a panel of scores.

A value wins only if more than half of the scores carry it. Unscored

(NaN) scores count towards the total rather than being filtered out of

it, so a panel member that fails to produce a value withholds a vote

without lowering the bar for the remaining values. Where nothing reaches

a majority the reduced score is unscored, rather than being decided by

the order the panel was declared in.

For dict and list values the threshold applies per key and per index, and

the total is still the number of scores reduced: a value missing from one

key (or a score that is unscored at the root) withholds a vote for that

key alone, or for every key, respectively.

The reduced score's metadata records the individual votes under a

`panel` key (replacing any `panel` carried over from the first score),

since a majority is only auditable alongside what was cast.

"""

def reduce(scores: list[Score]) -> Score:

panel_size = len(scores)

def strict_majority(

counts: Counter[str | int | float | bool],

) -> str | int | float | bool:

value, count = counts.most_common(1)[0]

return value if count * 2 > panel_size else float("nan")

representative = _first_scored(scores)

if representative is None:

reduced = _nan_score(scores)

elif isinstance(representative.value, dict):

reduced = _count_dict(scores, strict_majority)

elif isinstance(representative.value, list):

reduced = _count_list(scores, strict_majority)

else:

reduced = _count_scalar(scores, strict_majority)

return _with_panel_metadata(reduced, scores)

return reduce

@score_reducer(name="mean")

def mean_score(value_to_float: ValueToFloat = value_to_float()) -> ScoreReducer:

r"""Take the mean of a list of scores.

Args:

value_to_float: Function to convert the value to a float

"""

def reduce(scores: list[Score]) -> Score:

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

return _compute_dict_stat(scores, value_to_float, statistics.mean)

elif isinstance(representative.value, list):

return _compute_list_stat(scores, value_to_float, statistics.mean)

else:

return _compute_scalar_stat(scores, value_to_float, statistics.mean)

return reduce

@score_reducer(name="median")

def median_score(value_to_float: ValueToFloat = value_to_float()) -> ScoreReducer:

r"""Take the median value from a list of scores.

Args:

value_to_float: Function to convert the value to a float

"""

def reduce(scores: list[Score]) -> Score:

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

return _compute_dict_stat(scores, value_to_float, statistics.median)

elif isinstance(representative.value, list):

return _compute_list_stat(scores, value_to_float, statistics.median)

else:

return _compute_scalar_stat(scores, value_to_float, statistics.median)

return reduce

@score_reducer

def at_least(

k: int, value: float = 1.0, value_to_float: ValueToFloat = value_to_float()

) -> ScoreReducer:

r"""Score correct if there are at least k score values greater than or equal to the value.

Args:

k: Number of score values that must exceed `value`.

value: Score value threshold.

value_to_float: Function to convert score values to float.

"""

def reduce(scores: list[Score]) -> Score:

def gte_n(

counter: Counter[str | int | float | bool],

) -> str | int | float | bool:

count_gte_n = sum(

count for key, count in counter.items() if value_to_float(key) >= value

)

return 1 if count_gte_n >= k else 0

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

return _count_dict(scores, gte_n)

elif isinstance(representative.value, list):

return _count_list(scores, gte_n)

else:

return _count_scalar(scores, gte_n)

return reduce

@score_reducer

def pass_at(

k: int, value: float = 1.0, value_to_float: ValueToFloat = value_to_float()

) -> ScoreReducer:

r"""Probability of at least 1 correct sample given `k` epochs (<https://arxiv.org/pdf/2107.03374>).

Args:

k: Epochs to compute probability for.

value: Score value threshold.

value_to_float: Function to convert score values to float.

"""

def reduce(scores: list[Score]) -> Score:

def pass_at_k(values: list[float]) -> float:

import numpy as np

total = len(values)

correct = sum(1 for v in values if v >= value)

if total < k:

# NaN-filtering left fewer than k scored epochs, so the

# pass@k estimator is undefined; surface the unscored

# sentinel rather than the spurious 1.0 the short-circuit

# below would otherwise produce.

return float("nan")

if total - correct < k:

return 1.0

else:

return 1.0 - cast( # type: ignore[redundant-cast]

float,

np.prod(1.0 - k / np.arange(total - correct + 1, total + 1)).item(),

)

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

return _compute_dict_stat(scores, value_to_float, pass_at_k)

elif isinstance(representative.value, list):

return _compute_list_stat(scores, value_to_float, pass_at_k)

else:

return _compute_scalar_stat(scores, value_to_float, pass_at_k)

return reduce

@score_reducer

def pass_k(

k: int, value: float = 1.0, value_to_float: ValueToFloat = value_to_float()

) -> ScoreReducer:

r"""Probability that all `k` epoch attempts succeed (<https://arxiv.org/pdf/2406.12045>).

Computed as the draw-without-replacement estimator

`C(correct, k) / C(total, k)`, dual to `pass_at`'s Chen 2021 estimator.

Args:

k: Epochs to compute probability for.

value: Score value threshold.

value_to_float: Function to convert score values to float.

"""

def reduce(scores: list[Score]) -> Score:

def pass_k_k(values: list[float]) -> float:

total = len(values)

if total < k:

# NaN-filtering left fewer than k scored epochs, so the

# pass^k estimator is undefined; surface the unscored

# sentinel.

return float("nan")

correct = sum(1 for v in values if v >= value)

return math.comb(correct, k) / math.comb(total, k)

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

return _compute_dict_stat(scores, value_to_float, pass_k_k)

elif isinstance(representative.value, list):

return _compute_list_stat(scores, value_to_float, pass_k_k)

else:

return _compute_scalar_stat(scores, value_to_float, pass_k_k)

return reduce

@score_reducer(name="max")

def max_score(value_to_float: ValueToFloat = value_to_float()) -> ScoreReducer:

r"""Take the maximum value from a list of scores.

Args:

value_to_float: Function to convert the value to a float

"""

def reduce(scores: list[Score]) -> Score:

representative = _first_scored(scores)

if representative is None:

return _nan_score(scores)

if isinstance(representative.value, dict):

dict_scores = _partition_dict_scores(scores)

dict_result: dict[str, str | int | float | bool | None] = {}

keys = dict_scores[0].value.keys() # type: ignore

for key in keys:

key_values = [

cast(str | int | float | bool, score.value[key]) # type: ignore

for score in dict_scores

if _is_reducible(score.value[key]) # type: ignore

]

if len(key_values) == 0:

dict_result[key] = float("nan")

else:

dict_result[key] = max(key_values, key=value_to_float) # type: ignore

return _reduced_score(dict_result, scores)

elif isinstance(representative.value, list):

list_scores = _partition_list_scores(scores)

list_result: list[str | int | float | bool] = []

list_size = len(list_scores[0].value) # type: ignore

for i in range(list_size):

index_values = [

cast(str | int | float | bool, score.value[i]) # type: ignore

for score in list_scores

if _is_reducible(score.value[i]) # type: ignore

]

if len(index_values) == 0:

list_result.append(float("nan"))

else:

max_value = max(index_values, key=value_to_float) # type: ignore

if max_value is None:

raise ValueError(

"List of scores values unexpectedly had a `None` max score"

)

list_result.append(max_value)

return _reduced_score(list_result, scores)

else:

scalar_scores = [s for s in scores if not _is_unscored(s.value)]

max_score = max(

scalar_scores, key=lambda score: value_to_float(score.value)

)

return _reduced_score(max_score.value, scores)

return reduce

@score_reducer(name="collect")

def collect_score() -> ScoreReducer:

r"""Collect each score's value into a list, preserving every value.

Keeps the individual values intact instead of aggregating them into one.

Score values must be scalar; unscored (NaN) scores are dropped.

"""

def reduce(scores: list[Score]) -> Score:

values: list[str | int | float | bool] = []

for score in scores:

try:

value = score._as_scalar()

except ValueError:

raise ValueError(

"collect reducer requires scalar score values, but got "

f"{type(score.value).__name__}. It preserves each scorer's "

"scalar value as a list and cannot collect dict/list values."

) from None

if _is_reducible(value):

values.append(value)

if not values:

return _nan_score(scores)

return _reduced_score(values, scores)

return reduce

def _count_scalar(

scores: list[Score],

counter_fn: Callable[[Counter[str | int | float | bool]], str | int | float | bool],

) -> Score:

r"""Counts scores and provides Counter to a counter_fn

Args:

scores: a list of Scores.

counter_fn: a function which returns a scalar value based upon the counter

"""

score_values: list[str | int | float | bool] = []

for score in scores:

scalar_value = score._as_scalar()

if _is_reducible(scalar_value):

score_values.append(scalar_value)

# there are no reducible values

if len(score_values) == 0:

return _nan_score(scores)

counts = Counter(score_values)

return _reduced_score(counter_fn(counts), scores)

def _count_dict(

scores: list[Score],

counter_fn: Callable[[Counter[str | int | float | bool]], str | int | float | bool],

) -> Score:

r"""Counts scores within a dictionary and provides Counter (for each key) to a counter_fn

Args:

scores: a list of Scores.

counter_fn: a function which returns a scalar value based upon the counter

"""

# Filter to dict-shaped scores (skipping NaN-at-root unscored sentinels).

dict_scores = _partition_dict_scores(scores)

if len(dict_scores) == 0:

return _nan_score(scores)

dict_result: dict[str, str | int | float | bool] = {}

keys = dict_scores[0].value.keys() # type: ignore

for key in keys:

key_values = []

for score in dict_scores:

key_value = cast(str | int | float | bool, score.value[key]) # type: ignore

if _is_reducible(key_value):

key_values.append(key_value)

# there are no reducible values

if len(key_values) == 0:

dict_result[key] = float("nan")

else:

counts: Counter[str | int | float | bool] = Counter(key_values)

dict_result[key] = counter_fn(counts)

return _reduced_score(

cast(dict[str, str | int | float | bool | None], dict_result), scores

)

def _count_list(

scores: list[Score],

counter_fn: Callable[[Counter[str | int | float | bool]], str | int | float | bool],

) -> Score:

r"""Counts scores within a list and provides Counter (for each index) to a counter_fn

Args:

scores: a list of Scores.

counter_fn: a function which returns a scalar value based upon the counter

"""

# Filter to list-shaped scores (skipping NaN-at-root unscored sentinels).

list_scores = _partition_list_scores(scores)

if len(list_scores) == 0:

return _nan_score(scores)

list_result: list[str | int | float | bool] = []

list_size = len(list_scores[0].value) # type: ignore

for i in range(list_size):

index_values = []

for score in list_scores:

index_value = cast(str | int | float | bool, score.value[i]) # type:ignore

if _is_reducible(index_value):

index_values.append(index_value)

if len(index_values) == 0:

list_result.append(float("nan"))

else:

counts: Counter[str | int | float | bool] = Counter(index_values)

list_result.append(counter_fn(counts))

return _reduced_score(list_result, scores)

def _compute_dict_stat(

scores: list[Score],

value_to_float: ValueToFloat,

statistic: Callable[[list[float]], float],

) -> Score:

r"""Applies a statistic function to reduce key by key a dictionary

Args:

scores: a list of Scores.

value_to_float: Function to convert the value to a float

statistic: the statistic to apply

"""

# Filter to dict-shaped scores (skipping NaN-at-root unscored sentinels).

dict_scores = _partition_dict_scores(scores)

if len(dict_scores) == 0:

return _nan_score(scores)

dict_result: dict[str, str | int | float | bool | None] = {}

for key in dict_scores[0].value.keys(): # type: ignore

values = []

for score in dict_scores:

key_value = value_to_float(score.value[key]) # type: ignore

if _is_reducible(key_value):

values.append(key_value)

if len(values) == 0:

dict_result[key] = float("nan")

else:

dict_result[key] = statistic(values)

return _reduced_score(dict_result, scores)

def _compute_list_stat(

scores: list[Score],

value_to_float: ValueToFloat,

statistic: Callable[[list[float]], float],

) -> Score:

r"""Applies a statistic function to reduce index by index a list

Args:

scores: a list of Scores.

value_to_float: function to convert the value to a float

statistic: the statistic to apply

"""

# Filter to list-shaped scores (skipping NaN-at-root unscored sentinels).

list_scores = _partition_list_scores(scores)

if len(list_scores) == 0:

return _nan_score(scores)

list_result: list[str | int | float | bool] = []

list_size = len(list_scores[0].value) # type: ignore

for i in range(list_size):

values = []

for score in list_scores:

list_values = cast(list[str | int | float | bool], score.value)

value = value_to_float(list_values[i])

if _is_reducible(value):

values.append(value)

if len(values) == 0:

list_result.append(float("nan"))

else:

list_result.append(statistic(values))

return _reduced_score(list_result, scores)

def _compute_scalar_stat(

scores: list[Score],

value_to_float: ValueToFloat,

statistic: Callable[[list[float]], float],

) -> Score:

r"""Applies a statistic function to reduce scalar scores

Args:

scores: a list of Scores.

value_to_float: function to convert the value to a float

statistic: the statistic to apply

"""

values = []

for score in scores:

if _is_reducible(value_to_float(score.value)):

values.append(value_to_float(score.value))

# there are no reducible values

if len(values) == 0:

return _nan_score(scores)

result = statistic(values)

return _reduced_score(result, scores)

def _is_unscored(value: Value) -> bool:

r"""Check if a score value is the NaN-at-root unscored sentinel."""

return isinstance(value, float) and math.isnan(value)

def _first_scored(scores: list[Score]) -> Score | None:

r"""Return the first score that is not NaN-at-root, or None if all are unscored."""

for score in scores:

if not _is_unscored(score.value):

return score

return None

def _partition_dict_scores(scores: list[Score]) -> list[Score]:

r"""Return the subset of scores whose value is a dict.

Skips scores with NaN-at-root (treated as unscored). Raises ValueError

if any score has a value that is neither a dict nor a NaN scalar, or if the

dict-shaped scores don't all share the same keys.

"""

result: list[Score] = []

for score in scores:

if isinstance(score.value, dict):

result.append(score)

elif _is_unscored(score.value):

continue

else:

raise ValueError(

"Attempting to reduce a dictionary score for a non-dictionary value"

)

# Reducers walk the keys of the first dict and look them up in every other

# dict, so differing keys across epochs either crash with a KeyError or

# silently drop the extra keys. Reject the inconsistency up front.

if result:

keys = set(result[0].as_dict().keys())

for score in result[1:]:

score_keys = set(score.as_dict().keys())

if score_keys != keys:

raise ValueError(

"Cannot reduce dictionary scores with mismatched keys: "

f"{sorted(keys)} vs {sorted(score_keys)}. "

"Every epoch must score the same keys; return a NaN score to "

"mark an individual epoch as unscored."

)

return result

def _partition_list_scores(scores: list[Score]) -> list[Score]:

r"""Return the subset of scores whose value is a list.

Skips scores with NaN-at-root (treated as unscored). Raises ValueError

if any score has a value that is neither a list nor a NaN scalar, or if the

list-shaped scores don't all share the same length.

"""

result: list[Score] = []

for score in scores:

if isinstance(score.value, list):

result.append(score)

elif _is_unscored(score.value):

continue

else:

raise ValueError("Attempting to reduce a list score for a non-list value")

# Reducers walk the indices of the first list and read them from every other

# list, so differing lengths across epochs either crash with an IndexError

# or silently drop the trailing values. Reject the inconsistency up front.

if result:

length = len(result[0].as_list())

for score in result[1:]:

score_length = len(score.as_list())

if score_length != length:

raise ValueError(

"Cannot reduce list scores with mismatched lengths: "

f"{length} vs {score_length}. "

"Every epoch must produce the same number of values; return a "

"NaN score to mark an individual epoch as unscored."

)

return result

def _reduced_score(value: Value, scores: list[Score]) -> Score:

r"""Create a Score based upon a single Value and list of Scores that produced it

Args:

value: the reduced Value

scores: ths list of scores being reduced

"""

return Score(

value=value,

# retain remaining fields only if equal across all Scores

answer=scores[0].answer

if len(set(score.answer for score in scores)) == 1

else None,

explanation=scores[0].explanation

if len(set(score.explanation for score in scores)) == 1

else None,

reason=scores[0].reason

if len(set(score.reason for score in scores)) == 1

else None,

metadata=scores[0].metadata,

)

def _unscored_reason(score: Score) -> str | None:

"""Machine-readable reason a score is unscored, or None.

Reads the first-class `Score.reason` field where it exists, falling back to

the legacy `metadata["unscored_reason"]` convention. Checks `is None` rather

than truthiness so an explicitly-set empty reason is not overridden by a

stale legacy value.

"""

reason = getattr(score, "reason", None)

if reason is not None:

return reason if isinstance(reason, str) else None

legacy = (score.metadata or {}).get("unscored_reason")

return legacy if isinstance(legacy, str) else None

def _with_panel_metadata(reduced: Score, scores: list[Score]) -> Score:

r"""Record the votes behind a reduced panel score in its metadata.

Reduction otherwise carries over only the first score's metadata, which

for a panel loses what a majority has to be read against: which member

voted for what, and which ones didn't vote at all.

Args:

reduced: the reduced Score to annotate

scores: the list of scores that produced it

"""

votes: list[Value | None] = []

failures: list[dict[str, Any]] = []

for index, score in enumerate(scores):

unscored = _is_unscored(score.value)

# copy container values: the record is an audit trail of what was cast,

# so it must not alias (and drift with) the score it came from

if unscored:

votes.append(None)

elif isinstance(score.value, dict):

votes.append(dict(score.value))

elif isinstance(score.value, list):

votes.append(list(score.value))

else:

votes.append(score.value)

if unscored:

failures.append(

dict(

index=index,

reason=_unscored_reason(score),

explanation=score.explanation,

)

)

panel = dict(votes=votes, size=len(scores), failures=failures)

return reduced.model_copy(

update=dict(metadata={**(reduced.metadata or {}), "panel": panel})

)

def _nan_score(scores: list[Score]) -> Score:

r"""Create a NaN Score based upon a single Value and list of Scores that produced it

Args:

value: the reduced Value

scores: ths list of scores being reduced

"""

# An empty list still routes here via `_first_scored` returning None; there

# are no fields to carry over, so return a bare NaN score rather than

# indexing `scores[0]`.

if not scores:

return Score(value=float("nan"))

return Score(

value=float("nan"),

# retain remaining fields only if equal across all Scores

answer=scores[0].answer

if len(set(score.answer for score in scores)) == 1

else None,

explanation=scores[0].explanation

if len(set(score.explanation for score in scores)) == 1

else None,

reason=scores[0].reason

if len(set(score.reason for score in scores)) == 1

else None,

metadata=scores[0].metadata,

)

def _is_reducible(value: str | int | float | bool) -> bool:

"""Check if a value is reducible (not a NaN float)."""

return not (isinstance(value, float) and math.isnan(value))