Skip to content

PassHatK metric

Bases: SampledRewardMetric

pass^k: fraction of problems solved in all k samples (consistency).

A batched metric (each batch is one problem's k samples). Unbiased estimator of the probability that k samples drawn (without replacement) from the n batch samples are all correct::

pass^k = C(c, k) / C(n, k)

(= 0.0 when c < k). Averaged over the dataset. Always <= the corresponding PassAtK; the difference is reported by GapK.

Example:

import synalinks

class Question(synalinks.DataModel):
    question: str

class Answer(synalinks.DataModel):
    answer: str

K = 5

# Sampling (temperature > 0) so the K rollouts of one prompt differ.
inputs = synalinks.Input(data_model=Question)
outputs = await synalinks.Generator(
    data_model=Answer,
    language_model=synalinks.LanguageModel(model="openai/gpt-4o-mini"),
    temperature=0.8,
)(inputs)
program = synalinks.Program(inputs=inputs, outputs=outputs)

program.compile(
    reward=synalinks.rewards.ExactMatch(in_mask=["answer"]),
    metrics=[synalinks.metrics.PassHatK(k=K)],
)

# repeat == batch_size == K  ->  each batch is one problem's K samples.
dataset = synalinks.HuggingFaceDataset(
    hf_dataset_name="openai/gsm8k",
    hf_config_name="main",
    split="test",
    input_data_model=Question,
    output_data_model=Answer,
    input_template='{"question": {{ question | tojson }}}',
    output_template='{"answer": {{ answer.split("####")[-1].strip() | tojson }}}',
    batch_size=K,
    repeat=K,
    limit=20,
)

metrics = await program.evaluate(x=dataset())
print(metrics["pass_hat_k"])  # consistency: solved in ALL K samples

Parameters:

Name Type Description Default
k int

The number of samples k.

1
reward Reward

Per-sample correctness signal (default ExactMatch).

None
pass_threshold float

Reward threshold for a sample to count as correct. Defaults to 1.0.

1.0
name str

Optional. Name of the metric instance.

'pass_hat_k'
Source code in synalinks/src/metrics/agents_metrics.py
@synalinks_export(
    [
        "synalinks.metrics.PassHatK",
        "synalinks.PassHatK",
    ]
)
class PassHatK(SampledRewardMetric):
    """``pass^k``: fraction of problems solved in *all* k samples (consistency).

    A batched metric (each batch is one problem's ``k`` samples). Unbiased
    estimator of the probability that ``k`` samples drawn (without replacement)
    from the ``n`` batch samples are *all* correct::

        pass^k = C(c, k) / C(n, k)

    (``= 0.0`` when ``c < k``). Averaged over the dataset. Always ``<=`` the
    corresponding `PassAtK`; the difference is reported by `GapK`.

    Example:

    ```python
    import synalinks

    class Question(synalinks.DataModel):
        question: str

    class Answer(synalinks.DataModel):
        answer: str

    K = 5

    # Sampling (temperature > 0) so the K rollouts of one prompt differ.
    inputs = synalinks.Input(data_model=Question)
    outputs = await synalinks.Generator(
        data_model=Answer,
        language_model=synalinks.LanguageModel(model="openai/gpt-4o-mini"),
        temperature=0.8,
    )(inputs)
    program = synalinks.Program(inputs=inputs, outputs=outputs)

    program.compile(
        reward=synalinks.rewards.ExactMatch(in_mask=["answer"]),
        metrics=[synalinks.metrics.PassHatK(k=K)],
    )

    # repeat == batch_size == K  ->  each batch is one problem's K samples.
    dataset = synalinks.HuggingFaceDataset(
        hf_dataset_name="openai/gsm8k",
        hf_config_name="main",
        split="test",
        input_data_model=Question,
        output_data_model=Answer,
        input_template='{"question": {{ question | tojson }}}',
        output_template='{"answer": {{ answer.split("####")[-1].strip() | tojson }}}',
        batch_size=K,
        repeat=K,
        limit=20,
    )

    metrics = await program.evaluate(x=dataset())
    print(metrics["pass_hat_k"])  # consistency: solved in ALL K samples
    ```

    Args:
        k (int): The number of samples ``k``.
        reward (Reward): Per-sample correctness signal (default `ExactMatch`).
        pass_threshold (float): Reward threshold for a sample to count as
            correct. Defaults to ``1.0``.
        name (str): Optional. Name of the metric instance.
    """

    def __init__(self, k=1, reward=None, pass_threshold=1.0, name="pass_hat_k"):
        super().__init__(k=k, reward=reward, pass_threshold=pass_threshold, name=name)

    def _estimate(self, n, c):
        k = min(self.k, n)
        if k <= 0 or c < k:
            return 0.0
        return math.comb(c, k) / math.comb(n, k)