Skip to content

Utils

promptolution.utils

Module for utility functions and classes.

BestPromptCallback

Bases: BaseCallback

Callback for tracking the best prompt during optimization.

This callback keeps track of the prompt with the highest score.

Attributes:

Name Type Description
best_prompt str

The prompt with the highest score so far.
best_score float

The highest score achieved so far.
Source code in promptolution/utils/callbacks.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class BestPromptCallback(BaseCallback):
    """Callback for tracking the best prompt during optimization.

    This callback keeps track of the prompt with the highest score.

    Attributes:
        best_prompt (str): The prompt with the highest score so far.
        best_score (float): The highest score achieved so far.
    """

    def __init__(self) -> None:
        """Initialize the BestPromptCallback."""
        self.best_prompt = ""
        self.best_score = -99999.0

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Update the best prompt and score if a new high score is achieved.

        Args:
        optimizer: The optimizer object that called the callback.
        """
        if optimizer.scores[0] > self.best_score:
            self.best_score = optimizer.scores[0]
            self.best_prompt = optimizer.prompts[0]

        return True

    def get_best_prompt(self) -> Tuple[str, float]:
        """Get the best prompt and score achieved during optimization.

        Returns:
        Tuple[str, float]: The best prompt and score.
        """
        return self.best_prompt, self.best_score

__init__()

Initialize the BestPromptCallback.

Source code in promptolution/utils/callbacks.py 
183
184
185
186
def __init__(self) -> None:
    """Initialize the BestPromptCallback."""
    self.best_prompt = ""
    self.best_score = -99999.0

get_best_prompt()

Get the best prompt and score achieved during optimization.

Returns: Tuple[str, float]: The best prompt and score.

Source code in promptolution/utils/callbacks.py 
200
201
202
203
204
205
206
def get_best_prompt(self) -> Tuple[str, float]:
    """Get the best prompt and score achieved during optimization.

    Returns:
    Tuple[str, float]: The best prompt and score.
    """
    return self.best_prompt, self.best_score

on_step_end(optimizer)

Update the best prompt and score if a new high score is achieved.

Args: optimizer: The optimizer object that called the callback.

Source code in promptolution/utils/callbacks.py 
188
189
190
191
192
193
194
195
196
197
198
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Update the best prompt and score if a new high score is achieved.

    Args:
    optimizer: The optimizer object that called the callback.
    """
    if optimizer.scores[0] > self.best_score:
        self.best_score = optimizer.scores[0]
        self.best_prompt = optimizer.prompts[0]

    return True

FileOutputCallback

Bases: BaseCallback

Callback for saving optimization progress to a specified file type.

This callback saves information about each step to a file.

Attributes:

Name Type Description
dir Union[str, Path]

Directory the file is saved to.
step int

The current step number.
file_type str

The type of file to save the output to.
Source code in promptolution/utils/callbacks.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class FileOutputCallback(BaseCallback):
    """Callback for saving optimization progress to a specified file type.

    This callback saves information about each step to a file.

    Attributes:
        dir (Union[str, Path]): Directory the file is saved to.
        step (int): The current step number.
        file_type (str): The type of file to save the output to.
    """

    def __init__(self, dir: Union[str, Path], file_type: Literal["parquet", "csv"] = "parquet") -> None:
        """Initialize the FileOutputCallback.

        Args:
        dir (Union[str, Path]): Directory the CSV file is saved to.
        file_type (str): The type of file to save the output to.
        """
        if not os.path.exists(dir):
            os.makedirs(dir)

        self.file_type = file_type

        if file_type == "parquet":
            self.path = os.path.join(dir, "step_results.parquet")
        elif file_type == "csv":
            self.path = os.path.join(dir, "step_results.csv")
        else:
            raise ValueError(f"File type {file_type} not supported.")

        self.step = 0
        self.rows: list = []

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Append this step's rows and rewrite the results file.

        Args:
        optimizer: The optimizer object that called the callback
        """
        self.step += 1
        ts = datetime.now().timestamp()
        for score, prompt in zip(optimizer.scores, optimizer.prompts):
            self.rows.append(
                {
                    "step": self.step,
                    "input_tokens": optimizer.predictor.llm.input_token_count,
                    "output_tokens": optimizer.predictor.llm.output_token_count,
                    "time": ts,
                    "score": score,
                    "prompt": str(prompt),
                }
            )
        # rewrite the whole file each step -> engine-agnostic (no fastparquet-only append)
        df = pd.DataFrame(self.rows)
        if self.file_type == "parquet":
            df.to_parquet(self.path, index=False)
        else:
            df.to_csv(self.path, index=False)
        return True

__init__(dir, file_type='parquet')

Initialize the FileOutputCallback.

Args: dir (Union[str, Path]): Directory the CSV file is saved to. file_type (str): The type of file to save the output to.

Source code in promptolution/utils/callbacks.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def __init__(self, dir: Union[str, Path], file_type: Literal["parquet", "csv"] = "parquet") -> None:
    """Initialize the FileOutputCallback.

    Args:
    dir (Union[str, Path]): Directory the CSV file is saved to.
    file_type (str): The type of file to save the output to.
    """
    if not os.path.exists(dir):
        os.makedirs(dir)

    self.file_type = file_type

    if file_type == "parquet":
        self.path = os.path.join(dir, "step_results.parquet")
    elif file_type == "csv":
        self.path = os.path.join(dir, "step_results.csv")
    else:
        raise ValueError(f"File type {file_type} not supported.")

    self.step = 0
    self.rows: list = []

on_step_end(optimizer)

Append this step's rows and rewrite the results file.

Args: optimizer: The optimizer object that called the callback

Source code in promptolution/utils/callbacks.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Append this step's rows and rewrite the results file.

    Args:
    optimizer: The optimizer object that called the callback
    """
    self.step += 1
    ts = datetime.now().timestamp()
    for score, prompt in zip(optimizer.scores, optimizer.prompts):
        self.rows.append(
            {
                "step": self.step,
                "input_tokens": optimizer.predictor.llm.input_token_count,
                "output_tokens": optimizer.predictor.llm.output_token_count,
                "time": ts,
                "score": score,
                "prompt": str(prompt),
            }
        )
    # rewrite the whole file each step -> engine-agnostic (no fastparquet-only append)
    df = pd.DataFrame(self.rows)
    if self.file_type == "parquet":
        df.to_parquet(self.path, index=False)
    else:
        df.to_csv(self.path, index=False)
    return True

LoggerCallback

Bases: BaseCallback

Callback for logging optimization progress.

This callback logs information about each step, epoch, and the end of training.

Attributes:

Name Type Description
logger

The logger object to use for logging.
step int

The current step number.
Source code in promptolution/utils/callbacks.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class LoggerCallback(BaseCallback):
    """Callback for logging optimization progress.

    This callback logs information about each step, epoch, and the end of training.

    Attributes:
        logger: The logger object to use for logging.
        step (int): The current step number.
    """

    def __init__(self, logger: "Logger") -> None:
        """Initialize the LoggerCallback."""
        self.logger = logger
        self.step = 0

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Log information about the current step."""
        self.step += 1
        time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
        self.logger.critical(f"{time} - ✨ Step {self.step} ended ✨")
        for i, (prompt, score) in enumerate(zip(optimizer.prompts, optimizer.scores)):
            self.logger.critical(f"📝 Prompt {i}: Score: {score}")
            self.logger.critical(f"💬 {prompt}")

        return True

    def on_train_end(self, optimizer: "BaseOptimizer", logs: Optional[Any] = None) -> bool:
        """Log information at the end of training.

        Args:
        optimizer: The optimizer object that called the callback.
        logs: Additional information to log.
        """
        time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
        if logs is None:
            self.logger.critical(f"{time} - 🏁 Training ended")
        else:
            self.logger.critical(f"{time} - 🏁 Training ended - {logs}")

        return True

__init__(logger)

Initialize the LoggerCallback.

Source code in promptolution/utils/callbacks.py 
80
81
82
83
def __init__(self, logger: "Logger") -> None:
    """Initialize the LoggerCallback."""
    self.logger = logger
    self.step = 0

on_step_end(optimizer)

Log information about the current step.

Source code in promptolution/utils/callbacks.py 
85
86
87
88
89
90
91
92
93
94
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Log information about the current step."""
    self.step += 1
    time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
    self.logger.critical(f"{time} - ✨ Step {self.step} ended ✨")
    for i, (prompt, score) in enumerate(zip(optimizer.prompts, optimizer.scores)):
        self.logger.critical(f"📝 Prompt {i}: Score: {score}")
        self.logger.critical(f"💬 {prompt}")

    return True

on_train_end(optimizer, logs=None)

Log information at the end of training.

Args: optimizer: The optimizer object that called the callback. logs: Additional information to log.

Source code in promptolution/utils/callbacks.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def on_train_end(self, optimizer: "BaseOptimizer", logs: Optional[Any] = None) -> bool:
    """Log information at the end of training.

    Args:
    optimizer: The optimizer object that called the callback.
    logs: Additional information to log.
    """
    time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
    if logs is None:
        self.logger.critical(f"{time} - 🏁 Training ended")
    else:
        self.logger.critical(f"{time} - 🏁 Training ended - {logs}")

    return True

ProgressBarCallback

Bases: BaseCallback

Callback for displaying a progress bar during optimization.

This callback uses tqdm to display a progress bar that updates at each step.

Attributes:

Name Type Description
pbar tqdm

The tqdm progress bar object.
Source code in promptolution/utils/callbacks.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class ProgressBarCallback(BaseCallback):
    """Callback for displaying a progress bar during optimization.

    This callback uses tqdm to display a progress bar that updates at each step.

    Attributes:
        pbar (tqdm): The tqdm progress bar object.
    """

    def __init__(self, total_steps: int) -> None:
        """Initialize the ProgressBarCallback.

        Args:
        total_steps (int): The total number of steps in the optimization process.
        """
        self.pbar = tqdm(total=total_steps)

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Update the progress bar at the end of each step.

        Args:
        optimizer: The optimizer object that called the callback.
        """
        self.pbar.update(1)

        return True

    def on_train_end(self, optimizer: "BaseOptimizer") -> bool:
        """Close the progress bar at the end of training.

        Args:
        optimizer: The optimizer object that called the callback.
        """
        self.pbar.close()

        return True

__init__(total_steps)

Initialize the ProgressBarCallback.

Args: total_steps (int): The total number of steps in the optimization process.

Source code in promptolution/utils/callbacks.py 
218
219
220
221
222
223
224
def __init__(self, total_steps: int) -> None:
    """Initialize the ProgressBarCallback.

    Args:
    total_steps (int): The total number of steps in the optimization process.
    """
    self.pbar = tqdm(total=total_steps)

on_step_end(optimizer)

Update the progress bar at the end of each step.

Args: optimizer: The optimizer object that called the callback.

Source code in promptolution/utils/callbacks.py 
226
227
228
229
230
231
232
233
234
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Update the progress bar at the end of each step.

    Args:
    optimizer: The optimizer object that called the callback.
    """
    self.pbar.update(1)

    return True

on_train_end(optimizer)

Close the progress bar at the end of training.

Args: optimizer: The optimizer object that called the callback.

Source code in promptolution/utils/callbacks.py 
236
237
238
239
240
241
242
243
244
def on_train_end(self, optimizer: "BaseOptimizer") -> bool:
    """Close the progress bar at the end of training.

    Args:
    optimizer: The optimizer object that called the callback.
    """
    self.pbar.close()

    return True

Prompt

Represent a prompt consisting of an instruction and few-shot examples.

