Sample Sources

Sample sources require the development version of Inspect, which you can install from GitHub:

pip install git+https://github.com/UKGovernmentBEIS/inspect_ai

Overview

The dataset argument to Task is normally static: you pass a Dataset (or list of samples) and the task runs exactly those. A SampleSource generates samples dynamically instead — a seed plus follow-ups that depend on results — all within one task and one log.

Use it when the next samples to run depend on the results of the previous ones:

Reinforcement-learning loops that generate the next samples from scores.
Adaptive evaluation that branches on model performance (e.g. escalate difficulty until failure).
Open-ended generation that runs until some external condition stops it.

A SampleSource is just a value the dataset parameter accepts, so there is no separate argument. It is the sample-level mirror of Task Sources: use a TaskSource to generate whole tasks across a run, a SampleSource to generate samples within a task.

Defining a source

Subclass SampleSource and override the methods you need (the defaults are no-ops):

from inspect_ai import SampleSource
from inspect_ai.dataset import Sample
from inspect_ai.log import EvalSample


class MySource(SampleSource):
    def initial_samples(self) -> list[Sample]:
        """Seed samples to run first (synchronous; may be empty)."""
        ...

    async def next_samples(self) -> list[Sample] | None:
        """More samples, or None when the task is complete."""
        ...

    async def sample_complete(self, sample: EvalSample) -> list[Sample] | None:
        """Observe a finished sample; optionally return follow-up samples."""
        ...

Pass an instance as the task’s dataset:

from inspect_ai import Task, eval

eval(Task(dataset=MySource(), solver=my_solver(), scorer=my_scorer()),
     model="openai/gpt-4o")

initial_samples() is synchronous and returns the seed, so it must return immediately rather than await. It may be empty, in which case the task starts by calling next_samples(). next_samples() is async, called whenever no samples remain in flight, and may block (for example, awaiting external input); return None to end the task.

All generated samples run within the task — one log file, one set of results — and each runs for the task’s configured number of epochs. Samples without an id are assigned one automatically, continuing the seed’s numbering.

Returning follow-up samples

sample_complete fires as each sample completes. Besides observing the result, it can return samples to add to the task, which start as soon as there is free capacity:

class Adaptive(SampleSource):
    def initial_samples(self) -> list[Sample]:
        return [make_sample(difficulty=1)]

    async def sample_complete(self, sample: EvalSample) -> list[Sample] | None:
        # escalate difficulty while the model keeps passing
        difficulty = sample.metadata["difficulty"]
        if passed(sample) and difficulty < 10:
            return [make_sample(difficulty=difficulty + 1)]
        return None

A source that returns follow-ups from this callback needs no next_samples(): the task ends when completions return nothing and next_samples() returns None. Use next_samples() for the blocking case a per-result callback can’t express.

Note that with epochs > 1, sample_complete fires once per epoch, and each returned follow-up itself runs all epochs — so a per-completion follow-up pattern multiplies (the example above with epochs=2 would spawn two follow-ups per passed difficulty level, each running twice). Use sample.epoch (or dedupe on the source’s own state) to react once per logical sample.

Sources from callbacks

SampleSource.from_samples() builds a source from a seed and optional callbacks, without subclassing:

from inspect_ai import SampleSource

scores: list[float] = []

async def on_sample(sample):
    scores.append(score_value(sample))
    return [harder_sample()] if sum(scores) / len(scores) >= 0.8 else None

source = SampleSource.from_samples([easy_sample()], sample_complete=on_sample)

from_samples(initial_samples, *, next_samples=None, sample_complete=None) delegates to the callables. Omitting next_samples and returning nothing from the callback stops after the seed.

Adding samples imperatively

enqueue_sample() adds samples to the running task from any code — a solver, scorer, or tool — not only the SampleSource itself:

from inspect_ai import enqueue_sample
from inspect_ai.dataset import Sample
from inspect_ai.solver import Generate, TaskState, solver


@solver
def spawn_followup():
    async def solve(state: TaskState, generate: Generate) -> TaskState:
        enqueue_sample(Sample(input=followup_prompt(state)))
        return state

    return solve

Enqueued samples share a buffer with samples returned from sample_complete. enqueue_sample() is only available inside a task driven by a SampleSource (a plain task’s sample set is fixed) and raises otherwise.

Concurrency

A SampleSource task is live: a sample added mid-run starts as soon as there is free capacity, rather than waiting for the current samples to finish. Concurrency is bounded by max_samples as usual (see Parallelism).

Sandboxes

Samples added mid-run get the same sandbox startup as the seed: a sandbox configuration first seen in an added sample (including the task’s own configuration when the seed is empty) is initialized — images built/pulled, configuration validated, cleanup registered — before that sample runs, and configurations the seed already initialized are not re-initialized. The seed is therefore not required for sandboxes: a source driven entirely by next_samples() works with sandboxed samples too.

Limits and sample filtering

The --limit option caps the total number of samples in the task — the seed plus everything the source produces. For example, --limit 10 on a source with 5 seed samples allows 5 more generated samples; once the cap is reached, further additions are ignored (with a warning) and the task ends when the in-flight samples finish (next_samples() is not consulted again). The cap counts samples, not runs: each sample within the limit still runs for the task’s configured number of epochs. A range limit (--limit 10,20) is not supported for SampleSource tasks — it selects samples by dataset position, which generated samples don’t have.

The --sample-id option filters generated samples the same way it filters the seed: only samples whose ids match run. Since the source may produce a requested id while the task runs, it is not an error for a --sample-id value to be missing from the seed.

A fractional fail_on_error threshold (e.g. 0.5) is evaluated at the end of the run rather than mid-run for SampleSource tasks: the planned total grows while the task runs, so a mid-run check would measure early errors against a transiently small denominator. Absolute-count and any-error thresholds behave as usual. The early_stopping task option is not supported with a SampleSource (managers require a fixed sample set).

Retries

On a task retry (eval_set / task retry attempts), completed samples from the prior attempt are reused, and the source’s sample_complete is called for each reused sample — so a source that derives its follow-ups from sample_complete (the patterns above) regenerates them naturally, and regenerated samples whose ids match the prior attempt are themselves reused rather than re-run. Note that the retry re-drives the same source instance: a source that instead holds internal next_samples() state resumes where it left off, so such sources should be written to be resumable if used with retries.

Logs

All samples — seed and generated — are written to one log file, and results.total_samples reflects everything that ran. Note however that the log’s dataset field (eval.dataset.samples and eval.dataset.sample_ids) describes only the seed: tools that read it as the planned sample count (for example the dataset_samples column in analysis dataframes) will under-count for a dynamic task, so use the results or the samples themselves for the true total.