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 ADOPRO(Optimizer):
"""
:no-index:
ADOPRO (Adaptive Optimization by PROmpting) optimizer for numerical optimization using LLMs.
ADOPRO is an enhanced version of OPRO that dynamically adjusts prompts based on
optimization progress. It monitors the optimization trajectory and adapts the
prompting strategy to escape local optima and improve convergence.
This optimizer is best suited for:
- Complex optimization landscapes with multiple local optima
- Problems where OPRO gets stuck or converges prematurely
- Adaptive optimization strategies
- Problems requiring dynamic exploration-exploitation balance
Example:
>>> def rastrigin(x):
... A = 10
... n = len(x)
... return A*n + sum(float(i)**2 - A*math.cos(2*math.pi*float(i)) for i in x)
>>>
>>> adopro = ADOPRO(
... problem_text="Minimize the Rastrigin function (highly multimodal)",
... obj_func=rastrigin,
... api_key="your-api-key"
... )
>>> result = adopro.minimize(
... init_samples=[["1", "1"], ["2", "2"], ["0", "0"]],
... init_scores=[...],
... num_steps=20
... )
Note:
This class inherits from `Optimizer` and uses adaptive prompting strategies
to improve upon the basic OPRO approach.
"""
[docs]
def __init__(self, problem_text=None, obj_func=None, llm_model=None, api_key=None):
"""
Initialize the ADOPRO optimizer.
Args:
problem_text (str, optional): Natural language description of the
optimization problem. Should include information about the complexity
and any known challenges (e.g., multimodality).
obj_func (callable, optional): Objective function that takes a solution
and returns a numerical score.
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 ADOPRO 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: 50).
- batch_size (int): The number of new solutions to generate at each step (default: 5).
- temperature (float): The temperature for the LLM model (default: 1.0).
- callbacks (list): A list of callback functions to be triggered at the end of each step.
- optimization_type (str): "maximize" or "minimize" (default: "maximize").
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 ADOPRO 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
)