lm_eval.scorers¶

def score_doc(self, doc_id: int, doc_instances: list[Instance]) -> ScoredDoc:
    """Extract inputs from a document's instances and delegate to [score][.score]."""
    ref, preds, mkw = self._extract_inputs(doc_instances)
    return ScoredDoc(
        doc_id=doc_id,
        reference=ref,
        scores=self.score(ref, preds, metric_kwargs=mkw),
    )

def score(
    self,
    reference: str | list[str],
    predictions: list[str],
    metric_kwargs: dict[str, Any] | None = None,
) -> dict[str, list[float]]:
    """Per-document scoring.  Override for custom generation scoring.

    This is the simplest hook.  Receives clean inputs and returns
    metric scores — no need to work with ``Instance`` or ``ScoredDoc``.

    Args:
        reference: The gold answer(s).
        predictions: Model predictions (one per repeat).
        metric_kwargs: Optional per-instance metric overrides.

    Returns:
        ``{metric_name: [score_per_repeat]}``.

    Example:
        ```python
        @register_scorer("code_exec")
        @dataclass
        class CodeExecScorer(GenScorer):
            timeout: int = 10

            def score(self, reference, predictions, **kwargs):
                return {
                    "pass": [
                        1.0 if run(code, self.timeout) == reference else 0.0
                        for code in predictions
                    ]
                }
        ```
    """
    return self._dispatch_metrics(
        [reference], predictions, metric_kwargs=metric_kwargs
    )

def score_doc(self, doc_id: int, doc_instances: list[Instance]) -> ScoredDoc:
    from lm_eval.api.metrics.results import LLResults

    metric_kwargs = doc_instances[0].metadata.get("metric_kwargs")
    results_obj = LLResults.from_instances(doc_instances, self.name)
    references = results_obj.targets
    per_doc = self._dispatch_metrics(
        references, results_obj, metric_kwargs=metric_kwargs
    )
    return ScoredDoc(
        doc_id=doc_id,
        reference=references,
        scores={mn: [v] for mn, v in per_doc.items()},
    )

@classmethod
def from_dict(
    cls,
    cfg: _ScorerCfg,
    *,
    output_type: str | None = None,
    **kwargs: Any,
) -> Self:
    """Build a Scorer from a normalised pipeline config.

    *cfg* is a [_ScorerCfg][_ScorerCfg] produced by
    ``TaskConfig._normalize_scoring_config()``:

        ```python
        {
            "name": "strict-match",
            "filter": [
                {"function": "regex", "regex_pattern": "..."},
                {"function": "take_first"},
            ],
            "metric_list": [
                {"metric": "exact_match", "aggregation": "mean", ...},
            ],
        }
        ```

    *output_type* is used as a last-resort fallback when neither the
    config nor the class provides metrics (see
    ``DEFAULT_METRIC_REGISTRY``).

    Any extra *kwargs* are forwarded to the constructor (e.g., custom
    dataclass fields on scorer subclasses).
    """
    name = cfg.get("name", "none")
    return cls(
        name=name,
        filter=cls._build_filter(name, cfg),
        metrics=cls._build_metrics(cfg, output_type=output_type),
        **kwargs,
    )

@classmethod
def default_scorer(cls, name: str = "none", **kwargs: Any) -> Self:
    """Build the default scorer (no explicit config).

    Filter defaults to ``cls.default_filter_cfg`` if set, otherwise
    ``noop``.
    """
    return cls.from_dict({"name": name, "filter": [], "metric_list": []}, **kwargs)

def apply_filter(self, instances: list[Instance]) -> None:
    self.filter.apply(instances)

def score_instances(
    self, instances: Mapping[int, list[Instance]]
) -> dict[int, ScoredDoc]:
    """Score all documents' instances, returning a ``ScoredDoc`` per document.

    Delegates per-document scoring to [score_doc][.score_doc], which subclasses
    must implement.
    """
    return {
        doc_id: self.score_doc(doc_id, doc_instances)
        for doc_id, doc_instances in instances.items()
    }

def score_doc(self, doc_id: int, doc_instances: list[Instance]) -> ScoredDoc:
    """Score a single document's instances. Subclasses must implement this.

    Override this method to define custom scoring logic.  Use
    [_dispatch_metrics][._dispatch_metrics] as a helper to run configured metrics,
    or compute scores directly and return a [ScoredDoc][lm_eval.scorers.ScoredDoc].
    """
    raise NotImplementedError(
        f"{type(self).__name__} must implement score_doc(). "
        "Override score_doc() on your Scorer subclass, "
        "or subclass GenScorer and override score() for per-doc scoring, "
        "or override score_instances() for batch scoring."
    )