Source code in promptolution/utils/prompt.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class Prompt:
    """Represent a prompt consisting of an instruction and few-shot examples."""

    def __init__(
        self, instruction: str, few_shots: Optional[List[str]] = None, downstream_template: Optional[str] = None
    ) -> None:
        """Initialize the Prompt with an instruction and associated examples.

        Args:
            instruction (str): The instruction or prompt text.
            few_shots (List[str]): List of examples as string.
            downstream_template (str, optional): Template for formatting the full prompt.
        """
        self.instruction = instruction.strip()
        self.few_shots = few_shots or []
        if downstream_template is None:
            if self.few_shots:
                downstream_template = DOWNSTREAM_TEMPLATE_W_FEWSHOTS
            else:
                downstream_template = DOWNSTREAM_TEMPLATE
        self.downstream_template = downstream_template

    def construct_prompt(self) -> str:
        """Construct the full prompt string by replacing placeholders in the template with the instruction and formatted examples.

        Returns:
            str: The constructed prompt string.
        """
        few_shot_str = "\n\n".join(self.few_shots).strip()
        prompt = (
            self.downstream_template.replace("<instruction>", self.instruction)
            .replace("<few_shots>", few_shot_str)
            .replace("\n\n\n\n", "\n\n")  # replace extra newlines if no few shots are provided
            .strip()
        )
        return prompt

    def __str__(self) -> str:
        """Return the string representation of the prompt."""
        return self.construct_prompt()

    def __eq__(self, other: object) -> bool:
        """Structural equality for use in lists, sets, and dict keys."""
        if not isinstance(other, Prompt):
            return False
        return (
            self.instruction == other.instruction
            and self.few_shots == other.few_shots
            and self.downstream_template == other.downstream_template
        )

    def __hash__(self) -> int:
        """Hash function for use in sets and dict keys."""
        return hash((self.instruction, tuple(self.few_shots), self.downstream_template))

__eq__(other)

Structural equality for use in lists, sets, and dict keys.

Source code in promptolution/utils/prompt.py 
51
52
53
54
55
56
57
58
59
def __eq__(self, other: object) -> bool:
    """Structural equality for use in lists, sets, and dict keys."""
    if not isinstance(other, Prompt):
        return False
    return (
        self.instruction == other.instruction
        and self.few_shots == other.few_shots
        and self.downstream_template == other.downstream_template
    )

__hash__()

Hash function for use in sets and dict keys.

Source code in promptolution/utils/prompt.py 
61
62
63
def __hash__(self) -> int:
    """Hash function for use in sets and dict keys."""
    return hash((self.instruction, tuple(self.few_shots), self.downstream_template))

__init__(instruction, few_shots=None, downstream_template=None)

Initialize the Prompt with an instruction and associated examples.

Parameters:

Name Type Description Default
instruction str

The instruction or prompt text.

 required
few_shots List[str]

List of examples as string.

 None
downstream_template str

Template for formatting the full prompt.

 None
Source code in promptolution/utils/prompt.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self, instruction: str, few_shots: Optional[List[str]] = None, downstream_template: Optional[str] = None
) -> None:
    """Initialize the Prompt with an instruction and associated examples.

    Args:
        instruction (str): The instruction or prompt text.
        few_shots (List[str]): List of examples as string.
        downstream_template (str, optional): Template for formatting the full prompt.
    """
    self.instruction = instruction.strip()
    self.few_shots = few_shots or []
    if downstream_template is None:
        if self.few_shots:
            downstream_template = DOWNSTREAM_TEMPLATE_W_FEWSHOTS
        else:
            downstream_template = DOWNSTREAM_TEMPLATE
    self.downstream_template = downstream_template

__str__()

Return the string representation of the prompt.

Source code in promptolution/utils/prompt.py 
47
48
49
def __str__(self) -> str:
    """Return the string representation of the prompt."""
    return self.construct_prompt()

construct_prompt()

Construct the full prompt string by replacing placeholders in the template with the instruction and formatted examples.

Returns:

Name Type Description
str str

The constructed prompt string.
Source code in promptolution/utils/prompt.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def construct_prompt(self) -> str:
    """Construct the full prompt string by replacing placeholders in the template with the instruction and formatted examples.

    Returns:
        str: The constructed prompt string.
    """
    few_shot_str = "\n\n".join(self.few_shots).strip()
    prompt = (
        self.downstream_template.replace("<instruction>", self.instruction)
        .replace("<few_shots>", few_shot_str)
        .replace("\n\n\n\n", "\n\n")  # replace extra newlines if no few shots are provided
        .strip()
    )
    return prompt

TokenCountCallback

Bases: BaseCallback

Callback for stopping optimization based on the total token count.

Source code in promptolution/utils/callbacks.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class TokenCountCallback(BaseCallback):
    """Callback for stopping optimization based on the total token count."""

    def __init__(
        self,
        max_tokens_for_termination: int,
        token_type_for_termination: Literal["input_tokens", "output_tokens", "total_tokens"],
    ) -> None:
        """Initialize the TokenCountCallback.

        Args:
        max_tokens_for_termination (int): Maximum number of tokens which is allowed befor the algorithm is stopped.
        token_type_for_termination (str): Can be one of either "input_tokens", "output_tokens" or "total_tokens".
        """
        self.max_tokens_for_termination = max_tokens_for_termination
        self.token_type_for_termination = token_type_for_termination

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Check if the total token count exceeds the maximum allowed. If so, stop the optimization."""
        token_counts = optimizer.predictor.llm.get_token_count()

        if token_counts[self.token_type_for_termination] > self.max_tokens_for_termination:
            return False

        return True

__init__(max_tokens_for_termination, token_type_for_termination)

Initialize the TokenCountCallback.

Args: max_tokens_for_termination (int): Maximum number of tokens which is allowed befor the algorithm is stopped. token_type_for_termination (str): Can be one of either "input_tokens", "output_tokens" or "total_tokens".

Source code in promptolution/utils/callbacks.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
def __init__(
    self,
    max_tokens_for_termination: int,
    token_type_for_termination: Literal["input_tokens", "output_tokens", "total_tokens"],
) -> None:
    """Initialize the TokenCountCallback.

    Args:
    max_tokens_for_termination (int): Maximum number of tokens which is allowed befor the algorithm is stopped.
    token_type_for_termination (str): Can be one of either "input_tokens", "output_tokens" or "total_tokens".
    """
    self.max_tokens_for_termination = max_tokens_for_termination
    self.token_type_for_termination = token_type_for_termination

on_step_end(optimizer)

Check if the total token count exceeds the maximum allowed. If so, stop the optimization.

Source code in promptolution/utils/callbacks.py 
264
265
266
267
268
269
270
271
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Check if the total token count exceeds the maximum allowed. If so, stop the optimization."""
    token_counts = optimizer.predictor.llm.get_token_count()

    if token_counts[self.token_type_for_termination] > self.max_tokens_for_termination:
        return False

    return True

create_prompt_variation(prompt, llm, meta_prompt=None)

Generate a variation of the given prompt(s) while keeping the semantic meaning.

Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910

Parameters:

Name Type Description Default
prompt Union[List[str], str]

The prompt(s) to generate variations of.

 required
llm BaseLLM

The language model to use for generating the variations.

 required
meta_prompt str

The meta prompt to use for generating the variations. If None, a default meta prompt is used. Should contain tag.

 None

Returns:

Type Description
List[str]

List[str]: A list of generated variations of the input prompt(s).
Source code in promptolution/utils/prompt_creation.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def create_prompt_variation(
    prompt: Union[List[str], str], llm: "BaseLLM", meta_prompt: Optional[str] = None
) -> List[str]:
    """Generate a variation of the given prompt(s) while keeping the semantic meaning.

    Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910

    Args:
        prompt (Union[List[str], str]): The prompt(s) to generate variations of.
        llm (BaseLLM): The language model to use for generating the variations.
        meta_prompt (str): The meta prompt to use for generating the variations.
            If None, a default meta prompt is used. Should contain <prev_prompt> tag.

    Returns:
        List[str]: A list of generated variations of the input prompt(s).
    """
    meta_prompt = PROMPT_VARIATION_TEMPLATE if meta_prompt is None else meta_prompt

    if isinstance(prompt, str):
        prompt = [prompt]
    varied_prompts = llm.get_response([meta_prompt.replace("<prev_prompt>", p) for p in prompt])
    varied_prompts = extract_from_tag(varied_prompts, "<prompt>", "</prompt>")

    return varied_prompts

create_prompts_from_samples(task, llm, meta_prompt=None, n_samples=3, task_description=None, n_prompts=1, get_uniform_labels=False)

Generate a set of prompts from dataset examples sampled from a given task.

Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910 Samples are selected, such that (1) all possible classes are represented (2) the samples are as representative as possible

Parameters:

Name Type Description Default
task BaseTask

The task to generate prompts for. Xs and Ys from this object are used to generate the prompts.

 required
llm BaseLLM

The language model to use for generating the prompts.

 required
meta_prompt str

The meta prompt to use for generating the prompts. If None, a default meta prompt is used.

 None
n_samples int

The number of samples to use for generating prompts.

 3
task_description str

The description of the task to include in the prompt.

 None
n_prompts int

The number of prompts to generate.

 1
get_uniform_labels bool

If True, samples are selected such that all classes are represented.

 False

Returns:

Type Description
List[str]

List[str]: A list of generated prompts.
Source code in promptolution/utils/prompt_creation.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def create_prompts_from_samples(
    task: "BaseTask",
    llm: "BaseLLM",
    meta_prompt: Optional[str] = None,
    n_samples: int = 3,
    task_description: Optional[str] = None,
    n_prompts: int = 1,
    get_uniform_labels: bool = False,
) -> List[str]:
    """Generate a set of prompts from dataset examples sampled from a given task.

    Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910
    Samples are selected, such that
    (1) all possible classes are represented
    (2) the samples are as representative as possible

    Args:
        task (BaseTask): The task to generate prompts for.
            Xs and Ys from this object are used to generate the prompts.
        llm (BaseLLM): The language model to use for generating the prompts.
        meta_prompt (str): The meta prompt to use for generating the prompts.
            If None, a default meta prompt is used.
        n_samples (int): The number of samples to use for generating prompts.
        task_description (str): The description of the task to include in the prompt.
        n_prompts (int): The number of prompts to generate.
        get_uniform_labels (bool): If True, samples are selected such that all classes are represented.

    Returns:
        List[str]: A list of generated prompts.
    """
    meta_prompt_template: str
    if meta_prompt is None:
        if task_description is None:
            meta_prompt_template = PROMPT_CREATION_TEMPLATE
        else:
            meta_prompt_template = PROMPT_CREATION_TEMPLATE_TD.replace("<task_desc>", task_description)
    else:
        if task_description is None:
            meta_prompt_template = meta_prompt
        else:
            meta_prompt_template = meta_prompt.replace("<task_desc>", task_description)

    meta_prompts = []
    for _ in range(n_prompts):
        if task.task_type == "classification" and get_uniform_labels:
            # if classification task sample such that all classes are represented
            unique_labels, counts = np.unique(task.ys, return_counts=True)
            proportions = counts / len(task.ys)
            samples_per_class = np.round(proportions * n_samples).astype(int)
            samples_per_class = np.maximum(samples_per_class, 1)

            # sample
            xs: List[str] = []
            ys: List[str] = []
            for label, n_per_class in zip(unique_labels, samples_per_class):
                indices = np.where(task.ys == label)[0]
                indices = np.random.choice(indices, n_per_class, replace=False)
                xs.extend(task.xs[indices])
                ys.extend(task.ys[indices])

        else:
            # if not classification task, sample randomly
            indices = np.random.choice(len(task.xs), n_samples, replace=False)
            xs = [task.xs[i] for i in indices]
            ys = [task.ys[i] for i in indices]

        examples = "\n\n".join([f"Input: {x}\nOutput: {y}" for x, y in zip(xs, ys)])
        meta_prompt = meta_prompt_template.replace("<input_output_pairs>", examples)
        meta_prompts.append(meta_prompt)

    prompts = llm.get_response(meta_prompts)
    prompts = extract_from_tag(prompts, "<prompt>", "</prompt>")

    return prompts

