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 HLMSA(Optimizer):
"""
:no-index:
HLMSA (Hyper-heuristic LLM-driven Simulated Annealing) optimizer for numerical optimization.
HLMSA combines simulated annealing principles with LLM optimization to provide
controlled exploration of the solution space. It uses adaptive cooling rates and
perturbation strategies to balance exploration and exploitation.
This optimizer is best suited for:
- Problems with many local optima requiring careful exploration
- Fine-tuning solutions where small improvements matter
- Temperature-sensitive optimization problems
- Problems where controlled convergence is important
Example:
>>> def multimodal(x):
... # Function with many local optima
... return math.sin(5*x) + math.cos(3*x) + x**2
>>>
>>> hlmsa = HLMSA(
... problem_text="Find global minimum of multimodal function",
... obj_func=multimodal,
... api_key="your-api-key"
... )
>>> result = hlmsa.minimize(
... init_samples=[["0"], ["1"], ["-1"]],
... init_scores=[...],
... num_steps=30,
... batch_size=5
... )
Note:
HLMSA uses simulated annealing principles guided by LLM prompts to
dynamically adjust cooling rates and perturbation strategies.
"""
[docs]
def __init__(self, problem_text=None, obj_func=None, llm_model=None, api_key=None):
"""
Initialize the HLMSA optimizer.
Args:
problem_text (str, optional): Natural language description of the
optimization problem. For multimodal problems, mention the
presence of local optima.
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)
def _accept_solutions(self, prev_solutions, prev_scores, solution_array, step_scores, sa_temperature, optimization_type="maximize"):
current_solutions = []
current_scores = []
for i in range(len(solution_array)):
delta_e = step_scores[i] - prev_scores[i]
if optimization_type == "maximize":
accept_better = delta_e > 0
else:
accept_better = delta_e < 0
acceptance_probability = np.exp(-delta_e/sa_temperature)
if np.random.rand() < acceptance_probability or accept_better:
current_solutions.append(solution_array[i])
current_scores.append(step_scores[i])
else:
current_solutions.append(prev_solutions[i])
current_scores.append(prev_scores[i])
return current_solutions, current_scores
[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 HLMSA 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 HLMSA 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]
init_sa_temperature = 1000 #initial temperature
final_sa_temperature = 1.0 #final temperature
cooling_rate = 0.95 #initial cooling rate
# Call the helper function to initialize callbacks
self._initialize_callbacks(callbacks, temperature)
for step in range(num_steps+1):
if step == 0:
sa_temperature = init_sa_temperature
if verbose > 0:
log_info(f"Step {step} - SA Temperature: {sa_temperature:.2f} - 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
hp_text = "The solutions below are generated randomly."
prev_solutions, prev_scores = init_samples, init_scores
continue
if verbose > 1:
log_debug(f"Example pairs: {example_pairs}")
prompt = self.meta_prompt(batch_size, example_pairs, optimization_type, hp_text)
if verbose > 3:
log_debug(f"Prompt: {prompt}")
solution_array, hp = self._generate_solutions(client, prompt, temperature,
batch_size, verbose, hp_parse=True)
# If hyperparameter parsing failed but we got solutions, try again without hp_parse
if solution_array is not None and hp is None and len(solution_array) >= batch_size:
log_warning("Hyperparameter parsing failed, proceeding with default cooling rate")
hp = [0.95] # Default cooling rate
# Check if hp is not None and has the correct format
if hp is None or hp[0] < 0 or hp[0] > 1:
log_warning("Invalid or missing hyperparameters format.")
hp_text = "The cooling rate used in previous step are unknown."
else:
hp_text = f"""The hyperparameter (cooling rate) used in previous step is: <hp> {hp} <\\\\hp>"""
cooling_rate = hp[0]
best_score, best_solution, step_scores, best_step_score = self._evaluate_solutions(solution_array, best_solution,
optimization_type, verbose, best_score, parallel_n_jobs)
current_solutions, current_scores = self._accept_solutions(prev_solutions, prev_scores,
solution_array, step_scores, sa_temperature, optimization_type)
sa_temperature = sa_temperature * cooling_rate
new_pairs = parse_pairs(current_solutions, current_scores)
example_pairs = new_pairs
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} - SA Temperature: {sa_temperature:.2f} - Current Best Score: {best_score:.3f}, Average Batch Score: {avg_step_score:.3f} - Best Batch Score: {best_step_score:.3f}")
if verbose > 1:
log_info(f"Best solution: {best_solution}")
# 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
if sa_temperature < final_sa_temperature:
log_warning(f"SA Temperature is too low. Stopping the optimization process.")
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
)