Skip to content

GapK metric

Bases: SampledRewardMetric

gap-k: the reliability gap pass@k - pass^k.

A batched metric (each batch is one problem's k samples). The flakiness between optimistic and consistent performance: 0 means a problem is either always solved or never solved across the k samples; larger values mean the agent's success is sample-dependent. Lower is more reliable (direction = "down"). 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.
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)

# Report all three together: the optimistic bound, the consistency floor,
# and the gap between them.
program.compile(
    reward=synalinks.rewards.ExactMatch(in_mask=["answer"]),
    metrics=[
        synalinks.metrics.PassAtK(k=K),
        synalinks.metrics.PassHatK(k=K),
        synalinks.metrics.GapK(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["gap_k"])  # flakiness = pass@k - pass^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.

'gap_k'
Source code in synalinks/src/metrics/agents_metrics.py
@synalinks_export(
    [
        "synalinks.metrics.GapK",
        "synalinks.GapK",
    ]
)
class GapK(SampledRewardMetric):
    """``gap-k``: the reliability gap ``pass@k - pass^k``.

    A batched metric (each batch is one problem's ``k`` samples). The flakiness
    between optimistic and consistent performance: ``0`` means a problem is
    either always solved or never solved across the k samples; larger values
    mean the agent's success is sample-dependent. Lower is more reliable
    (``direction = "down"``). 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.
    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)

    # Report all three together: the optimistic bound, the consistency floor,
    # and the gap between them.
    program.compile(
        reward=synalinks.rewards.ExactMatch(in_mask=["answer"]),
        metrics=[
            synalinks.metrics.PassAtK(k=K),
            synalinks.metrics.PassHatK(k=K),
            synalinks.metrics.GapK(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["gap_k"])  # flakiness = pass@k - pass^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.
    """

    direction = "down"

    def __init__(self, k=1, reward=None, pass_threshold=1.0, name="gap_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:
            pass_at_k = 1.0
        else:
            pass_at_k = 1.0 - (math.comb(n - c, k) / math.comb(n, k))
        pass_hat_k = 0.0 if c < k else math.comb(c, k) / math.comb(n, k)
        return pass_at_k - pass_hat_k