create_prompts_from_task_description(task_description, llm, meta_prompt=None, n_prompts=10, n_retries=3)

Generate a set of prompts from a given task description.

Parameters:

Name Type Description Default
task_description str

The description of the task to generate prompts for.

 required
llm BaseLLM

The language model to use for generating the prompts.

 required
meta_prompt str

The meta prompt to use for generating the prompts. If None, a default meta prompt is used.

 None
n_prompts int

The number of prompts to generate.

 10
n_retries int

The number of retries to attempt if prompt generation fails.

 3
Source code in promptolution/utils/prompt_creation.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def create_prompts_from_task_description(
    task_description: str,
    llm: "BaseLLM",
    meta_prompt: Optional[str] = None,
    n_prompts: int = 10,
    n_retries: int = 3,
) -> List[str]:
    """Generate a set of prompts from a given task description.

    Args:
        task_description (str): The description of the task to generate prompts for.
        llm (BaseLLM): The language model to use for generating the prompts.
        meta_prompt (str): The meta prompt to use for generating the prompts.
            If None, a default meta prompt is used.
        n_prompts (int): The number of prompts to generate.
        n_retries (int): The number of retries to attempt if prompt generation fails.
    """
    if meta_prompt is None:
        meta_prompt = PROMPT_CREATION_TEMPLATE_FROM_TASK_DESCRIPTION

    meta_prompt = meta_prompt.replace("<task_desc>", task_description).replace("<n_prompts>", str(n_prompts))
    final_prompts = None
    for _ in range(n_retries):
        prompts_str = llm.get_response(meta_prompt)[0]
        try:
            prompts = json.loads(prompts_str)
            assert isinstance(prompts, list) and all(isinstance(p, str) for p in prompts) and len(prompts) == n_prompts
            final_prompts = prompts
            break
        except (json.JSONDecodeError, AssertionError):
            logger.warning("Failed to parse prompts JSON, retrying...")

    if final_prompts is None:
        logger.error(
            f"Failed to generate prompts from task description after {n_retries} retries, returning default prompts."
        )
        final_prompts = default_prompts[:n_prompts]

    return final_prompts

dev_test_split(df, test_frac=0.2, seed=42)

Split a DataFrame into a dev and test set.

The optimizer selects prompts on the dev set. The test set is held out for the final evaluation.

Parameters:

Name Type Description Default
df DataFrame

The data to split (or a df-like with .to_pandas()).

 required
test_frac float

Fraction of rows held out for the test split.

 0.2
seed int

Random seed for the split.

 42

Returns:

Type Description
Tuple[DataFrame, DataFrame]

Tuple[pd.DataFrame, pd.DataFrame]: (dev_df, test_df), both with a fresh index.
Source code in promptolution/utils/evaluation.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def dev_test_split(df: pd.DataFrame, test_frac: float = 0.2, seed: int = 42) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """Split a DataFrame into a dev and test set.

    The optimizer selects prompts on the dev set. The test set is held out for the final evaluation.

    Args:
        df (pd.DataFrame): The data to split (or a df-like with ``.to_pandas()``).
        test_frac (float): Fraction of rows held out for the test split.
        seed (int): Random seed for the split.

    Returns:
        Tuple[pd.DataFrame, pd.DataFrame]: ``(dev_df, test_df)``, both with a fresh index.
    """
    if not isinstance(df, pd.DataFrame):
        df = df.to_pandas()
    test_df = df.sample(frac=test_frac, random_state=seed)
    dev_df = df.drop(test_df.index)
    return dev_df.reset_index(drop=True), test_df.reset_index(drop=True)

evaluate_prompts(prompts, task, predictor)

Score prompts on a task and return a sorted prompt/score table.

Parameters:

Name Type Description Default
prompts Union[List[Prompt], List[str]]

Prompts to score.

 required
task BaseTask

The task to score on (its full dataset is used).

 required
predictor BasePredictor

The predictor that runs the prompts.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: Columns prompt and score, best first.
Source code in promptolution/utils/evaluation.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def evaluate_prompts(
    prompts: Union[List[Prompt], List[str]], task: "BaseTask", predictor: "BasePredictor"
) -> pd.DataFrame:
    """Score prompts on a task and return a sorted prompt/score table.

    Args:
        prompts (Union[List[Prompt], List[str]]): Prompts to score.
        task (BaseTask): The task to score on (its full dataset is used).
        predictor (BasePredictor): The predictor that runs the prompts.

    Returns:
        pd.DataFrame: Columns ``prompt`` and ``score``, best first.
    """
    if isinstance(prompts[0], str):
        str_prompts = cast(List[str], list(prompts))
        prompt_objs = [Prompt(p) for p in str_prompts]
    else:
        prompt_objs = cast(List[Prompt], list(prompts))
        str_prompts = [p.construct_prompt() for p in prompt_objs]
    logger.warning("📊 Starting evaluation...")
    results = task.evaluate(prompt_objs, predictor, eval_strategy="full")
    return pd.DataFrame({"prompt": str_prompts, "score": results.agg_scores.tolist()}).sort_values(
        "score", ascending=False, ignore_index=True
    )

get_logger(name, level=None)

Get a logger with the specified name and level.

This function provides a standardized way to get loggers throughout the library, ensuring consistent formatting and behavior.

Parameters:

Name Type Description Default
name str

Name of the logger, typically name of the module.

 required
level int

Logging level. Defaults to None, which uses the root logger's level.

 None

Returns:

Type Description
Logger

logging.Logger: Configured logger instance.
Source code in promptolution/utils/logging.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def get_logger(name: str, level: Optional[int] = None) -> logging.Logger:
    """Get a logger with the specified name and level.

    This function provides a standardized way to get loggers throughout the library,
    ensuring consistent formatting and behavior.

    Args:
        name (str): Name of the logger, typically __name__ of the module.
        level (int, optional): Logging level. Defaults to None, which uses the root logger's level.

    Returns:
        logging.Logger: Configured logger instance.
    """
    logger = logging.getLogger(name)
    if level is not None:
        logger.setLevel(level)
    return logger

get_test_statistic_func(name)

Get the test statistic function based on the name provided.

Parameters:

Name Type Description Default
name str

Name of the test statistic function.

 required

Returns:

Name Type Description
callable Callable[..., bool]

The corresponding test statistic function.
Source code in promptolution/utils/test_statistics.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_test_statistic_func(name: TestStatistics) -> Callable[..., bool]:
    """
    Get the test statistic function based on the name provided.

    Args:
        name (str): Name of the test statistic function.

    Returns:
        callable: The corresponding test statistic function.
    """
    if name == "paired_t_test":
        return paired_t_test
    else:
        raise ValueError(f"Unknown test statistic function: {name}. Should be one of {TestStatistics.__args__}.")

get_token_counter(llm)

Get a token counter function for the given LLM.

This function returns a callable that counts tokens based on the LLM's tokenizer or a simple split method if no tokenizer is available.

Parameters:

Name Type Description Default
llm BaseLLM

The language model object that may have a tokenizer.

 required

Returns:

Type Description
Callable[[str], float]

A callable that takes a text input and returns the token count.
Source code in promptolution/utils/token_counter.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def get_token_counter(llm: "BaseLLM") -> Callable[[str], float]:
    """Get a token counter function for the given LLM.

    This function returns a callable that counts tokens based on the LLM's tokenizer
    or a simple split method if no tokenizer is available.

    Args:
        llm: The language model object that may have a tokenizer.

    Returns:
        A callable that takes a text input and returns the token count.

    """
    if llm.tokenizer is not None:
        tokenizer: "PreTrainedTokenizer" = llm.tokenizer
        return lambda x: float(len(tokenizer.encode(x)))
    else:
        logger.warning("⚠️ The LLM does not have a tokenizer. Using simple token count.")
        return lambda x: float(len(x.split()))

paired_t_test(scores_a, scores_b, alpha=0.05)

Uses a paired t-test to test if candidate A's accuracy is significantly higher than candidate B's accuracy within a confidence interval of 1-lpha. Assumptions: - The samples are paired. - The differences between the pairs are normally distributed (-> n > 30).

Parameters:

Name Type Description Default
scores_a List[float]

Array of accuracy scores for candidate A.

 required
scores_b List[float]

Array of accuracy scores for candidate B.

 required
alpha float

Significance level (default 0.05 for 95% confidence).

 0.05

Returns:

Name Type Description
bool bool

True if candidate A is significantly better than candidate B, False otherwise.
Source code in promptolution/utils/test_statistics.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def paired_t_test(
    scores_a: List[float],
    scores_b: List[float],
    alpha: float = 0.05,
) -> bool:
    """
    Uses a paired t-test to test if candidate A's accuracy is significantly
    higher than candidate B's accuracy within a confidence interval of 1-\alpha.
    Assumptions:
    - The samples are paired.
    - The differences between the pairs are normally distributed (-> n > 30).

    Parameters:
        scores_a (List[float]): Array of accuracy scores for candidate A.
        scores_b (List[float]): Array of accuracy scores for candidate B.
        alpha (float): Significance level (default 0.05 for 95% confidence).

    Returns:
        bool: True if candidate A is significantly better than candidate B, False otherwise.
    """
    scores_a = np.array(scores_a)
    scores_b = np.array(scores_b)

    _, p_value = ttest_rel(scores_a, scores_b, alternative="greater")

    result = p_value < alpha

    return bool(result)

setup_logging(level=logging.INFO)

Set up logging for the promptolution library.

This function configures the root logger for the library with appropriate formatting and level.

Parameters:

Name Type Description Default
level int

Logging level. Defaults to logging.INFO.

 INFO
