Skip to content

PassAtK metric

Bases: SampledRewardMetric

pass@k: fraction of problems solved in at least one of k samples.

A batched metric: each batch is the k samples of one problem (set batch_size = k). Uses the unbiased HumanEval estimator over the n samples in the batch, of which c are correct::

pass@k = 1 - C(n - c, k) / C(n, k)

(= 1.0 when n - c < k). Averaged over the dataset.

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;
# with greedy decoding all K are identical and pass@k collapses to pass@1.
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.PassAtK(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_at_k"])

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_at_k'
Source code in synalinks/src/metrics/agents_metrics.py
@synalinks_export(
    [
        "synalinks.metrics.PassAtK",
        "synalinks.PassAtK",
    ]
)
class PassAtK(SampledRewardMetric):
    """``pass@k``: fraction of problems solved in *at least one* of k samples.

    A batched metric: each batch is the ``k`` samples of one problem (set
    ``batch_size = k``). Uses the unbiased HumanEval estimator over the ``n``
    samples in the batch, of which ``c`` are correct::

        pass@k = 1 - C(n - c, k) / C(n, k)

    (``= 1.0`` when ``n - c < k``). Averaged over the dataset.

    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;
    # with greedy decoding all K are identical and pass@k collapses to pass@1.
    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.PassAtK(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_at_k"])
    ```

    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_at_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:
            return 0.0
        if n - c < k:
            return 1.0
        return 1.0 - (math.comb(n - c, k) / math.comb(n, k))