def reduce(self, scored_docs: dict[int, ScoredDoc]) -> dict[int, ReducedDoc]:
    """Reduce per-doc ``list[T]`` → ``T`` for each document.

    Pure function: takes [ScoredDoc][lm_eval.scorers.ScoredDoc] objects (immutable raw scores)
    and returns ``{doc_id: {metric: scalar}}`` dicts ready for aggregation.

    For each metric in each document:

    * **Single value** — passed through as-is (no reduction needed).
    * **Multiple values + reduction fn** — calls ``Metric.reduction(reference, values)``.
      If the reduction returns a dict, composite keys like ``"pass@1(metric)"``
      are created.
    * **No reduction fn** — warns and takes the first value.
    """
    metrics_by_name = self._metrics_by_name
    result: dict[int, ReducedDoc] = {}

    for sd in scored_docs.values():
        values_dict: ReducedDoc = {}
        for metric_name, score_list in sd.scores.items():
            if len(score_list) == 1:
                values_dict[metric_name] = score_list[0]
                continue
            m = metrics_by_name.get(metric_name)
            if m is not None and m.reduction is not None:
                res = m.reduction(sd.reference, score_list)
                if isinstance(res, dict):
                    for sub_metric_name, sub_value in res.items():
                        values_dict[f"{sub_metric_name}({metric_name})"] = sub_value
                else:
                    values_dict[metric_name] = res
            else:
                raise ValueError(
                    f"Metric '{metric_name}' in scorer '{self.name}' has "
                    f"{len(score_list)} values per document (repeats > 1) "
                    f"but no reduction function is configured. Set a "
                    f"reduction (e.g., 'take_first', 'pass@k', 'mean') in "
                    f"your metric config, or set repeats to 1."
                )

        result[sd.doc_id] = values_dict

    return result

def aggregate(
    self,
    reduced_docs: Mapping[int, ReducedDoc],
    bootstrap_iters: int | None = 100000,
    aggregation_overrides: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], int]:
    """Aggregate reduced docs and compute stderr.

    Pure function: takes ``{doc_id: {metric: value}}`` and produces
    aggregated ``"metric,scorer"`` keyed results.  When
    *aggregation_overrides* is supplied (legacy Python tasks that override
    ``Task.aggregation()``), those functions take precedence over the
    ``mean`` fallback for metrics not covered by a ``Metric`` object.

    Returns ``(agg_metrics, sample_len)`` where keys are in
    ``"metric,{self.name}"`` / ``"metric_stderr,{self.name}"`` format.
    """
    from lm_eval.api.metrics import mean, stderr_for_metric

    # Transpose doc-first → metric-first: {metric: [values]}
    results: dict[str, list[float]] = {}
    for rd in reduced_docs.values():
        for mn, val in rd.items():
            results.setdefault(mn, []).append(val)

    agg: dict[str, Any] = {}
    sample_len = 0
    metrics_by_name = self._metrics_by_name

    for metric_name, values in results.items():
        if not values:
            continue
        sample_len = max(sample_len, len(values))

        # Resolve metric object (check parent for composite keys like "pass@1(exact_match)")
        m = metrics_by_name.get(metric_name)
        if m is None and (
            parent := MetricKey(metric_name, self.name).parent_metric
        ):
            m = metrics_by_name.get(parent)

        # Resolve aggregation function: metric > legacy override > mean fallback
        if m is not None and m.aggregation is not None:
            agg_fn = m.aggregation
        elif aggregation_overrides and metric_name in aggregation_overrides:
            agg_fn = aggregation_overrides[metric_name]
        else:
            eval_logger.error(
                "No aggregation function for metric '%s' in scorer '%s'. "
                "Defaulting to 'mean'. WARNING: this will produce INCORRECT "
                "results for corpus-level metrics (BLEU, perplexity, F1, etc.). "
                "Set 'aggregation' explicitly in your metric config.",
                metric_name,
                self.name,
            )
            agg_fn = mean

        # Aggregate + stderr
        key = str(MetricKey(metric_name, self.name))
        agg[key] = agg_fn(values)

        stderr_key = str(MetricKey(metric_name, self.name, is_stderr=True))
        if isinstance(bootstrap_iters, int) and bootstrap_iters > 0:
            stderr_fn = stderr_for_metric(
                metric=agg_fn, bootstrap_iters=bootstrap_iters
            )
            agg[stderr_key] = (
                stderr_fn(values) if (stderr_fn and len(values) > 1) else "N/A"
            )
        else:
            agg[stderr_key] = "N/A"

    return agg, sample_len

def export_reduced(self) -> dict[int, ReducedDoc]:
    """Export ``{doc_id: {metric: value}}`` for distributed gathering.

    Since ``ReducedDoc`` is a plain ``dict[str, float]``, this is a
    shallow copy.  Merge across ranks is a simple ``dict.update``
    since doc IDs are unique per rank.
    """
    return dict(self._reduced_docs)

def import_reduced(self, doc_data: dict[int, ReducedDoc]) -> None:
    """Import merged results after distributed gather.

    Raw scores are not available after import (they live on the
    source ranks).
    """
    self._raw_docs = {}
    self._reduced_docs = dict(doc_data)

def set_results(self, scored_docs: dict[int, ScoredDoc]) -> None:
    """Store raw scored documents and compute reduction."""
    self._raw_docs = scored_docs
    self._reduced_docs = self.reduce(scored_docs)

def __str__(self) -> str:
    name = f"{self.metric}_stderr" if self.is_stderr else self.metric
    return f"{name},{self.scorer}"

@classmethod
def parse(cls, key: str) -> MetricKey | None:
    """Parse a ``'metric,scorer'`` string. Returns ``None`` if not a metric key."""
    if "," not in key:
        return None
    left, _, scorer = key.partition(",")
    if left.endswith("_stderr"):
        return cls(metric=left[: -len("_stderr")], scorer=scorer, is_stderr=True)
    return cls(metric=left, scorer=scorer)