import numpy as np
from ..base import Optimizer, OptimizationResult
from ..llm.llm_init import initialize_llm
from ..utils.parsing import parse_pairs
from ..utils.truncate import truncate_pairs
from ..utils.logger import log_info, log_warning, log_error, log_critical, log_debug
from ..callbacks import EarlyStopping, OptimalScoreStopping, AdaptTempOnPlateau
[docs]
class OPRO(Optimizer):
"""
:no-index:
OPRO (Optimization by PROmpting) optimizer for numerical optimization using LLMs.
OPRO is the original approach that directly prompts LLMs to generate better solutions
based on previous examples. It works by showing the LLM a history of solutions and
their scores, then asking it to generate new solutions that improve upon the best ones.
This optimizer is best suited for:
- Simple optimization problems with clear patterns
- Problems where the relationship between solutions and scores is easily learnable
- Quick prototyping and testing
- Problems with relatively small search spaces
Example:
>>> def sphere(x):
... return sum(float(i)**2 for i in x)
>>>
>>> opro = OPRO(
... problem_text="Minimize the sphere function sum(x_i^2)",
... obj_func=sphere,
... api_key="your-api-key"
... )
>>> result = opro.minimize(
... init_samples=[["1", "1"], ["2", "2"]],
... init_scores=[2, 8],
... num_steps=10
... )
Note:
This class inherits from `Optimizer` and uses the default configuration
parameters unless overridden during initialization or optimization.
"""
[docs]
def __init__(self, problem_text=None, obj_func=None, llm_model=None, api_key=None):
"""
Initialize the OPRO optimizer.
Args:
problem_text (str, optional): Natural language description of the
optimization problem. This should clearly state what needs to be
optimized and any constraints.
obj_func (callable, optional): Objective function that takes a solution
and returns a numerical score. Higher scores indicate better solutions
for maximization, lower scores for minimization.
llm_model (str, optional): Name of the LLM model to use. If None,
uses the default from configuration file.
api_key (str, optional): API key for the LLM service. If None,
will attempt to read from environment variables.
"""
super().__init__(problem_text=problem_text, obj_func=obj_func, llm_model=llm_model, api_key=api_key)
[docs]
def optimize(self, init_samples=None, init_scores=None, num_steps=None, batch_size=None,
temperature=None, callbacks=None, verbose=1, optimization_type="maximize", parallel_n_jobs=None):
"""
Run the OPRO optimization algorithm.
Parameters:
- init_samples (list): A list of initial solutions.
- init_scores (list): A list of initial scores corresponding to init_samples.
- num_steps (int): The number of optimization steps (default from config).
- batch_size (int): The number of new solutions to generate at each step (default from config).
- temperature (float): The temperature for the LLM model (default from config).
- callbacks (list): A list of callback functions to be triggered at the end of each step.
- optimization_type (str): "maximize" or "minimize" (default: "maximize").
- parallel_n_jobs (int): Number of parallel jobs for evaluation (default from config).
Returns:
- results (OptimizationResult): An object containing the optimization results.
"""
from ..config import get_config
config = get_config()
# Use config defaults if not provided
if num_steps is None:
num_steps = config.default_num_steps
if batch_size is None:
batch_size = config.default_batch_size
if temperature is None:
temperature = config.temperature
if parallel_n_jobs is None:
parallel_n_jobs = config.parallel_n_jobs
client = initialize_llm(self.llm_model, self.api_key)
if verbose > 0:
log_info(f"Running OPRO optimization with {num_steps} steps and batch size {batch_size}...")
best_solution = None
if optimization_type == "maximize":
best_score = np.max(init_scores)
elif optimization_type == "minimize":
best_score = np.min(init_scores)
else:
log_critical("Invalid optimization_type. Choose 'maximize' or 'minimize'.")
raise ValueError("optimization_type must be 'maximize' or 'minimize'")
best_score_history = [best_score]
avg_score_per_step = [np.average(init_scores)]
best_score_per_step = [best_score]
#max_examples = batch_size
# Call the helper function to initialize callbacks
self._initialize_callbacks(callbacks, temperature)
for step in range(num_steps+1):
if step == 0:
if verbose > 0:
log_info(f"Step {step} - Best Initial Score: {best_score:.3f}, Average Initial Score: {np.average(init_scores):.3f}")
init_pairs = parse_pairs(init_samples, init_scores)
example_pairs = init_pairs
continue
if verbose > 1:
log_debug(f"Example pairs: {example_pairs}")
prompt = self.meta_prompt(batch_size, example_pairs, optimization_type)
solution_array = self._generate_solutions(client, prompt, temperature,
batch_size, verbose)
best_score, best_solution, step_scores, best_step_score = self._evaluate_solutions(solution_array, best_solution,
optimization_type, verbose, best_score, parallel_n_jobs)
new_pairs = parse_pairs(solution_array, step_scores)
example_pairs = example_pairs + new_pairs
#example_pairs = truncate_pairs(example_pairs, max_examples, optimization_type)
avg_step_score = sum(step_scores) / len(solution_array)
best_score_per_step.append(best_step_score)
avg_score_per_step.append(avg_step_score)
best_score_history.append(best_score)
if verbose > 0:
log_info(f"Step {step} - Current Best Score: {best_score:.3f}, Average Batch Score: {avg_step_score:.3f} - Best Batch Score: {best_step_score:.3f}")
# Callbacks: Trigger at the end of each step
if callbacks:
for callback in callbacks:
logs = {callback.monitor: best_score} # Pass logs with the monitored metric
new_temperature = callback.on_step_end(step, logs) # Callback could adjust the temperature
if new_temperature is not None:
temperature = new_temperature # Update temperature if needed
# Check if early stopping is triggered
early_stop = isinstance(callback, EarlyStopping) and callback.wait >= callback.patience
optimal_stop = isinstance(callback, OptimalScoreStopping) and callback.on_step_end(step, logs)
if early_stop or optimal_stop:
break
if early_stop or optimal_stop:
break
return OptimizationResult(
best_solution=best_solution,
best_score=best_score,
best_score_history=best_score_history,
best_score_per_step=best_score_per_step,
avg_score_per_step=avg_score_per_step
)