Source code in promptolution/utils/logging.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def setup_logging(level: int = logging.INFO) -> None:
    """Set up logging for the promptolution library.

    This function configures the root logger for the library with appropriate
    formatting and level.

    Args:
        level (int, optional): Logging level. Defaults to logging.INFO.
    """
    # Configure the root logger
    logging.basicConfig(
        level=level,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

sort_prompts_by_scores(prompts, scores, top_k=None)

Sort prompts by score, accepting scalar, 1D, or multi-dimensional scores.

Scores can be provided as Python lists or NumPy arrays. If scores are multi- dimensional (e.g., per-subsample results), they are aggregated with a nanmean across all non-leading axes before sorting.

Parameters:

Name Type Description Default
prompts List[Prompt]

Prompt objects to sort.

 required
scores Sequence[float] | ndarray

Corresponding scores; can be nested lists or arrays.

 required
top_k Optional[int]

Limit the result to the top_k prompts.

 None

Returns:

Type Description
List[Prompt]

Tuple[List[Prompt], List[float]]: Prompts and their aggregated scores,
List[float]

sorted in descending order.
Source code in promptolution/utils/prompt.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def sort_prompts_by_scores(
    prompts: List[Prompt], scores: Union[Sequence[float], np.ndarray], top_k: Optional[int] = None
) -> Tuple[List[Prompt], List[float]]:
    """Sort prompts by score, accepting scalar, 1D, or multi-dimensional scores.

    Scores can be provided as Python lists or NumPy arrays. If scores are multi-
    dimensional (e.g., per-subsample results), they are aggregated with a
    ``nanmean`` across all non-leading axes before sorting.

    Args:
        prompts (List[Prompt]): Prompt objects to sort.
        scores (Sequence[float] | np.ndarray): Corresponding scores; can be nested lists or arrays.
        top_k (Optional[int]): Limit the result to the top_k prompts.

    Returns:
        Tuple[List[Prompt], List[float]]: Prompts and their aggregated scores,
        sorted in descending order.
    """
    scores_arr = np.asarray(scores, dtype=float)
    if scores_arr.ndim == 0:
        scores_arr = scores_arr.reshape(1)

    assert scores_arr.shape[0] == len(prompts), "Prompts and scores must have the same length."

    if scores_arr.ndim > 1:
        axes_to_reduce = tuple(range(1, scores_arr.ndim))
        scores_arr = np.nanmean(scores_arr, axis=axes_to_reduce)

    prompt_score_pairs = list(zip(prompts, scores_arr.tolist()))
    prompt_score_pairs.sort(key=lambda pair: pair[1], reverse=True)

    if top_k is not None:
        prompt_score_pairs = prompt_score_pairs[:top_k]

    sorted_prompts = [p for p, _ in prompt_score_pairs]
    sorted_scores = [s for _, s in prompt_score_pairs]

    return sorted_prompts, sorted_scores

callbacks

Callback classes for logging, saving, and tracking optimization progress.

BaseCallback

Bases: ABC

Base class for optimization callbacks.

Callbacks can be used to monitor the optimization process, save checkpoints, log metrics, or implement early stopping criteria.

Source code in promptolution/utils/callbacks.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class BaseCallback(ABC):
    """Base class for optimization callbacks.

    Callbacks can be used to monitor the optimization process, save checkpoints,
    log metrics, or implement early stopping criteria.

    """

    def __init__(self, **kwargs: Any) -> None:
        """Initialize the callback.

        Args:
            **kwargs: Additional keyword arguments.
        """
        pass

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Call at the end of each optimization step.

        Args:
            optimizer: The optimizer object that called the callback.

        Returns:
            Bool: True if the optimization should continue, False if it should stop.
        """
        return True

    def on_epoch_end(self, optimizer: "BaseOptimizer") -> bool:
        """Call at the end of each optimization epoch.

        Args:
            optimizer: The optimizer object that called the callback.

        Returns:
            Bool: True if the optimization should continue, False if it should stop.
        """
        return True

    def on_train_end(self, optimizer: "BaseOptimizer") -> bool:
        """Call at the end of the entire optimization process.

        Args:
            optimizer: The optimizer object that called the callback.

        Returns:
            Bool: True if the optimization should continue, False if it should stop.
        """
        return True
__init__(**kwargs)

Initialize the callback.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments.

 {}
Source code in promptolution/utils/callbacks.py 
28
29
30
31
32
33
34
def __init__(self, **kwargs: Any) -> None:
    """Initialize the callback.

    Args:
        **kwargs: Additional keyword arguments.
    """
    pass
on_epoch_end(optimizer)

Call at the end of each optimization epoch.

Parameters:

Name Type Description Default
optimizer BaseOptimizer

The optimizer object that called the callback.

 required

Returns:

Name Type Description
Bool bool

True if the optimization should continue, False if it should stop.
Source code in promptolution/utils/callbacks.py 
47
48
49
50
51
52
53
54
55
56
def on_epoch_end(self, optimizer: "BaseOptimizer") -> bool:
    """Call at the end of each optimization epoch.

    Args:
        optimizer: The optimizer object that called the callback.

    Returns:
        Bool: True if the optimization should continue, False if it should stop.
    """
    return True
on_step_end(optimizer)

Call at the end of each optimization step.

Parameters:

Name Type Description Default
optimizer BaseOptimizer

The optimizer object that called the callback.

 required

Returns:

Name Type Description
Bool bool

True if the optimization should continue, False if it should stop.
Source code in promptolution/utils/callbacks.py 
36
37
38
39
40
41
42
43
44
45
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Call at the end of each optimization step.

    Args:
        optimizer: The optimizer object that called the callback.

    Returns:
        Bool: True if the optimization should continue, False if it should stop.
    """
    return True
on_train_end(optimizer)

Call at the end of the entire optimization process.

Parameters:

Name Type Description Default
optimizer BaseOptimizer

The optimizer object that called the callback.

 required

Returns:

Name Type Description
Bool bool

True if the optimization should continue, False if it should stop.
Source code in promptolution/utils/callbacks.py 
58
59
60
61
62
63
64
65
66
67
def on_train_end(self, optimizer: "BaseOptimizer") -> bool:
    """Call at the end of the entire optimization process.

    Args:
        optimizer: The optimizer object that called the callback.

    Returns:
        Bool: True if the optimization should continue, False if it should stop.
    """
    return True

BestPromptCallback

Bases: BaseCallback

Callback for tracking the best prompt during optimization.

This callback keeps track of the prompt with the highest score.

Attributes:

Name Type Description
best_prompt str

The prompt with the highest score so far.
best_score float

The highest score achieved so far.
Source code in promptolution/utils/callbacks.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class BestPromptCallback(BaseCallback):
    """Callback for tracking the best prompt during optimization.

    This callback keeps track of the prompt with the highest score.

    Attributes:
        best_prompt (str): The prompt with the highest score so far.
        best_score (float): The highest score achieved so far.
    """

    def __init__(self) -> None:
        """Initialize the BestPromptCallback."""
        self.best_prompt = ""
        self.best_score = -99999.0

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Update the best prompt and score if a new high score is achieved.

        Args:
        optimizer: The optimizer object that called the callback.
        """
        if optimizer.scores[0] > self.best_score:
            self.best_score = optimizer.scores[0]
            self.best_prompt = optimizer.prompts[0]

        return True

    def get_best_prompt(self) -> Tuple[str, float]:
        """Get the best prompt and score achieved during optimization.

        Returns:
        Tuple[str, float]: The best prompt and score.
        """
        return self.best_prompt, self.best_score
__init__()

Initialize the BestPromptCallback.

Source code in promptolution/utils/callbacks.py 
183
184
185
186
def __init__(self) -> None:
    """Initialize the BestPromptCallback."""
    self.best_prompt = ""
    self.best_score = -99999.0
get_best_prompt()

Get the best prompt and score achieved during optimization.

Returns: Tuple[str, float]: The best prompt and score.

Source code in promptolution/utils/callbacks.py 
200
201
202
203
204
205
206
def get_best_prompt(self) -> Tuple[str, float]:
    """Get the best prompt and score achieved during optimization.

    Returns:
    Tuple[str, float]: The best prompt and score.
    """
    return self.best_prompt, self.best_score
on_step_end(optimizer)

Update the best prompt and score if a new high score is achieved.

Args: optimizer: The optimizer object that called the callback.

Source code in promptolution/utils/callbacks.py 
188
189
190
191
192
193
194
195
196
197
198
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Update the best prompt and score if a new high score is achieved.

    Args:
    optimizer: The optimizer object that called the callback.
    """
    if optimizer.scores[0] > self.best_score:
        self.best_score = optimizer.scores[0]
        self.best_prompt = optimizer.prompts[0]

    return True

FileOutputCallback

Bases: BaseCallback

Callback for saving optimization progress to a specified file type.

This callback saves information about each step to a file.

Attributes:

Name Type Description
dir Union[str, Path]

Directory the file is saved to.
step int

The current step number.
file_type str

The type of file to save the output to.
Source code in promptolution/utils/callbacks.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class FileOutputCallback(BaseCallback):
    """Callback for saving optimization progress to a specified file type.

    This callback saves information about each step to a file.

    Attributes:
        dir (Union[str, Path]): Directory the file is saved to.
        step (int): The current step number.
        file_type (str): The type of file to save the output to.
    """

    def __init__(self, dir: Union[str, Path], file_type: Literal["parquet", "csv"] = "parquet") -> None:
        """Initialize the FileOutputCallback.

        Args:
        dir (Union[str, Path]): Directory the CSV file is saved to.
        file_type (str): The type of file to save the output to.
        """
        if not os.path.exists(dir):
            os.makedirs(dir)

        self.file_type = file_type

        if file_type == "parquet":
            self.path = os.path.join(dir, "step_results.parquet")
        elif file_type == "csv":
            self.path = os.path.join(dir, "step_results.csv")
        else:
            raise ValueError(f"File type {file_type} not supported.")

        self.step = 0
        self.rows: list = []

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Append this step's rows and rewrite the results file.

        Args:
        optimizer: The optimizer object that called the callback
        """
        self.step += 1
        ts = datetime.now().timestamp()
        for score, prompt in zip(optimizer.scores, optimizer.prompts):
            self.rows.append(
                {
                    "step": self.step,
                    "input_tokens": optimizer.predictor.llm.input_token_count,
                    "output_tokens": optimizer.predictor.llm.output_token_count,
                    "time": ts,
                    "score": score,
                    "prompt": str(prompt),
                }
            )
        # rewrite the whole file each step -> engine-agnostic (no fastparquet-only append)
        df = pd.DataFrame(self.rows)
        if self.file_type == "parquet":
            df.to_parquet(self.path, index=False)
        else:
            df.to_csv(self.path, index=False)
        return True
__init__(dir, file_type='parquet')

Initialize the FileOutputCallback.

Args: dir (Union[str, Path]): Directory the CSV file is saved to. file_type (str): The type of file to save the output to.

Source code in promptolution/utils/callbacks.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def __init__(self, dir: Union[str, Path], file_type: Literal["parquet", "csv"] = "parquet") -> None:
    """Initialize the FileOutputCallback.

    Args:
    dir (Union[str, Path]): Directory the CSV file is saved to.
    file_type (str): The type of file to save the output to.
    """
    if not os.path.exists(dir):
        os.makedirs(dir)

    self.file_type = file_type

    if file_type == "parquet":
        self.path = os.path.join(dir, "step_results.parquet")
    elif file_type == "csv":
        self.path = os.path.join(dir, "step_results.csv")
    else:
        raise ValueError(f"File type {file_type} not supported.")

    self.step = 0
    self.rows: list = []
on_step_end(optimizer)

Append this step's rows and rewrite the results file.

Args: optimizer: The optimizer object that called the callback

Source code in promptolution/utils/callbacks.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Append this step's rows and rewrite the results file.

    Args:
    optimizer: The optimizer object that called the callback
    """
    self.step += 1
    ts = datetime.now().timestamp()
    for score, prompt in zip(optimizer.scores, optimizer.prompts):
        self.rows.append(
            {
                "step": self.step,
                "input_tokens": optimizer.predictor.llm.input_token_count,
                "output_tokens": optimizer.predictor.llm.output_token_count,
                "time": ts,
                "score": score,
                "prompt": str(prompt),
            }
        )
    # rewrite the whole file each step -> engine-agnostic (no fastparquet-only append)
    df = pd.DataFrame(self.rows)
    if self.file_type == "parquet":
        df.to_parquet(self.path, index=False)
    else:
        df.to_csv(self.path, index=False)
    return True

LoggerCallback

Bases: BaseCallback

Callback for logging optimization progress.

This callback logs information about each step, epoch, and the end of training.

Attributes:

Name Type Description
logger

The logger object to use for logging.
step int

The current step number.
Source code in promptolution/utils/callbacks.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class LoggerCallback(BaseCallback):
    """Callback for logging optimization progress.

    This callback logs information about each step, epoch, and the end of training.

    Attributes:
        logger: The logger object to use for logging.
        step (int): The current step number.
    """

    def __init__(self, logger: "Logger") -> None:
        """Initialize the LoggerCallback."""
        self.logger = logger
        self.step = 0

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Log information about the current step."""
        self.step += 1
        time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
        self.logger.critical(f"{time} - ✨ Step {self.step} ended ✨")
        for i, (prompt, score) in enumerate(zip(optimizer.prompts, optimizer.scores)):
            self.logger.critical(f"📝 Prompt {i}: Score: {score}")
            self.logger.critical(f"💬 {prompt}")

        return True

    def on_train_end(self, optimizer: "BaseOptimizer", logs: Optional[Any] = None) -> bool:
        """Log information at the end of training.

        Args:
        optimizer: The optimizer object that called the callback.
        logs: Additional information to log.
        """
        time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
        if logs is None:
            self.logger.critical(f"{time} - 🏁 Training ended")
        else:
            self.logger.critical(f"{time} - 🏁 Training ended - {logs}")

        return True
__init__(logger)

Initialize the LoggerCallback.

Source code in promptolution/utils/callbacks.py 
80
81
82
83
def __init__(self, logger: "Logger") -> None:
    """Initialize the LoggerCallback."""
    self.logger = logger
    self.step = 0
on_step_end(optimizer)

Log information about the current step.

Source code in promptolution/utils/callbacks.py 
85
86
87
88
89
90
91
92
93
94
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Log information about the current step."""
    self.step += 1
    time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
    self.logger.critical(f"{time} - ✨ Step {self.step} ended ✨")
    for i, (prompt, score) in enumerate(zip(optimizer.prompts, optimizer.scores)):
        self.logger.critical(f"📝 Prompt {i}: Score: {score}")
        self.logger.critical(f"💬 {prompt}")

    return True
on_train_end(optimizer, logs=None)

Log information at the end of training.

Args: optimizer: The optimizer object that called the callback. logs: Additional information to log.

Source code in promptolution/utils/callbacks.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def on_train_end(self, optimizer: "BaseOptimizer", logs: Optional[Any] = None) -> bool:
    """Log information at the end of training.

    Args:
    optimizer: The optimizer object that called the callback.
    logs: Additional information to log.
    """
    time = datetime.now().strftime("%d-%m-%y %H:%M:%S:%f")
    if logs is None:
        self.logger.critical(f"{time} - 🏁 Training ended")
    else:
        self.logger.critical(f"{time} - 🏁 Training ended - {logs}")

    return True

ProgressBarCallback

Bases: BaseCallback

Callback for displaying a progress bar during optimization.

This callback uses tqdm to display a progress bar that updates at each step.

Attributes:

Name Type Description
pbar tqdm

The tqdm progress bar object.
Source code in promptolution/utils/callbacks.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class ProgressBarCallback(BaseCallback):
    """Callback for displaying a progress bar during optimization.

    This callback uses tqdm to display a progress bar that updates at each step.

    Attributes:
        pbar (tqdm): The tqdm progress bar object.
    """

    def __init__(self, total_steps: int) -> None:
        """Initialize the ProgressBarCallback.

        Args:
        total_steps (int): The total number of steps in the optimization process.
        """
        self.pbar = tqdm(total=total_steps)

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Update the progress bar at the end of each step.

        Args:
        optimizer: The optimizer object that called the callback.
        """
        self.pbar.update(1)

        return True

    def on_train_end(self, optimizer: "BaseOptimizer") -> bool:
        """Close the progress bar at the end of training.

        Args:
        optimizer: The optimizer object that called the callback.
        """
        self.pbar.close()

        return True
__init__(total_steps)

Initialize the ProgressBarCallback.

Args: total_steps (int): The total number of steps in the optimization process.

Source code in promptolution/utils/callbacks.py 
218
219
220
221
222
223
224
def __init__(self, total_steps: int) -> None:
    """Initialize the ProgressBarCallback.

    Args:
    total_steps (int): The total number of steps in the optimization process.
    """
    self.pbar = tqdm(total=total_steps)
on_step_end(optimizer)

Update the progress bar at the end of each step.

Args: optimizer: The optimizer object that called the callback.

Source code in promptolution/utils/callbacks.py 
226
227
228
229
230
231
232
233
234
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Update the progress bar at the end of each step.

    Args:
    optimizer: The optimizer object that called the callback.
    """
    self.pbar.update(1)

    return True
on_train_end(optimizer)

Close the progress bar at the end of training.

Args: optimizer: The optimizer object that called the callback.

Source code in promptolution/utils/callbacks.py 
236
237
238
239
240
241
242
243
244
def on_train_end(self, optimizer: "BaseOptimizer") -> bool:
    """Close the progress bar at the end of training.

    Args:
    optimizer: The optimizer object that called the callback.
    """
    self.pbar.close()

    return True

TokenCountCallback

Bases: BaseCallback

Callback for stopping optimization based on the total token count.

Source code in promptolution/utils/callbacks.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class TokenCountCallback(BaseCallback):
    """Callback for stopping optimization based on the total token count."""

    def __init__(
        self,
        max_tokens_for_termination: int,
        token_type_for_termination: Literal["input_tokens", "output_tokens", "total_tokens"],
    ) -> None:
        """Initialize the TokenCountCallback.

        Args:
        max_tokens_for_termination (int): Maximum number of tokens which is allowed befor the algorithm is stopped.
        token_type_for_termination (str): Can be one of either "input_tokens", "output_tokens" or "total_tokens".
        """
        self.max_tokens_for_termination = max_tokens_for_termination
        self.token_type_for_termination = token_type_for_termination

    def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
        """Check if the total token count exceeds the maximum allowed. If so, stop the optimization."""
        token_counts = optimizer.predictor.llm.get_token_count()

        if token_counts[self.token_type_for_termination] > self.max_tokens_for_termination:
            return False

        return True
__init__(max_tokens_for_termination, token_type_for_termination)

Initialize the TokenCountCallback.

Args: max_tokens_for_termination (int): Maximum number of tokens which is allowed befor the algorithm is stopped. token_type_for_termination (str): Can be one of either "input_tokens", "output_tokens" or "total_tokens".

Source code in promptolution/utils/callbacks.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
def __init__(
    self,
    max_tokens_for_termination: int,
    token_type_for_termination: Literal["input_tokens", "output_tokens", "total_tokens"],
) -> None:
    """Initialize the TokenCountCallback.

    Args:
    max_tokens_for_termination (int): Maximum number of tokens which is allowed befor the algorithm is stopped.
    token_type_for_termination (str): Can be one of either "input_tokens", "output_tokens" or "total_tokens".
    """
    self.max_tokens_for_termination = max_tokens_for_termination
    self.token_type_for_termination = token_type_for_termination
on_step_end(optimizer)

Check if the total token count exceeds the maximum allowed. If so, stop the optimization.

Source code in promptolution/utils/callbacks.py 
264
265
266
267
268
269
270
271
def on_step_end(self, optimizer: "BaseOptimizer") -> bool:
    """Check if the total token count exceeds the maximum allowed. If so, stop the optimization."""
    token_counts = optimizer.predictor.llm.get_token_count()

    if token_counts[self.token_type_for_termination] > self.max_tokens_for_termination:
        return False

    return True

capo_utils

Shared utilities for CAPO-style optimizers.

build_few_shot_examples(instruction, num_examples, optimizer)

Create few-shot examples with optional reasoning replacement.

Source code in promptolution/utils/capo_utils.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def build_few_shot_examples(
    instruction: str,
    num_examples: int,
    optimizer,
) -> List[str]:
    """Create few-shot examples with optional reasoning replacement."""
    if num_examples == 0:
        return []

    few_shot_samples = optimizer.df_few_shots.sample(num_examples, replace=False)
    sample_inputs = few_shot_samples[optimizer.task.x_column].values.astype(str)
    sample_targets = few_shot_samples[optimizer.task.y_column].values
    few_shots = [
        CAPO_FEWSHOT_TEMPLATE.replace("<input>", i).replace(
            "<output>", f"{optimizer.target_begin_marker}{t}{optimizer.target_end_marker}"
        )
        for i, t in zip(sample_inputs, sample_targets)
    ]

    if not optimizer.create_fs_reasoning:
        return few_shots

    preds, seqs = optimizer.predictor.predict(
        [instruction] * num_examples,
        list(sample_inputs),
    )
    if isinstance(seqs, str):
        seqs = [seqs]
    if isinstance(preds, str):
        preds = [preds]

    for j in range(num_examples):
        seqs[j] = seqs[j].replace(sample_inputs[j], "", 1).strip()
        if preds[j] == sample_targets[j] or not optimizer.check_fs_accuracy:
            few_shots[j] = CAPO_FEWSHOT_TEMPLATE.replace("<input>", sample_inputs[j]).replace("<output>", seqs[j])

    return few_shots

perform_crossover(parents, optimizer, parent_select_func=None)

Generate crossover offspring prompts.

Source code in promptolution/utils/capo_utils.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def perform_crossover(
    parents: List[Prompt],
    optimizer,
    parent_select_func: Optional[Callable] = None,
) -> List[Prompt]:
    """Generate crossover offspring prompts."""
    crossover_prompts: List[str] = []
    offspring_few_shots: List[List[str]] = []
    for _ in range(optimizer.crossovers_per_iter):
        if parent_select_func:
            mother, father = parent_select_func(parents)
        else:
            mother, father = random.sample(parents, 2)
        crossover_prompt = (
            optimizer.crossover_template.replace("<mother>", mother.instruction)
            .replace("<father>", father.instruction)
            .strip()
        )
        crossover_prompts.append(crossover_prompt)
        combined_few_shots = mother.few_shots + father.few_shots
        num_few_shots = (len(mother.few_shots) + len(father.few_shots)) // 2
        offspring_few_shot = random.sample(combined_few_shots, num_few_shots) if combined_few_shots else []
        offspring_few_shots.append(offspring_few_shot)

    child_instructions = optimizer.meta_llm.get_response(crossover_prompts)
    return [
        Prompt(extract_from_tag(instr, "<prompt>", "</prompt>"), examples)
        for instr, examples in zip(child_instructions, offspring_few_shots)
    ]

perform_mutation(offsprings, optimizer)

Mutate offspring prompts.

Source code in promptolution/utils/capo_utils.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def perform_mutation(
    offsprings: List[Prompt],
    optimizer,
) -> List[Prompt]:
    """Mutate offspring prompts."""
    mutation_template = optimizer.mutation_template
    meta_llm = optimizer.meta_llm
    upper_shots = optimizer.upper_shots

    mutation_prompts = [mutation_template.replace("<instruction>", prompt.instruction) for prompt in offsprings]
    new_instructions = meta_llm.get_response(mutation_prompts)

    mutated: List[Prompt] = []
    for new_instruction, prompt in zip(new_instructions, offsprings):
        new_instruction = extract_from_tag(new_instruction, "<prompt>", "</prompt>")
        p = random.random()

        if p < 1 / 3 and len(prompt.few_shots) < upper_shots:
            new_few_shot = build_few_shot_examples(
                instruction=new_instruction,
                num_examples=1,
                optimizer=optimizer,
            )
            new_few_shots = prompt.few_shots + new_few_shot
        elif 1 / 3 <= p < 2 / 3 and len(prompt.few_shots) > 0:
            new_few_shots = random.sample(prompt.few_shots, len(prompt.few_shots) - 1)
        else:
            new_few_shots = prompt.few_shots

        random.shuffle(new_few_shots)
        mutated.append(Prompt(new_instruction, new_few_shots))

    return mutated

evaluation

Component-level evaluation helpers: split off a held-out set, score prompts on a task.

dev_test_split(df, test_frac=0.2, seed=42)

Split a DataFrame into a dev and test set.

The optimizer selects prompts on the dev set. The test set is held out for the final evaluation.

Parameters:

Name Type Description Default
df DataFrame

The data to split (or a df-like with .to_pandas()).

 required
test_frac float

Fraction of rows held out for the test split.

 0.2
seed int

Random seed for the split.

 42

Returns:

Type Description
Tuple[DataFrame, DataFrame]

Tuple[pd.DataFrame, pd.DataFrame]: (dev_df, test_df), both with a fresh index.
Source code in promptolution/utils/evaluation.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def dev_test_split(df: pd.DataFrame, test_frac: float = 0.2, seed: int = 42) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """Split a DataFrame into a dev and test set.

    The optimizer selects prompts on the dev set. The test set is held out for the final evaluation.

    Args:
        df (pd.DataFrame): The data to split (or a df-like with ``.to_pandas()``).
        test_frac (float): Fraction of rows held out for the test split.
        seed (int): Random seed for the split.

    Returns:
        Tuple[pd.DataFrame, pd.DataFrame]: ``(dev_df, test_df)``, both with a fresh index.
    """
    if not isinstance(df, pd.DataFrame):
        df = df.to_pandas()
    test_df = df.sample(frac=test_frac, random_state=seed)
    dev_df = df.drop(test_df.index)
    return dev_df.reset_index(drop=True), test_df.reset_index(drop=True)

evaluate_prompts(prompts, task, predictor)

Score prompts on a task and return a sorted prompt/score table.

Parameters:

Name Type Description Default
prompts Union[List[Prompt], List[str]]

Prompts to score.

 required
task BaseTask

The task to score on (its full dataset is used).

 required
predictor BasePredictor

The predictor that runs the prompts.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: Columns prompt and score, best first.
Source code in promptolution/utils/evaluation.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def evaluate_prompts(
    prompts: Union[List[Prompt], List[str]], task: "BaseTask", predictor: "BasePredictor"
) -> pd.DataFrame:
    """Score prompts on a task and return a sorted prompt/score table.

    Args:
        prompts (Union[List[Prompt], List[str]]): Prompts to score.
        task (BaseTask): The task to score on (its full dataset is used).
        predictor (BasePredictor): The predictor that runs the prompts.

    Returns:
        pd.DataFrame: Columns ``prompt`` and ``score``, best first.
    """
    if isinstance(prompts[0], str):
        str_prompts = cast(List[str], list(prompts))
        prompt_objs = [Prompt(p) for p in str_prompts]
    else:
        prompt_objs = cast(List[Prompt], list(prompts))
        str_prompts = [p.construct_prompt() for p in prompt_objs]
    logger.warning("📊 Starting evaluation...")
    results = task.evaluate(prompt_objs, predictor, eval_strategy="full")
    return pd.DataFrame({"prompt": str_prompts, "score": results.agg_scores.tolist()}).sort_values(
        "score", ascending=False, ignore_index=True
    )

formatting

Utils for formatting prompts and outputs.

extract_from_tag(text, start_tag, end_tag) 

extract_from_tag(
    text: str, start_tag: str, end_tag: str
) -> str

extract_from_tag(
    text: List[str], start_tag: str, end_tag: str
) -> List[str]

Extract content from a string between specified start and end tags.

Parameters:

Name Type Description Default
text str

The input text to extract from.

 required
start_tag str

The start tag to look for.

 required
end_tag str

The end tag to look for.

 required

Returns:

Type Description
Union[List[str], str]

Union[List[str], str]: The extracted content, either as a list or a single string.
Source code in promptolution/utils/formatting.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def extract_from_tag(text: Union[str, List[str]], start_tag: str, end_tag: str) -> Union[List[str], str]:
    """Extract content from a string between specified start and end tags.

    Args:
        text (str): The input text to extract from.
        start_tag (str): The start tag to look for.
        end_tag (str): The end tag to look for.

    Returns:
        Union[List[str], str]: The extracted content, either as a list or a single string.
    """
    was_list = True
    if isinstance(text, str):
        text = [text]
        was_list = False

    outs = []
    for t in text:
        out = t.split(start_tag)[-1].split(end_tag)[0].strip()
        outs.append(out)
    if was_list:
        return outs
    return outs[0]

logging

Logging configuration for the promptolution library.

get_logger(name, level=None)

Get a logger with the specified name and level.

This function provides a standardized way to get loggers throughout the library, ensuring consistent formatting and behavior.

Parameters:

Name Type Description Default
name str

Name of the logger, typically name of the module.

 required
level int

Logging level. Defaults to None, which uses the root logger's level.

 None

Returns:

Type Description
Logger

logging.Logger: Configured logger instance.
Source code in promptolution/utils/logging.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def get_logger(name: str, level: Optional[int] = None) -> logging.Logger:
    """Get a logger with the specified name and level.

    This function provides a standardized way to get loggers throughout the library,
    ensuring consistent formatting and behavior.

    Args:
        name (str): Name of the logger, typically __name__ of the module.
        level (int, optional): Logging level. Defaults to None, which uses the root logger's level.

    Returns:
        logging.Logger: Configured logger instance.
    """
    logger = logging.getLogger(name)
    if level is not None:
        logger.setLevel(level)
    return logger

setup_logging(level=logging.INFO)

Set up logging for the promptolution library.

This function configures the root logger for the library with appropriate formatting and level.

Parameters:

Name Type Description Default
level int

Logging level. Defaults to logging.INFO.

 INFO
Source code in promptolution/utils/logging.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def setup_logging(level: int = logging.INFO) -> None:
    """Set up logging for the promptolution library.

    This function configures the root logger for the library with appropriate
    formatting and level.

    Args:
        level (int, optional): Logging level. Defaults to logging.INFO.
    """
    # Configure the root logger
    logging.basicConfig(
        level=level,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

optimize

The lightweight entry points: optimize prompts on a dataset, and score them on another.

evaluate(prompts, data, task_description, task_type='classification', *, model_id='gpt-4o-mini', api_key=None, api_url=None, x_column='x', y_column='y')

Score prompts on a dataset and return a sorted prompt/score table.

The counterpart to :func:optimize: pass the prompts it returned and your held-out split. Evaluation runs on all of data.

Parameters:

Name Type Description Default
prompts Union[List[Prompt], List[str]]

Prompts to score, e.g. the output of :func:optimize.

 required
data Union[DataFrame, str, Path]

A DataFrame, or anything pandas.read_csv accepts.

 required
task_description str

Description of the task.

 required
task_type str

Only "classification" is supported; anything else raises.

 'classification'
model_id str

Model id of the LLM used for predictions.

 'gpt-4o-mini'
api_key Optional[str]

API key. If omitted, resolved from the environment.

 None
api_url Optional[str]

API base URL, for providers other than OpenAI.

 None
x_column str

Input column name in data.

 'x'
y_column str

Label column name in data.

 'y'

Returns:

Type Description
DataFrame

pd.DataFrame: Columns prompt and score, best first.
Source code in promptolution/utils/optimize.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def evaluate(
    prompts: Union[List[Prompt], List[str]],
    data: Union[pd.DataFrame, str, Path],
    task_description: str,
    task_type: str = "classification",
    *,
    model_id: str = "gpt-4o-mini",
    api_key: Optional[str] = None,
    api_url: Optional[str] = None,
    x_column: str = "x",
    y_column: str = "y",
) -> pd.DataFrame:
    """Score prompts on a dataset and return a sorted prompt/score table.

    The counterpart to :func:`optimize`: pass the prompts it returned and your held-out split.
    Evaluation runs on all of ``data``.

    Args:
        prompts (Union[List[Prompt], List[str]]): Prompts to score, e.g. the output of :func:`optimize`.
        data (Union[pd.DataFrame, str, Path]): A DataFrame, or anything ``pandas.read_csv`` accepts.
        task_description (str): Description of the task.
        task_type (str): Only ``"classification"`` is supported; anything else raises.
        model_id (str): Model id of the LLM used for predictions.
        api_key (Optional[str]): API key. If omitted, resolved from the environment.
        api_url (Optional[str]): API base URL, for providers other than OpenAI.
        x_column (str): Input column name in ``data``.
        y_column (str): Label column name in ``data``.

    Returns:
        pd.DataFrame: Columns ``prompt`` and ``score``, best first.
    """
    _check_task_type(task_type, "evaluate")
    _, predictor, task = _build(data, task_description, model_id, api_key, api_url, x_column, y_column, "full")
    return evaluate_prompts(prompts, task, predictor)

optimize(data, task_description, task_type='classification', *, model_id='gpt-4o-mini', api_key=None, api_url=None, x_column='x', y_column='y', optimizer='capo', n_steps=10, initial_prompts=None, output_dir=None)

Optimize prompts on a dataset, with sensible defaults.

Builds the components for you (an API LLM, a marker-based predictor, a classification task and the chosen optimizer), asking only for what the library cannot know: your data, an LLM and its credentials, and a description of the task.

All of data is used to select prompts, so pass your dev split here and keep a held-out split for :func:evaluate. Nothing is fitted on this data in the gradient sense; the optimizer only uses it to score and select candidate prompts.

Only classification is supported. For judge or reward tasks, or for full control over the components, build them yourself and call optimizer.optimize() (see the README). For config-driven runs, grids, SLURM and restart, use the promptolution-experiment CLI.

Parameters:

Name Type Description Default
data Union[DataFrame, str, Path]

A DataFrame, or anything pandas.read_csv accepts.

 required
task_description str

Description of the task, used to generate initial prompts and passed to the meta-LLM during optimization.

 required
task_type str

Only "classification" is supported; anything else raises.

 'classification'
model_id str

Model id of the LLM used for both predictions and prompt proposals.

 'gpt-4o-mini'
api_key Optional[str]

API key. If omitted, the underlying OpenAI client resolves it from the environment (e.g. OPENAI_API_KEY).

 None
api_url Optional[str]

API base URL, for providers other than OpenAI.

 None
x_column str

Input column name in data.

 'x'
y_column str

Label column name in data.

 'y'
optimizer str

One of capo, evopromptde, evopromptga, opro.

 'capo'
n_steps int

Number of optimization steps.

 10
initial_prompts Optional[List[str]]

Prompts to start from. If omitted, they are generated from task_description.

 None
output_dir Optional[Union[str, Path]]

If set, write the per-step trace and run info here. If omitted (the default), nothing is written to disk.

 None

Returns:

Type Description
List[Prompt]

List[Prompt]: The optimized prompts. Pass them to :func:evaluate to score them.
Source code in promptolution/utils/optimize.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def optimize(
    data: Union[pd.DataFrame, str, Path],
    task_description: str,
    task_type: str = "classification",
    *,
    model_id: str = "gpt-4o-mini",
    api_key: Optional[str] = None,
    api_url: Optional[str] = None,
    x_column: str = "x",
    y_column: str = "y",
    optimizer: str = "capo",
    n_steps: int = 10,
    initial_prompts: Optional[List[str]] = None,
    output_dir: Optional[Union[str, Path]] = None,
) -> List[Prompt]:
    """Optimize prompts on a dataset, with sensible defaults.

    Builds the components for you (an API LLM, a marker-based predictor, a classification task and
    the chosen optimizer), asking only for what the library cannot know: your data, an LLM and its
    credentials, and a description of the task.

    All of ``data`` is used to select prompts, so pass your dev split here and keep a held-out split
    for :func:`evaluate`. Nothing is fitted on this data in the gradient sense; the optimizer only
    uses it to score and select candidate prompts.

    Only classification is supported. For judge or reward tasks, or for full control over the
    components, build them yourself and call ``optimizer.optimize()`` (see the README). For
    config-driven runs, grids, SLURM and restart, use the ``promptolution-experiment`` CLI.

    Args:
        data (Union[pd.DataFrame, str, Path]): A DataFrame, or anything ``pandas.read_csv`` accepts.
        task_description (str): Description of the task, used to generate initial prompts and passed
            to the meta-LLM during optimization.
        task_type (str): Only ``"classification"`` is supported; anything else raises.
        model_id (str): Model id of the LLM used for both predictions and prompt proposals.
        api_key (Optional[str]): API key. If omitted, the underlying OpenAI client resolves it from
            the environment (e.g. ``OPENAI_API_KEY``).
        api_url (Optional[str]): API base URL, for providers other than OpenAI.
        x_column (str): Input column name in ``data``.
        y_column (str): Label column name in ``data``.
        optimizer (str): One of ``capo``, ``evopromptde``, ``evopromptga``, ``opro``.
        n_steps (int): Number of optimization steps.
        initial_prompts (Optional[List[str]]): Prompts to start from. If omitted, they are generated
            from ``task_description``.
        output_dir (Optional[Union[str, Path]]): If set, write the per-step trace and run info here.
            If omitted (the default), nothing is written to disk.

    Returns:
        List[Prompt]: The optimized prompts. Pass them to :func:`evaluate` to score them.
    """
    _check_task_type(task_type, "optimize")
    if optimizer not in OPTIMIZERS:
        raise ValueError(f"Unknown optimizer {optimizer!r}. Available: {', '.join(sorted(OPTIMIZERS))}.")

    # CAPO races prompts on data blocks, so the task needs a block eval_strategy
    llm, predictor, task = _build(
        data, task_description, model_id, api_key, api_url, x_column, y_column, "sequential_block"
    )

    # meta_llm is accepted by every concrete optimizer but not declared on BaseOptimizer, hence the ignore
    opt = OPTIMIZERS[optimizer](
        predictor=predictor, meta_llm=llm, task=task, initial_prompts=initial_prompts  # type: ignore[call-arg]
    )

    info = None
    if output_dir is not None:
        output_dir = Path(output_dir)
        output_dir.mkdir(parents=True, exist_ok=True)
        opt.callbacks.append(FileOutputCallback(dir=output_dir))
        info = start_runinfo(output_dir, output_dir.name)

    try:
        logger.warning("🔥 Starting optimization...")
        prompts = opt.optimize(n_steps=n_steps)
    except Exception as e:
        if info is not None and output_dir is not None:
            finish_runinfo(output_dir, info, status="failed", error=str(e))
        raise

    if info is not None and output_dir is not None:
        finish_runinfo(output_dir, info, status="finished")
        logger.warning("✅ Finished optimization: %s", output_dir)
    return prompts

prompt

Module defining the Prompt class and related utilities.

Prompt

Represent a prompt consisting of an instruction and few-shot examples.

Source code in promptolution/utils/prompt.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class Prompt:
    """Represent a prompt consisting of an instruction and few-shot examples."""

    def __init__(
        self, instruction: str, few_shots: Optional[List[str]] = None, downstream_template: Optional[str] = None
    ) -> None:
        """Initialize the Prompt with an instruction and associated examples.

        Args:
            instruction (str): The instruction or prompt text.
            few_shots (List[str]): List of examples as string.
            downstream_template (str, optional): Template for formatting the full prompt.
        """
        self.instruction = instruction.strip()
        self.few_shots = few_shots or []
        if downstream_template is None:
            if self.few_shots:
                downstream_template = DOWNSTREAM_TEMPLATE_W_FEWSHOTS
            else:
                downstream_template = DOWNSTREAM_TEMPLATE
        self.downstream_template = downstream_template

    def construct_prompt(self) -> str:
        """Construct the full prompt string by replacing placeholders in the template with the instruction and formatted examples.

        Returns:
            str: The constructed prompt string.
        """
        few_shot_str = "\n\n".join(self.few_shots).strip()
        prompt = (
            self.downstream_template.replace("<instruction>", self.instruction)
            .replace("<few_shots>", few_shot_str)
            .replace("\n\n\n\n", "\n\n")  # replace extra newlines if no few shots are provided
            .strip()
        )
        return prompt

    def __str__(self) -> str:
        """Return the string representation of the prompt."""
        return self.construct_prompt()

    def __eq__(self, other: object) -> bool:
        """Structural equality for use in lists, sets, and dict keys."""
        if not isinstance(other, Prompt):
            return False
        return (
            self.instruction == other.instruction
            and self.few_shots == other.few_shots
            and self.downstream_template == other.downstream_template
        )

    def __hash__(self) -> int:
        """Hash function for use in sets and dict keys."""
        return hash((self.instruction, tuple(self.few_shots), self.downstream_template))
__eq__(other)

Structural equality for use in lists, sets, and dict keys.

Source code in promptolution/utils/prompt.py 
51
52
53
54
55
56
57
58
59
def __eq__(self, other: object) -> bool:
    """Structural equality for use in lists, sets, and dict keys."""
    if not isinstance(other, Prompt):
        return False
    return (
        self.instruction == other.instruction
        and self.few_shots == other.few_shots
        and self.downstream_template == other.downstream_template
    )
__hash__()

Hash function for use in sets and dict keys.

Source code in promptolution/utils/prompt.py 
61
62
63
def __hash__(self) -> int:
    """Hash function for use in sets and dict keys."""
    return hash((self.instruction, tuple(self.few_shots), self.downstream_template))
__init__(instruction, few_shots=None, downstream_template=None)

Initialize the Prompt with an instruction and associated examples.

Parameters:

Name Type Description Default
instruction str

The instruction or prompt text.

 required
few_shots List[str]

List of examples as string.

 None
downstream_template str

Template for formatting the full prompt.

 None
Source code in promptolution/utils/prompt.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self, instruction: str, few_shots: Optional[List[str]] = None, downstream_template: Optional[str] = None
) -> None:
    """Initialize the Prompt with an instruction and associated examples.

    Args:
        instruction (str): The instruction or prompt text.
        few_shots (List[str]): List of examples as string.
        downstream_template (str, optional): Template for formatting the full prompt.
    """
    self.instruction = instruction.strip()
    self.few_shots = few_shots or []
    if downstream_template is None:
        if self.few_shots:
            downstream_template = DOWNSTREAM_TEMPLATE_W_FEWSHOTS
        else:
            downstream_template = DOWNSTREAM_TEMPLATE
    self.downstream_template = downstream_template
__str__()

Return the string representation of the prompt.

Source code in promptolution/utils/prompt.py 
47
48
49
def __str__(self) -> str:
    """Return the string representation of the prompt."""
    return self.construct_prompt()
construct_prompt()

Construct the full prompt string by replacing placeholders in the template with the instruction and formatted examples.

Returns:

Name Type Description
str str

The constructed prompt string.
Source code in promptolution/utils/prompt.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def construct_prompt(self) -> str:
    """Construct the full prompt string by replacing placeholders in the template with the instruction and formatted examples.

    Returns:
        str: The constructed prompt string.
    """
    few_shot_str = "\n\n".join(self.few_shots).strip()
    prompt = (
        self.downstream_template.replace("<instruction>", self.instruction)
        .replace("<few_shots>", few_shot_str)
        .replace("\n\n\n\n", "\n\n")  # replace extra newlines if no few shots are provided
        .strip()
    )
    return prompt

sort_prompts_by_scores(prompts, scores, top_k=None)

Sort prompts by score, accepting scalar, 1D, or multi-dimensional scores.

Scores can be provided as Python lists or NumPy arrays. If scores are multi- dimensional (e.g., per-subsample results), they are aggregated with a nanmean across all non-leading axes before sorting.

Parameters:

Name Type Description Default
prompts List[Prompt]

Prompt objects to sort.

 required
scores Sequence[float] | ndarray

Corresponding scores; can be nested lists or arrays.

 required
top_k Optional[int]

Limit the result to the top_k prompts.

 None

Returns:

Type Description
List[Prompt]

Tuple[List[Prompt], List[float]]: Prompts and their aggregated scores,
List[float]

sorted in descending order.
Source code in promptolution/utils/prompt.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def sort_prompts_by_scores(
    prompts: List[Prompt], scores: Union[Sequence[float], np.ndarray], top_k: Optional[int] = None
) -> Tuple[List[Prompt], List[float]]:
    """Sort prompts by score, accepting scalar, 1D, or multi-dimensional scores.

    Scores can be provided as Python lists or NumPy arrays. If scores are multi-
    dimensional (e.g., per-subsample results), they are aggregated with a
    ``nanmean`` across all non-leading axes before sorting.

    Args:
        prompts (List[Prompt]): Prompt objects to sort.
        scores (Sequence[float] | np.ndarray): Corresponding scores; can be nested lists or arrays.
        top_k (Optional[int]): Limit the result to the top_k prompts.

    Returns:
        Tuple[List[Prompt], List[float]]: Prompts and their aggregated scores,
        sorted in descending order.
    """
    scores_arr = np.asarray(scores, dtype=float)
    if scores_arr.ndim == 0:
        scores_arr = scores_arr.reshape(1)

    assert scores_arr.shape[0] == len(prompts), "Prompts and scores must have the same length."

    if scores_arr.ndim > 1:
        axes_to_reduce = tuple(range(1, scores_arr.ndim))
        scores_arr = np.nanmean(scores_arr, axis=axes_to_reduce)

    prompt_score_pairs = list(zip(prompts, scores_arr.tolist()))
    prompt_score_pairs.sort(key=lambda pair: pair[1], reverse=True)

    if top_k is not None:
        prompt_score_pairs = prompt_score_pairs[:top_k]

    sorted_prompts = [p for p, _ in prompt_score_pairs]
    sorted_scores = [s for _, s in prompt_score_pairs]

    return sorted_prompts, sorted_scores

prompt_creation

Utility functions for prompt creation.

create_prompt_variation(prompt, llm, meta_prompt=None)

Generate a variation of the given prompt(s) while keeping the semantic meaning.

Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910

Parameters:

Name Type Description Default
prompt Union[List[str], str]

The prompt(s) to generate variations of.

 required
llm BaseLLM

The language model to use for generating the variations.

 required
meta_prompt str

The meta prompt to use for generating the variations. If None, a default meta prompt is used. Should contain tag.

 None

Returns:

Type Description
List[str]

List[str]: A list of generated variations of the input prompt(s).
Source code in promptolution/utils/prompt_creation.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def create_prompt_variation(
    prompt: Union[List[str], str], llm: "BaseLLM", meta_prompt: Optional[str] = None
) -> List[str]:
    """Generate a variation of the given prompt(s) while keeping the semantic meaning.

    Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910

    Args:
        prompt (Union[List[str], str]): The prompt(s) to generate variations of.
        llm (BaseLLM): The language model to use for generating the variations.
        meta_prompt (str): The meta prompt to use for generating the variations.
            If None, a default meta prompt is used. Should contain <prev_prompt> tag.

    Returns:
        List[str]: A list of generated variations of the input prompt(s).
    """
    meta_prompt = PROMPT_VARIATION_TEMPLATE if meta_prompt is None else meta_prompt

    if isinstance(prompt, str):
        prompt = [prompt]
    varied_prompts = llm.get_response([meta_prompt.replace("<prev_prompt>", p) for p in prompt])
    varied_prompts = extract_from_tag(varied_prompts, "<prompt>", "</prompt>")

    return varied_prompts

create_prompts_from_samples(task, llm, meta_prompt=None, n_samples=3, task_description=None, n_prompts=1, get_uniform_labels=False)

Generate a set of prompts from dataset examples sampled from a given task.

Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910 Samples are selected, such that (1) all possible classes are represented (2) the samples are as representative as possible

Parameters:

Name Type Description Default
task BaseTask

The task to generate prompts for. Xs and Ys from this object are used to generate the prompts.

 required
llm BaseLLM

The language model to use for generating the prompts.

 required
meta_prompt str

The meta prompt to use for generating the prompts. If None, a default meta prompt is used.

 None
n_samples int

The number of samples to use for generating prompts.

 3
task_description str

The description of the task to include in the prompt.

 None
n_prompts int

The number of prompts to generate.

 1
get_uniform_labels bool

If True, samples are selected such that all classes are represented.

 False

Returns:

Type Description
List[str]

List[str]: A list of generated prompts.
Source code in promptolution/utils/prompt_creation.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def create_prompts_from_samples(
    task: "BaseTask",
    llm: "BaseLLM",
    meta_prompt: Optional[str] = None,
    n_samples: int = 3,
    task_description: Optional[str] = None,
    n_prompts: int = 1,
    get_uniform_labels: bool = False,
) -> List[str]:
    """Generate a set of prompts from dataset examples sampled from a given task.

    Idea taken from the paper Zhou et al. (2021) https://arxiv.org/pdf/2211.01910
    Samples are selected, such that
    (1) all possible classes are represented
    (2) the samples are as representative as possible

    Args:
        task (BaseTask): The task to generate prompts for.
            Xs and Ys from this object are used to generate the prompts.
        llm (BaseLLM): The language model to use for generating the prompts.
        meta_prompt (str): The meta prompt to use for generating the prompts.
            If None, a default meta prompt is used.
        n_samples (int): The number of samples to use for generating prompts.
        task_description (str): The description of the task to include in the prompt.
        n_prompts (int): The number of prompts to generate.
        get_uniform_labels (bool): If True, samples are selected such that all classes are represented.

    Returns:
        List[str]: A list of generated prompts.
    """
    meta_prompt_template: str
    if meta_prompt is None:
        if task_description is None:
            meta_prompt_template = PROMPT_CREATION_TEMPLATE
        else:
            meta_prompt_template = PROMPT_CREATION_TEMPLATE_TD.replace("<task_desc>", task_description)
    else:
        if task_description is None:
            meta_prompt_template = meta_prompt
        else:
            meta_prompt_template = meta_prompt.replace("<task_desc>", task_description)

    meta_prompts = []
    for _ in range(n_prompts):
        if task.task_type == "classification" and get_uniform_labels:
            # if classification task sample such that all classes are represented
            unique_labels, counts = np.unique(task.ys, return_counts=True)
            proportions = counts / len(task.ys)
            samples_per_class = np.round(proportions * n_samples).astype(int)
            samples_per_class = np.maximum(samples_per_class, 1)

            # sample
            xs: List[str] = []
            ys: List[str] = []
            for label, n_per_class in zip(unique_labels, samples_per_class):
                indices = np.where(task.ys == label)[0]
                indices = np.random.choice(indices, n_per_class, replace=False)
                xs.extend(task.xs[indices])
                ys.extend(task.ys[indices])

        else:
            # if not classification task, sample randomly
            indices = np.random.choice(len(task.xs), n_samples, replace=False)
            xs = [task.xs[i] for i in indices]
            ys = [task.ys[i] for i in indices]

        examples = "\n\n".join([f"Input: {x}\nOutput: {y}" for x, y in zip(xs, ys)])
        meta_prompt = meta_prompt_template.replace("<input_output_pairs>", examples)
        meta_prompts.append(meta_prompt)

    prompts = llm.get_response(meta_prompts)
    prompts = extract_from_tag(prompts, "<prompt>", "</prompt>")

    return prompts

create_prompts_from_task_description(task_description, llm, meta_prompt=None, n_prompts=10, n_retries=3)

Generate a set of prompts from a given task description.

Parameters:

Name Type Description Default
task_description str

The description of the task to generate prompts for.

 required
llm BaseLLM

The language model to use for generating the prompts.

 required
meta_prompt str

The meta prompt to use for generating the prompts. If None, a default meta prompt is used.

 None
n_prompts int

The number of prompts to generate.

 10
n_retries int

The number of retries to attempt if prompt generation fails.

 3
Source code in promptolution/utils/prompt_creation.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def create_prompts_from_task_description(
    task_description: str,
    llm: "BaseLLM",
    meta_prompt: Optional[str] = None,
    n_prompts: int = 10,
    n_retries: int = 3,
) -> List[str]:
    """Generate a set of prompts from a given task description.

    Args:
        task_description (str): The description of the task to generate prompts for.
        llm (BaseLLM): The language model to use for generating the prompts.
        meta_prompt (str): The meta prompt to use for generating the prompts.
            If None, a default meta prompt is used.
        n_prompts (int): The number of prompts to generate.
        n_retries (int): The number of retries to attempt if prompt generation fails.
    """
    if meta_prompt is None:
        meta_prompt = PROMPT_CREATION_TEMPLATE_FROM_TASK_DESCRIPTION

    meta_prompt = meta_prompt.replace("<task_desc>", task_description).replace("<n_prompts>", str(n_prompts))
    final_prompts = None
    for _ in range(n_retries):
        prompts_str = llm.get_response(meta_prompt)[0]
        try:
            prompts = json.loads(prompts_str)
            assert isinstance(prompts, list) and all(isinstance(p, str) for p in prompts) and len(prompts) == n_prompts
            final_prompts = prompts
            break
        except (json.JSONDecodeError, AssertionError):
            logger.warning("Failed to parse prompts JSON, retrying...")

    if final_prompts is None:
        logger.error(
            f"Failed to generate prompts from task description after {n_retries} retries, returning default prompts."
        )
        final_prompts = default_prompts[:n_prompts]

    return final_prompts

runinfo

Persist a run's status and timestamps to runinfo.json (part of the experiment output contract).

finish_runinfo(out_dir, info, status, **extra)

Update info with a finished/failed status and the current UTC timestamp, and rewrite runinfo.json.

Parameters:

Name Type Description Default
out_dir Union[str, Path]

Directory runinfo.json is written to.

 required
info Dict[str, Any]

The dict returned by :func:start_runinfo.

 required
status str

The run's final status, e.g. "finished" or "failed".

 required
**extra Any

Extra fields to record, e.g. error=str(e).

 {}

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The updated info dict.
Source code in promptolution/utils/runinfo.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def finish_runinfo(out_dir: Union[str, Path], info: Dict[str, Any], status: str, **extra: Any) -> Dict[str, Any]:
    """Update info with a finished/failed status and the current UTC timestamp, and rewrite runinfo.json.

    Args:
        out_dir (Union[str, Path]): Directory runinfo.json is written to.
        info (Dict[str, Any]): The dict returned by :func:`start_runinfo`.
        status (str): The run's final status, e.g. "finished" or "failed".
        **extra (Any): Extra fields to record, e.g. ``error=str(e)``.

    Returns:
        Dict[str, Any]: The updated info dict.
    """
    info.update(status=status, finished_at=datetime.now(timezone.utc).isoformat(), **extra)
    write_to_file(out_dir, info)
    return info

start_runinfo(out_dir, name)

Write a "running" runinfo.json with the current UTC timestamp.

Parameters:

Name Type Description Default
out_dir Union[str, Path]

Directory runinfo.json is written to.

 required
name str

The run's name.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The info dict, to be passed to :func:finish_runinfo later.
Source code in promptolution/utils/runinfo.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
def start_runinfo(out_dir: Union[str, Path], name: str) -> Dict[str, Any]:
    """Write a "running" runinfo.json with the current UTC timestamp.

    Args:
        out_dir (Union[str, Path]): Directory runinfo.json is written to.
        name (str): The run's name.

    Returns:
        Dict[str, Any]: The info dict, to be passed to :func:`finish_runinfo` later.
    """
    info = {"name": name, "status": "running", "started_at": datetime.now(timezone.utc).isoformat()}
    write_to_file(out_dir, info)
    return info

write_to_file(out_dir, info)

Write the run info to runinfo.json in out_dir.

Parameters:

Name Type Description Default
out_dir Union[str, Path]

Directory runinfo.json is written to.

 required
info Dict[str, Any]

The run info to serialize.

 required
Source code in promptolution/utils/runinfo.py 
42
43
44
45
46
47
48
49
def write_to_file(out_dir: Union[str, Path], info: Dict[str, Any]) -> None:
    """Write the run info to runinfo.json in ``out_dir``.

    Args:
        out_dir (Union[str, Path]): Directory runinfo.json is written to.
        info (Dict[str, Any]): The run info to serialize.
    """
    (Path(out_dir) / "runinfo.json").write_text(json.dumps(info, indent=2))

templates

Meta-prompt templates for different prompt optimization methods.

test_statistics

Implementation of statistical significance tests used in the racing algorithm. Contains paired t-test functionality to compare prompt performance and determine statistical significance between candidates.

get_test_statistic_func(name)

Get the test statistic function based on the name provided.

Parameters:

Name Type Description Default
name str

Name of the test statistic function.

 required

Returns:

Name Type Description
callable Callable[..., bool]

The corresponding test statistic function.
Source code in promptolution/utils/test_statistics.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_test_statistic_func(name: TestStatistics) -> Callable[..., bool]:
    """
    Get the test statistic function based on the name provided.

    Args:
        name (str): Name of the test statistic function.

    Returns:
        callable: The corresponding test statistic function.
    """
    if name == "paired_t_test":
        return paired_t_test
    else:
        raise ValueError(f"Unknown test statistic function: {name}. Should be one of {TestStatistics.__args__}.")

paired_t_test(scores_a, scores_b, alpha=0.05)

Uses a paired t-test to test if candidate A's accuracy is significantly higher than candidate B's accuracy within a confidence interval of 1-lpha. Assumptions: - The samples are paired. - The differences between the pairs are normally distributed (-> n > 30).

Parameters:

Name Type Description Default
scores_a List[float]

Array of accuracy scores for candidate A.

 required
scores_b List[float]

Array of accuracy scores for candidate B.

 required
alpha float

Significance level (default 0.05 for 95% confidence).

 0.05

Returns:

Name Type Description
bool bool

True if candidate A is significantly better than candidate B, False otherwise.
Source code in promptolution/utils/test_statistics.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def paired_t_test(
    scores_a: List[float],
    scores_b: List[float],
    alpha: float = 0.05,
) -> bool:
    """
    Uses a paired t-test to test if candidate A's accuracy is significantly
    higher than candidate B's accuracy within a confidence interval of 1-\alpha.
    Assumptions:
    - The samples are paired.
    - The differences between the pairs are normally distributed (-> n > 30).

    Parameters:
        scores_a (List[float]): Array of accuracy scores for candidate A.
        scores_b (List[float]): Array of accuracy scores for candidate B.
        alpha (float): Significance level (default 0.05 for 95% confidence).

    Returns:
        bool: True if candidate A is significantly better than candidate B, False otherwise.
    """
    scores_a = np.array(scores_a)
    scores_b = np.array(scores_b)

    _, p_value = ttest_rel(scores_a, scores_b, alternative="greater")

    result = p_value < alpha

    return bool(result)

token_counter

Token counter for LLMs.

This module provides a function to count the number of tokens in a given text.

get_token_counter(llm)

Get a token counter function for the given LLM.

This function returns a callable that counts tokens based on the LLM's tokenizer or a simple split method if no tokenizer is available.

Parameters:

Name Type Description Default
llm BaseLLM

The language model object that may have a tokenizer.

 required

Returns:

Type Description
Callable[[str], float]

A callable that takes a text input and returns the token count.
Source code in promptolution/utils/token_counter.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def get_token_counter(llm: "BaseLLM") -> Callable[[str], float]:
    """Get a token counter function for the given LLM.

    This function returns a callable that counts tokens based on the LLM's tokenizer
    or a simple split method if no tokenizer is available.

    Args:
        llm: The language model object that may have a tokenizer.

    Returns:
        A callable that takes a text input and returns the token count.

    """
    if llm.tokenizer is not None:
        tokenizer: "PreTrainedTokenizer" = llm.tokenizer
        return lambda x: float(len(tokenizer.encode(x)))
    else:
        logger.warning("⚠️ The LLM does not have a tokenizer. Using simple token count.")
        return lambda x: float(len(x.split()))