mellea.stdlib.sampling.base

Base Sampling Strategies.

Base Sampling Strategies.

Sampling strategies control how Mellea handles validation failures during generation:

• RejectionSamplingStrategy: Simple retry with the same prompt. Best for non-deterministic failures where the same instruction might succeed on retry.

• RepairTemplateStrategy: Single-turn repair by modifying the instruction with validation feedback. Adds failure reasons to the instruction and retries. Best for simple tasks where feedback can be incorporated into the instruction.

• MultiTurnStrategy: Multi-turn conversational repair (requires ChatContext). Adds validation failure reasons as new user messages in the conversation, allowing iterative improvement through dialogue. Best for complex tasks and agentic workflows.

Classes

CLASS BaseSamplingStrategy

Base class for multiple strategies that reject samples based on given instructions.

Args:

• loop_budget: Maximum number of generate/validate cycles per concurrent subsample. Must be greater than 0. Defaults to 1.
• concurrency_budget: Number of concurrent subsamples. Sampling generates at most loop_budget * concurrency_budget requests and stops at the first valid result. Must be greater than 0. Defaults to 1 (no concurrency). When allowing concurrency, number of generations may substantially increase (potentially causing rate-limiting issues depending on the provider) and the returned order of results will no longer be deterministic.
• requirements: Global requirements evaluated on every sample. When set, overrides per-call requirements.

Examples:

• loop_budget=1: no repair strategies are used.
• loop_budget=3, concurrency_budget=1: generate -> repair -> generate -> repair -> final generate.
• loop_budget=2, concurrency_budget=2: two concurrent subsamples, each with one repair.

Raises:

• ValueError: If loop_budget < 1 or concurrency_budget < 1.

Methods:

FUNC repair

repair(old_ctx: Context, new_ctx: Context, past_actions: list[Component], past_results: list[ComputedModelOutputThunk], past_val: list[list[tuple[Requirement, ValidationResult]]]) -> tuple[Component, Context]

Repair function that is being invoked if not all requirements are fulfilled. It should return a next action component.

Args:

• old_ctx: The context WITHOUT the last action + output.
• new_ctx: The context including the last action + output.
• past_actions: List of actions that have been executed (without success).
• past_results: List of (unsuccessful) generation results for these actions.
• past_val: List of validation results for the results.

Returns:

• The next action component and context to be used for the next generation attempt.

FUNC select_from_failure

select_from_failure(sampled_actions: list[Component], sampled_results: list[ComputedModelOutputThunk], sampled_val: list[list[tuple[Requirement, ValidationResult]]]) -> int

This function returns the index of the result that should be selected as .value iff the loop budget is exhausted and no success.

Args:

• sampled_actions: List of actions that have been executed (without success).
• sampled_results: List of (unsuccessful) generation results for these actions.
• sampled_val: List of validation results for the results.

Returns:

• The index of the result that should be selected as .value.

FUNC sample

sample(self, action: Component[S], context: Context, backend: Backend, requirements: list[Requirement] | None) -> SamplingResult[S]

This method performs a sampling operation based on the given instruction.

Args:

• action : The action object to be sampled.
• context: The context to be passed to the sampling strategy.
• backend: The backend used for generating samples.
• requirements: List of requirements to test against (merged with global requirements).
• validation_ctx: Optional context to use for validation. If None, validation_ctx = ctx.
• format: output format for structured outputs.
• model_options: model options to pass to the backend during generation / validation.
• tool_calls: True if tool calls should be used during this sampling strategy.
• show_progress: if true, a tqdm progress bar is used. Otherwise, messages will still be sent to flog.

Returns:

• SamplingResult[S]: A result object indicating the success or failure of the sampling process.

Raises:

• AssertionError: Asserts that all required components (repair, select_from_failure, validate, and generate) are provided before proceeding with the sampling.
• ValueError: If a SAMPLING_LOOP_START hook returns a non-positive loop_budget.

CLASS RejectionSamplingStrategy

Simple rejection sampling strategy that just repeats the same call on failure.

Methods:

FUNC select_from_failure

select_from_failure(sampled_actions: list[Component], sampled_results: list[ComputedModelOutputThunk], sampled_val: list[list[tuple[Requirement, ValidationResult]]]) -> int

Always returns the 0th index.

Args:

• sampled_actions: List of actions that have been executed (without success).
• sampled_results: List of (unsuccessful) generation results for these actions.
• sampled_val: List of validation results for the results.

Returns:

• The index of the result that should be selected as .value.

FUNC repair

repair(old_ctx: Context, new_ctx: Context, past_actions: list[Component], past_results: list[ComputedModelOutputThunk], past_val: list[list[tuple[Requirement, ValidationResult]]]) -> tuple[Component, Context]

Always returns the unedited, last action.

Args:

• old_ctx: The context WITHOUT the last action + output.
• new_ctx: The context including the last action + output.
• past_actions: List of actions that have been executed (without success).
• past_results: List of (unsuccessful) generation results for these actions.
• past_val: List of validation results for the results.

Returns:

• The next action component and context to be used for the next generation attempt.

CLASS RepairTemplateStrategy

A sampling strategy that adds a repair string to the instruction object.

Methods:

FUNC select_from_failure

select_from_failure(sampled_actions: list[Component], sampled_results: list[ComputedModelOutputThunk], sampled_val: list[list[tuple[Requirement, ValidationResult]]]) -> int

Always returns the 0th index.

Args:

• sampled_actions: List of actions that have been executed (without success).
• sampled_results: List of (unsuccessful) generation results for these actions.
• sampled_val: List of validation results for the results.

Returns:

• The index of the result that should be selected as .value.

FUNC repair

repair(old_ctx: Context, new_ctx: Context, past_actions: list[Component], past_results: list[ComputedModelOutputThunk], past_val: list[list[tuple[Requirement, ValidationResult]]]) -> tuple[Component, Context]

Adds a description of the requirements that failed to a copy of the original instruction.

Args:

• old_ctx: The context WITHOUT the last action + output.
• new_ctx: The context including the last action + output.
• past_actions: List of actions that have been executed (without success).
• past_results: List of (unsuccessful) generation results for these actions.
• past_val: List of validation results for the results.

Returns:

• The next action component and context to be used for the next generation attempt.

CLASS MultiTurnStrategy

Rejection sampling strategy with (agentic) multi-turn repair.

Methods:

FUNC select_from_failure

select_from_failure(sampled_actions: list[Component], sampled_results: list[ComputedModelOutputThunk], sampled_val: list[list[tuple[Requirement, ValidationResult]]]) -> int

Always returns the last index. The last message from the model will always be returned if all results are failures.

If utilizing concurrency, this will always be the last turn of one of the multi-turn samples; but there is no guarantee of which concurrent sampling it will be from.

Args:

• sampled_actions: List of actions that have been executed (without success).
• sampled_results: List of (unsuccessful) generation results for these actions.
• sampled_val: List of validation results for the results.

Returns:

• The index of the result that should be selected as .value.

FUNC repair

repair(old_ctx: Context, new_ctx: Context, past_actions: list[Component], past_results: list[ComputedModelOutputThunk], past_val: list[list[tuple[Requirement, ValidationResult]]]) -> tuple[Component, Context]

Returns a Message with a description (and validation reasons) of the failed requirements.

Args:

• old_ctx: The context WITHOUT the last action + output.
• new_ctx: The context including the last action + output.
• past_actions: List of actions that have been executed (without success).
• past_results: List of (unsuccessful) generation results for these actions.
• past_val: List of validation results for the results.

Returns:

• The next action component and context to be used for the next generation attempt.