From f88896aa6608bdb8e634999f997e5c61c99983d0 Mon Sep 17 00:00:00 2001 From: Pham Vu Tuan Dat Date: Sat, 4 Jul 2026 12:02:25 +0700 Subject: [PATCH] Add HSEvo method and task/example scripts Integrate HSEvo (Diversity-Driven Harmony Search + Genetic Algorithm, AAAI 2025) as llm4ad.method.hsevo, with HSEvoProfiler variants. Add run_hsevo.py entry scripts for every example task that already had run_eoh.py (bin packing, TSP, CVRP, QAP, FSSP, orienteering, VRPTW, pymoo MOEA/D, car racing, moon lander, circle packing), plus a method-level example and README under example/methods/hsevo/. Document HSEvo in the top-level README (quick start, method table, example links). Formatted with Black. --- README.md | 1 + example/methods/hsevo/README.md | 133 +++ example/methods/hsevo/run_hsevo_obp.py | 45 + .../EoH_settings&logs/run_hsevo.py | 45 + example/tasks/control_carracing/run_hsevo.py | 56 + example/tasks/control_moonlander/run_hsevo.py | 96 ++ example/tasks/cvrp_construct/run_hsevo.py | 45 + example/tasks/fssp_construct/run_hsevo.py | 45 + example/tasks/online_bin_packing/run_hsevo.py | 47 + .../tasks/orienteering_construct/run_hsevo.py | 51 + example/tasks/pymoo_moead/run_hsevo.py | 45 + example/tasks/qap/run_hsevo.py | 45 + example/tasks/tsp_construct/run_hsevo.py | 45 + example/tasks/vrptw_construct/run_hsevo.py | 45 + llm4ad/method/hsevo/__init__.py | 2 + llm4ad/method/hsevo/hsevo.py | 1040 +++++++++++++++++ llm4ad/method/hsevo/paras.yaml | 12 + llm4ad/method/hsevo/profiler.py | 217 ++++ llm4ad/method/hsevo/prompt.py | 159 +++ llm4ad/method/hsevo/util.py | 156 +++ 20 files changed, 2330 insertions(+) create mode 100644 example/methods/hsevo/README.md create mode 100644 example/methods/hsevo/run_hsevo_obp.py create mode 100644 example/tasks/circle_packing/EoH_settings&logs/run_hsevo.py create mode 100644 example/tasks/control_carracing/run_hsevo.py create mode 100644 example/tasks/control_moonlander/run_hsevo.py create mode 100644 example/tasks/cvrp_construct/run_hsevo.py create mode 100644 example/tasks/fssp_construct/run_hsevo.py create mode 100644 example/tasks/online_bin_packing/run_hsevo.py create mode 100644 example/tasks/orienteering_construct/run_hsevo.py create mode 100644 example/tasks/pymoo_moead/run_hsevo.py create mode 100644 example/tasks/qap/run_hsevo.py create mode 100644 example/tasks/tsp_construct/run_hsevo.py create mode 100644 example/tasks/vrptw_construct/run_hsevo.py create mode 100644 llm4ad/method/hsevo/__init__.py create mode 100644 llm4ad/method/hsevo/hsevo.py create mode 100644 llm4ad/method/hsevo/paras.yaml create mode 100644 llm4ad/method/hsevo/profiler.py create mode 100644 llm4ad/method/hsevo/prompt.py create mode 100644 llm4ad/method/hsevo/util.py diff --git a/README.md b/README.md index 47e55f22..67d68cc9 100644 --- a/README.md +++ b/README.md @@ -176,6 +176,7 @@ Check [GUI Introduction](https://llm4ad-doc.readthedocs.io/en/latest/getting_sta | Methods | Paper title | | --------------------------------- | ------------------------------------------------------------ | | **EoH** | [Evolution of Heuristics: Towards Efficient Automatic Algorithm Design Using Large Language Model](https://openreview.net/pdf?id=BwAkaxqiLB) (ICML 2024)
[Algorithm Evolution using Large Language Model](https://arxiv.org/abs/2311.15249) (Arxiv 2023, AEL, the early version of EoH) | +| **HSEvo** | [HSEvo: Elevating Automatic Heuristic Design with Diversity-Driven Harmony Search and Genetic Algorithm Using LLMs](https://arxiv.org/abs/2412.14995) (AAAI 2025) | | **MEoH** | [Multi-objective Evolution of Heuristic Using Large Language Model](https://arxiv.org/abs/2409.16867) (AAAI 25) | | **FunSearch** | [Mathematical Discoveries from Program Search with Large Language Models](https://www.nature.com/articles/s41586-023-06924-6) (Nature 2024) | | **(1+1)-EPS**
(HillClimbing) | [Understanding the Importance of Evolutionary Search in Automated Heuristic Design with Large Language Models](https://arxiv.org/abs/2407.10873) (PPSN 2024) | diff --git a/example/methods/hsevo/README.md b/example/methods/hsevo/README.md new file mode 100644 index 00000000..5a913bd7 --- /dev/null +++ b/example/methods/hsevo/README.md @@ -0,0 +1,133 @@ +# HSEvo: Diversity-Driven Harmony Search + Genetic Algorithm + +**HSEvo** (*Harmony Search Evolution*) is an LLM-based evolutionary program search method that combines genetic operators with a **harmony search (HS)** operator for automatic heuristic design. It is integrated into [LLM4AD](https://github.com/Optima-CityU/llm4ad) as `llm4ad.method.hsevo`. + +**Paper:** [HSEvo: Elevating Automatic Heuristic Design with Diversity-Driven Harmony Search and Genetic Algorithm Using LLMs](https://doi.org/10.1609/aaai.v39i25.34898) (AAAI 2025) + +**Upstream code:** [datphamvn/HSEvo](https://github.com/datphamvn/hsevo) + +--- + +## Overview + +Each HSEvo generation runs: + +1. **Selection** — random parent pairs with different objective values +2. **Reflection** — flash + comprehensive reflection prompts on the population +3. **Crossover** — LLM combines two parents into a new heuristic +4. **Mutation** — LLM mutates the elitist using reflection hints +5. **Harmony search** — pick one untuned individual, ask the LLM to expose tunable numeric parameters, then search parameter space with classical HS (`hmcr`, `par`, `bandwidth`) + +The HS step is the distinctive operator: the LLM rewrites hardcoded thresholds/weights as function defaults, defines `parameter_ranges`, and HSEvo evaluates multiple parameter settings on the **same** heuristic structure. + +Successful HS runs log `[HS-CHECK]` with `distinct_init_objs > 1`, confirming different parameter vectors produced different scores. + +--- + +## Quick Start + +Configure your LLM API, then run any task script: + +```bash +uv run python example/tasks/online_bin_packing/run_hsevo.py +``` + +Minimal Python example: + +```python +from llm4ad.task.optimization.online_bin_packing import OBPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + +llm = HttpsApi(host='xxx', key='sk-xxx', model='xxx', timeout=60) +task = OBPEvaluation() + +method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir='logs/hsevo', log_style='simple'), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, +) + +method.run() +``` + +A copy of this OBP script lives at [`run_hsevo_obp.py`](./run_hsevo_obp.py). + +--- + +## Hyper-parameters + +| Parameter | Default | Description | +|-----------|---------|-------------| +| `max_sample_nums` | `450` | Stop after this many function evaluations | +| `pop_size` | `10` | Parents used for crossover each generation | +| `init_pop_size` | `30` | Initial population size (rotating scientist personas) | +| `mutation_rate` | `0.5` | Fraction of `pop_size` mutated from the elitist | +| `hm_size` | `5` | Harmony memory size (initial random parameter vectors) | +| `hmcr` | `0.7` | Harmony memory considering rate | +| `par` | `0.5` | Pitch adjustment rate | +| `bandwidth` | `0.2` | Pitch adjustment bandwidth (fraction of each param range) | +| `max_iter` | `5` | HS improvisation iterations after memory init | +| `num_samplers` | `4` | Parallel LLM sampling threads | +| `num_evaluators` | `4` | Parallel evaluation workers | + +Defaults are also listed in [`llm4ad/method/hsevo/paras.yaml`](../../../llm4ad/method/hsevo/paras.yaml). + +--- + +## Task entry scripts + +Every task under [`example/tasks/`](../../tasks/) that has `run_eoh.py` also provides `run_hsevo.py`: + +| Task | Script | +|------|--------| +| Online bin packing | [`example/tasks/online_bin_packing/run_hsevo.py`](../../tasks/online_bin_packing/run_hsevo.py) | +| TSP constructive | [`example/tasks/tsp_construct/run_hsevo.py`](../../tasks/tsp_construct/run_hsevo.py) | +| CVRP constructive | [`example/tasks/cvrp_construct/run_hsevo.py`](../../tasks/cvrp_construct/run_hsevo.py) | +| QAP | [`example/tasks/qap/run_hsevo.py`](../../tasks/qap/run_hsevo.py) | +| FSSP / JSSP | [`example/tasks/fssp_construct/run_hsevo.py`](../../tasks/fssp_construct/run_hsevo.py) | +| Orienteering | [`example/tasks/orienteering_construct/run_hsevo.py`](../../tasks/orienteering_construct/run_hsevo.py) | +| VRPTW | [`example/tasks/vrptw_construct/run_hsevo.py`](../../tasks/vrptw_construct/run_hsevo.py) | +| Pymoo MOEA/D | [`example/tasks/pymoo_moead/run_hsevo.py`](../../tasks/pymoo_moead/run_hsevo.py) | +| Car racing control | [`example/tasks/control_carracing/run_hsevo.py`](../../tasks/control_carracing/run_hsevo.py) | +| Moon lander control | [`example/tasks/control_moonlander/run_hsevo.py`](../../tasks/control_moonlander/run_hsevo.py) | +| Circle packing | [`example/tasks/circle_packing/EoH_settings&logs/run_hsevo.py`](../../tasks/circle_packing/EoH_settings&logs/run_hsevo.py) | + +--- + +## Logging + +`HSEvoProfiler` writes: + +- `run_log.txt` — generation progress, `[HS-CHECK]`, `harmony_search: OK/FAILED` +- `samples/` — evaluated programs tagged by `operator` (`init`, `crossover`, `mutation`, `harmony_search`, …) +- `population/` — per-generation population checkpoints + +Profiler variants: `HSEvoTensorboardProfiler`, `HSEvoWandbProfiler`. + +--- + +## Citation + +```bibtex +@inproceedings{dat2025hsevo, + title={Hsevo: Elevating automatic heuristic design with diversity-driven harmony search and genetic algorithm using llms}, + author={Dat, Pham Vu Tuan and Doan, Long and Binh, Huynh Thi Thanh}, + booktitle={Proceedings of the AAAI Conference on Artificial Intelligence}, + volume={39}, + number={25}, + pages={26931--26938}, + year={2025} +} +``` diff --git a/example/methods/hsevo/run_hsevo_obp.py b/example/methods/hsevo/run_hsevo_obp.py new file mode 100644 index 00000000..acfa5f02 --- /dev/null +++ b/example/methods/hsevo/run_hsevo_obp.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.online_bin_packing import OBPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=60, + ) + + task = OBPEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/circle_packing/EoH_settings&logs/run_hsevo.py b/example/tasks/circle_packing/EoH_settings&logs/run_hsevo.py new file mode 100644 index 00000000..39af57f5 --- /dev/null +++ b/example/tasks/circle_packing/EoH_settings&logs/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../.."))) + +from evaluation import CirclePackingEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=120, + ) + + task = CirclePackingEvaluation(timeout_seconds=1200) + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/control_carracing/run_hsevo.py b/example/tasks/control_carracing/run_hsevo.py new file mode 100644 index 00000000..7ed69b9b --- /dev/null +++ b/example/tasks/control_carracing/run_hsevo.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.machine_learning.car_racing import RacingCarEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=120, + ) + + seeds = [1] + instance_set = {idx: seed for idx, seed in enumerate(seeds)} + using_seeds = list(range(10, 20)) + ins_to_be_solve_set = {idx: seed for idx, seed in enumerate(using_seeds)} + + task = RacingCarEvaluation( + whocall="eoh", + run_mode="Training", + instance_set=instance_set, + ins_to_be_solve_set=ins_to_be_solve_set, + objective_value=100, + ) + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/control_moonlander/run_hsevo.py b/example/tasks/control_moonlander/run_hsevo.py new file mode 100644 index 00000000..d6dc8861 --- /dev/null +++ b/example/tasks/control_moonlander/run_hsevo.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.machine_learning.moon_lander import ( + MoonLanderEvaluation, + moon_lander_feature, +) +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=120, + ) + + seeds = [ + 6, + 9, + 17, + 29, + 57, + 44, + 18, + 69, + 26, + 68, + 65, + 23, + 51, + 93, + 16, + 87, + 92, + 90, + 22, + 73, + 60, + 10, + 19, + 97, + 11, + 14, + 99, + 98, + 8, + 28, + 43, + 56, + 89, + 15, + 74, + ] + instance_set = {idx: seed for idx, seed in enumerate(seeds)} + using_seeds = list(range(100, 150)) + ins_to_be_solve_set = {idx: seed for idx, seed in enumerate(using_seeds)} + + task = MoonLanderEvaluation( + whocall="eoh", + instance_set=instance_set, + run_mode="Training", + ins_to_be_solve_set=ins_to_be_solve_set, + feature_pipeline=moon_lander_feature, + objective_value=230, + ) + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/cvrp_construct/run_hsevo.py b/example/tasks/cvrp_construct/run_hsevo.py new file mode 100644 index 00000000..10e94d96 --- /dev/null +++ b/example/tasks/cvrp_construct/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.cvrp_construct import CVRPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=60, + ) + + task = CVRPEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/fssp_construct/run_hsevo.py b/example/tasks/fssp_construct/run_hsevo.py new file mode 100644 index 00000000..ed3092ae --- /dev/null +++ b/example/tasks/fssp_construct/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.jssp_construct import JSSPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=120, + ) + + task = JSSPEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/online_bin_packing/run_hsevo.py b/example/tasks/online_bin_packing/run_hsevo.py new file mode 100644 index 00000000..6fdc5327 --- /dev/null +++ b/example/tasks/online_bin_packing/run_hsevo.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import os +import sys + +# Make `llm4ad` importable no matter where this script is launched from. +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.online_bin_packing import OBPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=60, + ) + + task = OBPEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, # HSEvo's 'max_fe' + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + # harmony search hyper-parameters + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/orienteering_construct/run_hsevo.py b/example/tasks/orienteering_construct/run_hsevo.py new file mode 100644 index 00000000..dd7e2052 --- /dev/null +++ b/example/tasks/orienteering_construct/run_hsevo.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.orienteering_construct import OrienteeringEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-4o-mini' + timeout=60, + ) + + task = OrienteeringEvaluation( + timeout_seconds=20, + n_instance=8, + problem_size=30, + max_length_ratio=0.35, + seed=2024, + ) + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/pymoo_moead/run_hsevo.py b/example/tasks/pymoo_moead/run_hsevo.py new file mode 100644 index 00000000..652693bb --- /dev/null +++ b/example/tasks/pymoo_moead/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.pymoo_moead import MOEAD_PYMOO_Evaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=100, + ) + + task = MOEAD_PYMOO_Evaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/qap/run_hsevo.py b/example/tasks/qap/run_hsevo.py new file mode 100644 index 00000000..e9f98ed3 --- /dev/null +++ b/example/tasks/qap/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.qap_construct import QAPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=120, + ) + + task = QAPEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/tsp_construct/run_hsevo.py b/example/tasks/tsp_construct/run_hsevo.py new file mode 100644 index 00000000..76ee9ff4 --- /dev/null +++ b/example/tasks/tsp_construct/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.tsp_construct import TSPEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=60, + ) + + task = TSPEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/example/tasks/vrptw_construct/run_hsevo.py b/example/tasks/vrptw_construct/run_hsevo.py new file mode 100644 index 00000000..315776f4 --- /dev/null +++ b/example/tasks/vrptw_construct/run_hsevo.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import sys + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from llm4ad.task.optimization.vrptw_construct import VRPTWEvaluation +from llm4ad.tools.llm.llm_api_https import HttpsApi +from llm4ad.method.hsevo import HSEvo, HSEvoProfiler + + +def main(): + llm = HttpsApi( + host="xxx", # your host endpoint, e.g., 'api.openai.com', 'api.deepseek.com' + key="sk-xxx", # your key, e.g., 'sk-abcdefghijklmn' + model="xxx", # your llm, e.g., 'gpt-3.5-turbo' + timeout=60, + ) + + task = VRPTWEvaluation() + + method = HSEvo( + llm=llm, + profiler=HSEvoProfiler(log_dir="logs/hsevo", log_style="simple"), + evaluation=task, + max_sample_nums=100, + pop_size=4, + init_pop_size=10, + mutation_rate=0.5, + hm_size=5, + hmcr=0.7, + par=0.5, + bandwidth=0.2, + max_iter=5, + num_samplers=4, + num_evaluators=4, + debug_mode=False, + ) + + method.run() + + +if __name__ == "__main__": + main() diff --git a/llm4ad/method/hsevo/__init__.py b/llm4ad/method/hsevo/__init__.py new file mode 100644 index 00000000..75b9a6ce --- /dev/null +++ b/llm4ad/method/hsevo/__init__.py @@ -0,0 +1,2 @@ +from .hsevo import HSEvo +from .profiler import HSEvoProfiler, HSEvoTensorboardProfiler, HSEvoWandbProfiler diff --git a/llm4ad/method/hsevo/hsevo.py b/llm4ad/method/hsevo/hsevo.py new file mode 100644 index 00000000..68b41fd5 --- /dev/null +++ b/llm4ad/method/hsevo/hsevo.py @@ -0,0 +1,1040 @@ +from __future__ import annotations + +import concurrent.futures +import contextlib +import logging +import math +import time +import traceback +from threading import Lock +from typing import Literal, Optional + +import numpy as np + +from .profiler import HSEvoProfiler +from .prompt import ( + SYSTEM_GENERATOR, + USER_GENERATOR, + SEED, + CROSSOVER, + MUTATION, + SYSTEM_REFLECTOR, + USER_FLASH_REFLECTION, + USER_COMPREHENSIVE_REFLECTION, + SYSTEM_HARMONY_SEARCH, + HARMONY_SEARCH, + SCIENTISTS, + make_func_signature, + make_seed_func, + make_func_desc, +) +from .util import ( + extract_code_from_generator, + filter_code, + extract_to_hs, + format_messages, +) +from ...base import ( + Evaluation, + LLM, + Function, + Program, + TextFunctionProgramConverter, + SecureEvaluator, +) +from ...tools.profiler import ProfilerBase + + +class HSEvo: + def __init__( + self, + llm: LLM, + evaluation: Evaluation, + profiler: ProfilerBase = None, + max_sample_nums: Optional[int] = 450, + pop_size: Optional[int] = 10, + init_pop_size: Optional[int] = 30, + mutation_rate: float = 0.5, + hm_size: int = 5, + hmcr: float = 0.7, + par: float = 0.5, + bandwidth: float = 0.2, + max_iter: int = 5, + num_samplers: int = 4, + num_evaluators: int = 4, + *, + resume_mode: bool = False, + debug_mode: bool = False, + multi_thread_or_process_eval: Literal["thread", "process"] = "thread", + **kwargs, + ): + """HSEvo: Diversity-Driven Harmony Search + Genetic Algorithm using LLMs. + + Args: + llm : an instance of 'llm4ad.base.LLM'. + evaluation : an instance of 'llm4ad.base.Evaluation'. + profiler : an instance of 'llm4ad.method.hsevo.HSEvoProfiler'. Pass 'None' to disable. + max_sample_nums : terminate after this many evaluated functions (maps to HSEvo's 'max_fe'). + pop_size : population size used for selection/crossover. + init_pop_size : number of individuals generated for the initial population (with rotating personas). + mutation_rate : fraction of pop_size mutated from the elitist each generation. + hm_size : harmony-memory size for the harmony search operator. + hmcr : harmony memory considering rate. + par : pitch adjustment rate. + bandwidth : pitch adjustment bandwidth (fraction of each parameter range). + max_iter : number of harmony-search improvisation iterations. + num_samplers : number of threads used for batched LLM sampling. + num_evaluators : number of workers used for parallel evaluation. + resume_mode : if True, skip the initial population creation (see note in `run`). + debug_mode : if True, print detailed information. + multi_thread_or_process_eval: 'thread' or 'process' pool for evaluation. + **kwargs : extra args passed to 'llm4ad.base.SecureEvaluator' (e.g. 'fork_proc'). + """ + # ----- LLM4AD framework handles ----- + self._llm = llm + self._profiler = profiler + self._max_sample_nums = max_sample_nums + self._resume_mode = resume_mode + self._debug_mode = debug_mode + llm.debug_mode = debug_mode + self._num_samplers = num_samplers + self._num_evaluators = num_evaluators + + # ----- HSEvo hyper-parameters (names mirror the original cfg.*) ----- + self.mutation_rate = mutation_rate + self.init_pop_size = init_pop_size + self.pop_size = pop_size + self.hm_size = hm_size + self.hmcr = hmcr + self.par = par + self.bandwidth = bandwidth + self.max_iter = max_iter + # Additive temperature boost applied ONLY when sampling the initial + # population (mirrors upstream's `cfg.temperature + 0.3`). Applied via + # `_temporarily_raise_temperature` if the LLM exposes `_temperature`. + self.init_temperature_boost = float(kwargs.pop("init_temperature_boost", 0.3)) + + # ----- HSEvo state (verbatim) ----- + self.iteration = 0 + self.function_evals = 0 + # generation counter + operator success bookkeeping (logging/verification only) + self._generation = 0 + self._op_stats = {"crossover": 0, "mutation": 0, "harmony_search": 0} + self._op_ok_gens = { + "crossover": set(), + "mutation": set(), + "harmony_search": set(), + } + self.elitist = None + self.best_obj_overall = None + self.best_code_overall = None + self.best_code_path_overall = None + self.long_term_reflection_str = "" + self.lst_good_reflection = [] + self.lst_bad_reflection = [] + self.population = [] + self.seed_ind = None + self.str_flash_memory = {"analyze": "", "exp": ""} + # LLM4AD maximizes `score`; HSEvo minimizes `obj`. We always map obj = -score. + self.obj_type = "min" + + # ----- Problem definition derived from the LLM4AD task ----- + self._template_program_str = evaluation.template_program + self._task_description_str = evaluation.task_description + self._template_program: Program = TextFunctionProgramConverter.text_to_program( + self._template_program_str + ) + self._function_to_evolve: Function = ( + TextFunctionProgramConverter.text_to_function(self._template_program_str) + ) + if self._function_to_evolve is None or self._template_program is None: + raise ValueError( + "HSEvo: could not parse the task `template_program` into a single function." + ) + self.func_name = self._function_to_evolve.name + self.problem_desc = self._task_description_str + self.func_signature = make_func_signature(self._function_to_evolve) + self.seed_func = make_seed_func(self._function_to_evolve) + self.func_desc = make_func_desc( + self._function_to_evolve, self._task_description_str + ) + self.external_knowledge = "" + self.str_comprehensive_memory = self.external_knowledge + + # ----- Prompts (verbatim HSEvo common templates) ----- + self.system_generator_prompt = SYSTEM_GENERATOR + self.user_generator_prompt = USER_GENERATOR + self.crossover_prompt = CROSSOVER + self.mutation_prompt = MUTATION + self.system_reflector_prompt = SYSTEM_REFLECTOR + self.user_flash_reflection_prompt = USER_FLASH_REFLECTION + self.user_comprehensive_reflection_prompt = USER_COMPREHENSIVE_REFLECTION + self.system_hs_prompt = SYSTEM_HARMONY_SEARCH + self.hs_prompt = HARMONY_SEARCH + self.seed_prompt = SEED.format( + seed_func=self.seed_func, func_name=self.func_name + ) + self.scientists = SCIENTISTS + + # ----- print-once flags (verbatim) ----- + self.print_crossover_prompt = True + self.print_mutate_prompt = True + self.print_flash_reflection_prompt = True + self.print_comprehensive_reflection_prompt = True + self.print_hs_prompt = True + self.local_sel_hs = None + + # ----- evaluator + executors ----- + self._evaluator = SecureEvaluator(evaluation, debug_mode=debug_mode, **kwargs) + assert multi_thread_or_process_eval in ["thread", "process"] + if multi_thread_or_process_eval == "thread": + self._evaluation_executor = concurrent.futures.ThreadPoolExecutor( + max_workers=num_evaluators + ) + else: + self._evaluation_executor = concurrent.futures.ProcessPoolExecutor( + max_workers=num_evaluators + ) + + # ----- sampling bookkeeping (for profiler sample_time / operator) ----- + self._cur_sample_time = 0.0 + self._cur_operator = "init" + self._uid_counter = 0 + self._uid_lock = Lock() + + logging.info("Problem: " + str(self.func_name)) + logging.info("Function name: " + str(self.func_name)) + + # pass parameters to profiler + if profiler is not None: + self._profiler.record_parameters(llm, evaluation, self) + + # ------------------------------------------------------------------ + # Boundary 1: LLM adapter (replaces HSEvo's multi_chat_completion) + # ------------------------------------------------------------------ + def _safe_draw(self, messages) -> str: + """Draw one sample. Pass the [system, user] message list directly to the + backend (HttpsApi/OpenAIAPI support it). Fall back to a concatenated + single string for backends that only accept strings.""" + try: + return self._llm.draw_sample(messages) + except Exception: + try: + text = "\n\n".join( + m.get("content", "") for m in messages if isinstance(m, dict) + ) + return self._llm.draw_sample(text) + except Exception: + if self._debug_mode: + traceback.print_exc() + return "" + + def _draw_batch(self, messages_list, n: int = 1): + """Mimic HSEvo's `multi_chat_completion(messages_list, n, model, temperature)`. + + - If a single message list is given, it is wrapped into a batch. + - `n > 1` duplicates the single prompt `n` times (LLM4AD backends do not + expose an `n` argument), matching HSEvo's multi-sample-per-prompt use. + - The per-call temperature/model used by upstream are ignored (the LLM + instance is pre-configured). + """ + if len(messages_list) > 0 and not isinstance(messages_list[0], list): + messages_list = [messages_list] + tasks = messages_list if n == 1 else messages_list * n + start = time.time() + with concurrent.futures.ThreadPoolExecutor( + max_workers=max(self._num_samplers, 1) + ) as ex: + responses = list(ex.map(self._safe_draw, tasks)) + elapsed = time.time() - start + self._cur_sample_time = elapsed / max(len(tasks), 1) + return responses + + @contextlib.contextmanager + def _temporarily_raise_temperature(self, delta: float): + """Temporarily add `delta` to the LLM's sampling temperature. + + LLM4AD's `LLM.draw_sample` has no per-call temperature argument, so this + mutates a `_temperature` attribute for backends that expose one (e.g. + the `VLLMChat` client used in the HSEvo comparison, which reads + `self._temperature` on every call). It is a no-op for backends without + such an attribute or when `delta == 0`. The original value is always + restored, even on error. Used to reproduce upstream's +0.3 init boost. + """ + llm = self._llm + if not delta or not hasattr(llm, "_temperature"): + yield + return + original = llm._temperature + try: + llm._temperature = original + delta + logging.info( + f"Init sampling temperature raised {original} -> {llm._temperature} " + f"(+{delta}) for initial-population diversity." + ) + yield + finally: + llm._temperature = original + + def _next_uid(self) -> str: + with self._uid_lock: + self._uid_counter += 1 + return f"ind_{self._uid_counter}" + + # ------------------------------------------------------------------ + # Boundary 2: evaluation adapter (replaces HSEvo's subprocess eval) + # ------------------------------------------------------------------ + def _code_to_program(self, code: str) -> Optional[Program]: + """Convert a generated code string into an LLM4AD `Program`. + + The generated function's full signature (including any default values + introduced by harmony search) is kept; only the function NAME is + normalized to the task's canonical name so name-based lookups work. + """ + try: + gen_program = TextFunctionProgramConverter.text_to_program(code) + if gen_program is None or len(gen_program.functions) != 1: + # LLM4AD's SecureEvaluator assumes exactly one top-level function. + return None + gen_func = gen_program.functions[0] + gen_func.name = self.func_name + return Program(preface=gen_program.preface, functions=[gen_func]) + except Exception: + return None + + def _register_function( + self, program: Program, score, sample_time, evaluate_time, operator + ): + if self._profiler is None: + return + func = TextFunctionProgramConverter.program_to_function(program) + if func is None: + return + func.score = score + func.sample_time = sample_time + func.evaluate_time = evaluate_time + func.operator = operator + if func.docstring: + func.algorithm = func.docstring + self._profiler.register_function(func, program=str(program)) + + def evaluate_population( + self, population: list[dict], hs_try_idx: int = None + ) -> list[dict]: + """Evaluate a population using LLM4AD's SecureEvaluator (parallel). + + Mirrors HSEvo's `evaluate_population` accounting (one `function_evals` + increment per individual, best-obj logging), but the inner evaluation is + delegated to `SecureEvaluator` and `obj = -score`. + """ + programs = [] + futures = [] + + # submit + for response_id in range(len(population)): + self.function_evals += 1 + individual = population[response_id] + + program = None + if individual["code"] is not None: + program = self._code_to_program(individual["code"]) + programs.append(program) + + if program is None: + self.mark_invalid_individual( + individual, "Invalid response / unparseable code!" + ) + futures.append(None) + continue + + try: + futures.append( + self._evaluation_executor.submit( + self._evaluator.evaluate_program_record_time, program + ) + ) + except Exception as e: + logging.info(f"Error for response_id {response_id}: {e}") + self.mark_invalid_individual(individual, str(e)) + futures.append(None) + + # collect + for response_id, future in enumerate(futures): + individual = population[response_id] + program = programs[response_id] + if future is None: + continue + try: + score, eval_time = future.result() + except Exception as e: + logging.info(f"Error for response_id {response_id}: {e}") + score, eval_time = None, None + + if score is None or (isinstance(score, float) and math.isnan(score)): + self.mark_invalid_individual(individual, "Invalid objective value!") + else: + individual["exec_success"] = True + individual["obj"] = -float( + score + ) # LLM4AD maximizes score => HSEvo minimizes obj + + self._register_function( + program, + score, + individual.get("_sample_time"), + eval_time, + individual.get("_operator", "unknown"), + ) + + # Log after all population is evaluated + valid_objs = [ind["obj"] for ind in population if ind.get("exec_success")] + best_obj = min(valid_objs) if valid_objs else float("inf") + logging.info(f"Eval={self.function_evals}, BestObj={best_obj}") + + return population + + # ------------------------------------------------------------------ + # Individual bookkeeping (verbatim, minus on-disk artifacts) + # ------------------------------------------------------------------ + def response_to_individual( + self, response: str, response_id: int, file_name: str = None + ) -> dict: + """Convert an LLM response (or a substituted code string) to an individual.""" + code = extract_code_from_generator(response) + individual = { + "code_path": self._next_uid(), + "code": code, + "response_id": response_id, + "tryHS": False, + "_sample_time": self._cur_sample_time, + "_operator": self._cur_operator, + } + return individual + + def mark_invalid_individual(self, individual: dict, traceback_msg: str) -> dict: + """Mark an individual as invalid (verbatim).""" + individual["exec_success"] = False + individual["obj"] = float("inf") + individual["traceback_msg"] = traceback_msg + return individual + + # ------------------------------------------------------------------ + # Initialization (verbatim, LLM/eval boundaries swapped) + # ------------------------------------------------------------------ + def init_population(self) -> None: + # Evaluate the seed function, and set it as Elite + logging.info("Evaluating seed function...") + self._cur_operator = "init" + self._cur_sample_time = 0.0 + code = extract_code_from_generator(self.seed_func).replace("v1", "v2") + logging.info("Seed function code: \n" + str(code)) + seed_ind = { + "code_path": self._next_uid(), + "code": code, + "response_id": 0, + "tryHS": False, + "_sample_time": 0.0, + "_operator": "init", + } + self.seed_ind = seed_ind + self.population = self.evaluate_population([seed_ind]) + + # If seed function is invalid, stop + if not self.seed_ind["exec_success"]: + raise RuntimeError( + "Seed function is invalid. Please check the task template_program." + ) + + self.update_iter() + + messages_lst = [] + for i in range(self.init_pop_size): + user_generator_prompt_full = self.user_generator_prompt.format( + seed=self.scientists[i % len(self.scientists)], + func_name=self.func_name, + problem_desc=self.problem_desc, + func_desc=self.func_desc, + ) + system_generator_prompt_full = self.system_generator_prompt.format( + seed=self.scientists[i % len(self.scientists)] + ) + system = system_generator_prompt_full + user = ( + user_generator_prompt_full + + "\n" + + self.seed_prompt + + "\n" + + self.long_term_reflection_str + ) + messages = format_messages(system, user) + messages_lst.append(messages) + + # Upstream HSEvo raises the sampling temperature by +0.3 for the initial + # population to diversify it (main.py: `self.cfg.temperature + 0.3`). + # LLM4AD's `LLM.draw_sample` has no per-call temperature, but backends + # that expose a mutable `_temperature` (e.g. the VLLMChat used for the + # HSEvo comparison) can be bumped for the duration of the init batch and + # restored afterwards. This restores upstream's init diversity, which is + # what prevents the population from collapsing onto the best-fit + # attractor (see _compare/ analysis). + self._cur_operator = "init" + with self._temporarily_raise_temperature(self.init_temperature_boost): + responses = self._draw_batch(messages_lst, 1) + population = [ + self.response_to_individual(response, response_id) + for response_id, response in enumerate(responses) + ] + + # Run code and evaluate population + population = self.evaluate_population(population) + + # Update iteration + self.population = population + self.update_iter() + + # ------------------------------------------------------------------ + # Update / selection (verbatim) + # ------------------------------------------------------------------ + def update_iter(self) -> None: + """Update after each iteration (verbatim).""" + population = self.population + objs = [individual["obj"] for individual in population] + best_obj, best_sample_idx = min(objs), np.argmin(np.array(objs)) + + # update best overall + if self.best_obj_overall is None or best_obj < self.best_obj_overall: + self.best_obj_overall = best_obj + self.best_code_overall = population[best_sample_idx]["code"] + self.best_code_path_overall = population[best_sample_idx]["code_path"] + + # update elitist + if self.elitist is None or best_obj < self.elitist["obj"]: + self.elitist = population[best_sample_idx] + logging.info(f"Iteration {self.iteration}: Elitist: {self.elitist['obj']}") + + self.iteration += 1 + + def random_select(self, population: list[dict]) -> list[dict]: + """Random selection with equal probability (verbatim). + + Note: HSEvo's black-box branch (which also filters `obj < seed_obj`) is + omitted because LLM4AD has no notion of `problem_type`; we always keep + the valid-individuals branch. + """ + selected_population = [] + # Eliminate invalid individuals + population = [ + individual for individual in population if individual["exec_success"] + ] + if len(population) < 2: + return None + trial = 0 + while len(selected_population) < 2 * self.pop_size: + trial += 1 + parents = np.random.choice(population, size=2, replace=False) + # If two parents have the same objective value, consider them as identical; + # otherwise, add them to the selected population + if parents[0]["obj"] != parents[1]["obj"]: + selected_population.extend(parents) + if trial > 1000: + return None + return selected_population + + # ------------------------------------------------------------------ + # Reflection (verbatim, LLM boundary swapped) + # ------------------------------------------------------------------ + def flash_reflection(self, population: list[dict]) -> None: + self._cur_operator = "flash_reflection" + lst_str_method = [] + seen_elements = set() + + sorted_population = sorted(population, key=lambda x: x["obj"], reverse=False) + for idx, individual in enumerate(sorted_population): + suffix = ( + "th" + if 11 <= idx + 1 <= 13 + else {1: "st", 2: "nd", 3: "rd"}.get((idx + 1) % 10, "th") + ) + str_idx_method = f"[Heuristics {idx + 1}{suffix}]" + str_code = individual["code"] + temp_str = str_idx_method + "\n" + str_code + "\n" + + if temp_str not in seen_elements: + seen_elements.add(temp_str) + lst_str_method.append(temp_str) + + system = self.system_reflector_prompt + user = self.user_flash_reflection_prompt.format( + problem_desc=self.problem_desc, + lst_method="\n".join(lst_str_method), + schema_reflection={"analyze": "str", "exp": "str"}, + ) + messages = format_messages(system, user) + + if self.print_flash_reflection_prompt: + logging.info( + "Flash reflection Prompt: \nSystem Prompt: \n" + + system + + "\nUser Prompt: \n" + + user + ) + self.print_flash_reflection_prompt = False + + flash_reflection_res = self._draw_batch([messages], 1)[0] + analyze_start = flash_reflection_res.find("**Analysis:**") + len( + "**Analysis:**" + ) + exp_start = flash_reflection_res.find("**Experience:**") + + analysis_text = flash_reflection_res[analyze_start:exp_start].strip() + experience_text = flash_reflection_res[ + exp_start + len("**Experience:**") : + ].strip() + + # Create the JSON structure + self.str_flash_memory = {"analyze": analysis_text, "exp": experience_text} + + def comprehensive_reflection(self): + self._cur_operator = "comprehensive_reflection" + system = self.system_reflector_prompt + + good_reflection = ( + "\n\n".join(self.lst_good_reflection) + if len(self.lst_good_reflection) > 0 + else "None" + ) + bad_reflection = ( + "\n\n".join(self.lst_bad_reflection) + if len(self.lst_bad_reflection) > 0 + else "None" + ) + + user = self.user_comprehensive_reflection_prompt.format( + bad_reflection=bad_reflection, + good_reflection=good_reflection, + curr_reflection=self.str_flash_memory["exp"], + ) + messages = format_messages(system, user) + + if self.print_comprehensive_reflection_prompt: + logging.info( + "Comprehensive reflection Prompt: \nSystem Prompt: \n" + + system + + "\nUser Prompt: \n" + + user + ) + self.print_comprehensive_reflection_prompt = False + + comprehensive_response = self._draw_batch([messages], 1)[0] + self.str_comprehensive_memory = ( + self.external_knowledge + "\n" + comprehensive_response + ) + + # ------------------------------------------------------------------ + # Crossover / mutation (verbatim, LLM boundary swapped) + # ------------------------------------------------------------------ + def crossover(self, population: list[dict]) -> list[dict]: + self._cur_operator = "crossover" + messages_lst = [] + for i in range(0, len(population), 2): + # Select two individuals + if population[i]["obj"] < population[i + 1]["obj"]: + parent_1 = population[i] + parent_2 = population[i + 1] + else: + parent_1 = population[i + 1] + parent_2 = population[i] + + # Crossover + system = self.system_generator_prompt.format(seed=self.scientists[0]) + func_signature_m1 = self.func_signature.format(version=0) + func_signature_m2 = self.func_signature.format(version=1) + user_generator_prompt_full = self.user_generator_prompt.format( + seed=self.scientists[0], + func_name=self.func_name, + problem_desc=self.problem_desc, + func_desc=self.func_desc, + ) + user = self.crossover_prompt.format( + user_generator=user_generator_prompt_full, + func_signature_m1=func_signature_m1, + func_signature_m2=func_signature_m2, + code_method1=filter_code(parent_1["code"]), + code_method2=filter_code(parent_2["code"]), + analyze=self.str_flash_memory["analyze"], + exp=self.str_comprehensive_memory, + func_name=self.func_name, + ) + messages = format_messages(system, user) + messages_lst.append(messages) + + # Print crossover prompt for the first iteration + if self.print_crossover_prompt: + logging.info( + "Crossover Prompt: \nSystem Prompt: \n" + + system + + "\nUser Prompt: \n" + + user + ) + self.print_crossover_prompt = False + + # Asynchronously generate responses + response_lst = self._draw_batch(messages_lst, 1) + crossed_population = [ + self.response_to_individual(response, response_id) + for response_id, response in enumerate(response_lst) + ] + + assert len(crossed_population) == self.pop_size + return crossed_population + + def mutate(self) -> list[dict]: + """Elitist-based mutation. We only mutate the best individual to generate n_pop new individuals.""" + self._cur_operator = "mutation" + system = self.system_generator_prompt.format(seed=self.scientists[0]) + func_signature1 = self.func_signature.format(version=1) + user_generator_prompt_full = self.user_generator_prompt.format( + seed=self.scientists[0], + func_name=self.func_name, + problem_desc=self.problem_desc, + func_desc=self.func_desc, + ) + + user = self.mutation_prompt.format( + user_generator=user_generator_prompt_full, + reflection=self.str_comprehensive_memory, + func_signature1=func_signature1, + elitist_code=filter_code(self.elitist["code"]), + func_name=self.func_name, + ) + messages = format_messages(system, user) + + if self.print_mutate_prompt: + logging.info( + "Mutation Prompt: \nSystem Prompt: \n" + + system + + "\nUser Prompt: \n" + + user + ) + self.print_mutate_prompt = False + + responses = self._draw_batch( + [messages], int(self.pop_size * self.mutation_rate) + ) + population = [ + self.response_to_individual(response, response_id) + for response_id, response in enumerate(responses) + ] + return population + + # ------------------------------------------------------------------ + # Harmony search (verbatim, LLM/eval boundaries swapped) + # ------------------------------------------------------------------ + def sel_individual_hs(self): + candidate_hs = [ + individual for individual in self.population if individual["tryHS"] is False + ] + best_candidate_id = self.find_best_obj(candidate_hs) + self.local_sel_hs = best_candidate_id + # NOTE: preserved verbatim from upstream HSEvo (best_candidate_id is an + # index into `candidate_hs` but is used to index `self.population`). + self.population[best_candidate_id]["tryHS"] = True + return self.population[best_candidate_id]["code"] + + def initialize_harmony_memory(self, bounds): + problem_size = len(bounds) + harmony_memory = np.zeros((self.hm_size, problem_size)) + for i in range(problem_size): + lower_bound, upper_bound = bounds[i] + harmony_memory[:, i] = np.random.uniform( + lower_bound, upper_bound, self.hm_size + ) + return harmony_memory + + def responses_to_population(self, responses, try_hs_idx=None) -> list[dict]: + """Convert responses (here, substituted code strings) to a population.""" + population = [] + for response_id, response in enumerate(responses): + individual = self.response_to_individual(response, response_id) + population.append(individual) + return population + + def create_population_hs( + self, str_code, parameter_ranges, harmony_memory, try_hs_idx=None + ): + str_create_pop = [] + for i in range(len(harmony_memory)): + tmp_str = str_code + for j in range(len(list(parameter_ranges))): + tmp_str = tmp_str.replace( + ("{" + list(parameter_ranges)[j] + "}"), str(harmony_memory[i][j]) + ) + if tmp_str == str_code: + return None + str_create_pop.append("```python\n" + tmp_str + "\n```") + + population_hs = self.responses_to_population(str_create_pop, try_hs_idx) + return self.evaluate_population(population_hs, try_hs_idx) + + def find_best_obj(self, population_hs): + objs = [individual["obj"] for individual in population_hs] + best_solution_id = np.argmin(np.array(objs)) + return best_solution_id + + def create_new_harmony(self, harmony_memory, bounds): + new_harmony = np.zeros((harmony_memory.shape[1],)) + for i in range(harmony_memory.shape[1]): + if np.random.rand() < self.hmcr: + new_harmony[i] = harmony_memory[ + np.random.randint(0, harmony_memory.shape[0]), i + ] + if np.random.rand() < self.par: + adjustment = ( + np.random.uniform(-1, 1) + * (bounds[i][1] - bounds[i][0]) + * self.bandwidth + ) + new_harmony[i] += adjustment + else: + new_harmony[i] = np.random.uniform(bounds[i][0], bounds[i][1]) + return new_harmony + + def update_harmony_memory( + self, + population_hs, + harmony_memory, + new_harmony, + func_block, + parameter_ranges, + try_hs_idx, + ): + objs = [individual["obj"] for individual in population_hs] + worst_index = np.argmax(np.array(objs)) + + new_individual = self.create_population_hs( + func_block, parameter_ranges, [new_harmony.tolist()], try_hs_idx + )[0] + + if new_individual["obj"] < population_hs[worst_index]["obj"]: + population_hs[worst_index] = new_individual + harmony_memory[worst_index] = new_harmony + return population_hs, harmony_memory + + def harmony_search(self): + # Safety guard (LLM4AD addition): if there is no candidate left to tune, + # skip rather than crashing on an empty argmin. + if not any(individual["tryHS"] is False for individual in self.population): + return None + + self._cur_operator = "harmony_search" + system = self.system_hs_prompt + user = self.hs_prompt.format(code_extract=self.sel_individual_hs()) + messages = format_messages(system, user) + # Print get hs prompt for the first iteration + if self.print_hs_prompt: + logging.info( + "Harmony Search Prompt: \nSystem Prompt: \n" + + system + + "\nUser Prompt: \n" + + user + ) + self.print_hs_prompt = False + + responses = self._draw_batch([messages], 1) + + logging.info("LLM Response for HS step: " + str(responses[0])) + parameter_ranges, func_block = extract_to_hs(responses[0]) + if parameter_ranges is None or func_block is None: + return None + bounds = [value for value in parameter_ranges.values()] + + harmony_memory = self.initialize_harmony_memory(bounds) + population_hs = self.create_population_hs( + func_block, parameter_ranges, harmony_memory + ) + + if population_hs is None: + return None + elif ( + len( + [ + individual + for individual in population_hs + if individual["exec_success"] is True + ] + ) + == 0 + ): + self.function_evals -= self.hm_size + return None + + # [HS-CHECK] + init_objs = [ind["obj"] for ind in population_hs if ind["exec_success"]] + n_distinct = len(set(init_objs)) + init_best = min(init_objs) if init_objs else float("inf") + + for iteration in range(self.max_iter): + new_harmony = self.create_new_harmony(harmony_memory, bounds) + population_hs, harmony_memory = self.update_harmony_memory( + population_hs, + harmony_memory, + new_harmony, + func_block, + parameter_ranges, + iteration, + ) + best_obj_id = self.find_best_obj(population_hs) + population_hs[best_obj_id]["tryHS"] = True + hs_best = population_hs[best_obj_id]["obj"] + logging.info( + f"[HS-CHECK] iter={self.iteration} hm_size={self.hm_size} " + f"valid={len(init_objs)} distinct_init_objs={n_distinct} " + f"init_best={init_best} hs_best={hs_best} " + f"improved_over_init={hs_best < init_best}" + ) + return population_hs[best_obj_id] + + # ------------------------------------------------------------------ + # Evolutionary loop (verbatim, minus on-disk artifacts) + # ------------------------------------------------------------------ + @staticmethod + def _valid_count(population: list[dict]) -> int: + return sum(1 for ind in population if ind.get("exec_success")) + + def evolve(self): + while self.function_evals < self._max_sample_nums: + self._generation += 1 + gen = self._generation + logging.info( + f"===== [Gen {gen}] start (function_evals={self.function_evals}, " + f"best_obj={self.best_obj_overall}) =====" + ) + # If all individuals are invalid, stop + if all([not individual["exec_success"] for individual in self.population]): + raise RuntimeError( + "All individuals are invalid. Please check the task evaluation." + ) + # Select + population_to_select = ( + self.population + if (self.elitist is None or self.elitist in self.population) + else [self.elitist] + self.population + ) # add elitist to population for selection + selected_population = self.random_select(population_to_select) + if selected_population is None: + raise RuntimeError("Selection failed. Please check the population.") + logging.info( + f"[Gen {gen}] selection: OK ({len(selected_population)} parents)" + ) + + # Reflection + self.flash_reflection(selected_population) + flash_ok = bool(self.str_flash_memory.get("analyze")) or bool( + self.str_flash_memory.get("exp") + ) + logging.info( + f"[Gen {gen}] flash_reflection: {'OK' if flash_ok else 'EMPTY'} " + f"(analyze={len(self.str_flash_memory.get('analyze', ''))} chars, " + f"exp={len(self.str_flash_memory.get('exp', ''))} chars)" + ) + self.comprehensive_reflection() + comp_ok = bool( + self.str_comprehensive_memory and self.str_comprehensive_memory.strip() + ) + logging.info( + f"[Gen {gen}] comprehensive_reflection: {'OK' if comp_ok else 'EMPTY'} " + f"({len(self.str_comprehensive_memory)} chars)" + ) + curr_code_path = self.elitist["code_path"] + + # Crossover + crossed_population = self.crossover(selected_population) + # Evaluate + self.population = self.evaluate_population(crossed_population) + n_cross_valid = self._valid_count(crossed_population) + logging.info( + f"[Gen {gen}] crossover: {n_cross_valid}/{len(crossed_population)} valid" + ) + self._op_stats["crossover"] += n_cross_valid + if n_cross_valid > 0: + self._op_ok_gens["crossover"].add(gen) + # Update + self.update_iter() + + # Mutate + mutated_population = self.mutate() + # Evaluate + evaluated_mut = self.evaluate_population(mutated_population) + self.population.extend(evaluated_mut) + n_mut_valid = self._valid_count(mutated_population) + logging.info( + f"[Gen {gen}] mutation: {n_mut_valid}/{len(mutated_population)} valid" + ) + self._op_stats["mutation"] += n_mut_valid + if n_mut_valid > 0: + self._op_ok_gens["mutation"].add(gen) + # Update + self.update_iter() + + if curr_code_path != self.elitist["code_path"]: + self.lst_good_reflection.append(self.str_flash_memory["exp"]) + else: + self.lst_bad_reflection.append(self.str_flash_memory["exp"]) + + # Harmony Search + try_hs_num = 3 + individual_hs = None + while try_hs_num: + individual_hs = self.harmony_search() + if individual_hs is not None: + self.population.extend([individual_hs]) + break + else: + try_hs_num -= 1 + if individual_hs is not None: + logging.info( + f"[Gen {gen}] harmony_search: OK (obj={individual_hs.get('obj')})" + ) + self._op_stats["harmony_search"] += 1 + self._op_ok_gens["harmony_search"].add(gen) + else: + logging.info(f"[Gen {gen}] harmony_search: FAILED after 3 tries") + self.update_iter() + + logging.info( + f"===== [Gen {gen}] done (function_evals={self.function_evals}, " + f"best_obj={self.best_obj_overall}) =====" + ) + + # Optional population checkpoint + if self._profiler is not None and isinstance(self._profiler, HSEvoProfiler): + self._profiler.register_population(self.population, self.iteration) + + return self.best_code_overall, self.best_code_path_overall + + # ------------------------------------------------------------------ + # LLM4AD entry point + # ------------------------------------------------------------------ + def run(self): + if not self._resume_mode: + # do initialization (upstream HSEvo runs this inside __init__) + self.init_population() + + # evolutionary search + try: + self.evolve() + except KeyboardInterrupt: + pass + except Exception as e: + logging.info(f"HSEvo evolution terminated: {e}") + if self._debug_mode: + traceback.print_exc() + finally: + # finish + if self._profiler is not None: + self._profiler.finish() + self._llm.close() + try: + self._evaluation_executor.shutdown(cancel_futures=True) + except Exception: + pass diff --git a/llm4ad/method/hsevo/paras.yaml b/llm4ad/method/hsevo/paras.yaml new file mode 100644 index 00000000..7017d381 --- /dev/null +++ b/llm4ad/method/hsevo/paras.yaml @@ -0,0 +1,12 @@ +name: HSEvo +max_sample_nums: 450 +pop_size: 10 +init_pop_size: 30 +mutation_rate: 0.5 +hm_size: 5 +hmcr: 0.7 +par: 0.5 +bandwidth: 0.2 +max_iter: 5 +num_samplers: 4 +num_evaluators: 4 diff --git a/llm4ad/method/hsevo/profiler.py b/llm4ad/method/hsevo/profiler.py new file mode 100644 index 00000000..b933bdc7 --- /dev/null +++ b/llm4ad/method/hsevo/profiler.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import json +import math +import os +from threading import Lock +from typing import List, Dict, Optional + +try: + import wandb +except: + pass + +from ...base import Function +from ...tools.profiler import TensorboardProfiler, ProfilerBase, WandBProfiler + + +def _serialize_population(population: List[Dict]) -> List[Dict]: + funcs_json = [] + for ind in population: + obj = ind.get("obj") + # JSON cannot represent inf cleanly; store None for invalid individuals. + if obj is None or ( + isinstance(obj, float) and (math.isinf(obj) or math.isnan(obj)) + ): + score = None + else: + # LLM4AD score (higher is better) = -obj. + score = -obj + funcs_json.append( + { + "code": ind.get("code"), + "score": score, + "obj": None if score is None else obj, + "exec_success": bool(ind.get("exec_success", False)), + "tryHS": bool(ind.get("tryHS", False)), + } + ) + return funcs_json + + +class HSEvoProfiler(ProfilerBase): + def __init__( + self, + log_dir: Optional[str] = None, + *, + initial_num_samples=0, + log_style="complex", + create_random_path=True, + **kwargs, + ): + """HSEvo Profiler. + Args: + log_dir : the directory of current run + initial_num_samples: the sample order start with `initial_num_samples`. + create_random_path : create a random log_path according to evaluation_name, method_name, time, ... + """ + super().__init__( + log_dir=log_dir, + initial_num_samples=initial_num_samples, + log_style=log_style, + create_random_path=create_random_path, + **kwargs, + ) + self._cur_gen = 0 + self._pop_lock = Lock() + if self._log_dir: + self._ckpt_dir = os.path.join(self._log_dir, "population") + os.makedirs(self._ckpt_dir, exist_ok=True) + + def register_population(self, population: List[Dict], generation: int): + """Checkpoint the current HSEvo population (list of individual dicts).""" + try: + self._pop_lock.acquire() + if not self._log_dir: + return + if self._num_samples == 0 or generation == self._cur_gen: + return + path = os.path.join(self._ckpt_dir, f"pop_{generation}.json") + with open(path, "w") as json_file: + json.dump(_serialize_population(population), json_file, indent=4) + self._cur_gen = generation + finally: + if self._pop_lock.locked(): + self._pop_lock.release() + + def _write_json( + self, function: Function, program="", *, record_type="history", record_sep=200 + ): + """Write function data to a JSON file. + Args: + function : The function object containing score and string representation. + record_type: Type of record, 'history' or 'best'. Defaults to 'history'. + record_sep : Separator for history records. Defaults to 200. + """ + assert record_type in ["history", "best"] + + if not self._log_dir: + return + + sample_order = self._num_samples + content = { + "sample_order": sample_order, + "algorithm": function.algorithm, + "function": str(function), + "score": function.score, + "operator": function.operator, + "program": program, + } + + if record_type == "history": + lower_bound = ((sample_order - 1) // record_sep) * record_sep + upper_bound = lower_bound + record_sep + filename = f"samples_{lower_bound + 1}~{upper_bound}.json" + else: + filename = "samples_best.json" + + path = os.path.join(self._samples_json_dir, filename) + + try: + with open(path, "r") as json_file: + data = json.load(json_file) + except (FileNotFoundError, json.JSONDecodeError): + data = [] + + data.append(content) + + with open(path, "w") as json_file: + json.dump(data, json_file, indent=4) + + +class HSEvoTensorboardProfiler(TensorboardProfiler, HSEvoProfiler): + + def __init__( + self, + log_dir: str | None = None, + *, + initial_num_samples=0, + log_style="complex", + create_random_path=True, + **kwargs, + ): + """Profiler for Tensorboard. + Args: + log_dir : the directory of current run + initial_num_samples: the sample order start with `initial_num_samples`. + create_random_path : create a random log_path according to evaluation_name, method_name, time, ... + **kwargs : kwargs for wandb + """ + HSEvoProfiler.__init__( + self, log_dir=log_dir, create_random_path=create_random_path, **kwargs + ) + TensorboardProfiler.__init__( + self, + log_dir=log_dir, + initial_num_samples=initial_num_samples, + log_style=log_style, + create_random_path=create_random_path, + **kwargs, + ) + + def finish(self): + if self._log_dir: + self._writer.close() + + filename = "end.json" + path = os.path.join(os.path.join(self._log_dir, "population"), filename) + + with open(path, "w") as json_file: + json.dump([], json_file, indent=4) + + +class HSEvoWandbProfiler(WandBProfiler, HSEvoProfiler): + _cur_gen = 0 + + def __init__( + self, + wandb_project_name: str, + log_dir: str | None = None, + *, + initial_num_samples=0, + log_style="complex", + create_random_path=True, + **kwargs, + ): + """Profiler for Wandb. + Args: + wandb_project_name : the name of the wandb project + log_dir : the directory of current run + initial_num_samples: the sample order start with `initial_num_samples`. + create_random_path : create a random log_path according to evaluation_name, method_name, time, ... + **kwargs : kwargs for wandb + """ + HSEvoProfiler.__init__( + self, log_dir=log_dir, create_random_path=create_random_path, **kwargs + ) + WandBProfiler.__init__( + self, + wandb_project_name=wandb_project_name, + log_dir=log_dir, + initial_num_samples=initial_num_samples, + log_style=log_style, + create_random_path=create_random_path, + **kwargs, + ) + self._pop_lock = Lock() + if self._log_dir: + self._ckpt_dir = os.path.join(self._log_dir, "population") + os.makedirs(self._ckpt_dir, exist_ok=True) + + def finish(self): + wandb.finish() + filename = "end.json" + path = os.path.join(os.path.join(self._log_dir, "population"), filename) + + with open(path, "w") as json_file: + json.dump([], json_file, indent=4) diff --git a/llm4ad/method/hsevo/prompt.py b/llm4ad/method/hsevo/prompt.py new file mode 100644 index 00000000..71e1e7aa --- /dev/null +++ b/llm4ad/method/hsevo/prompt.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import copy + +from ...base import Function + +# ---------------------------------------------------------------------------- +# Common prompt templates (verbatim copies of prompts/common/*.txt) +# ---------------------------------------------------------------------------- + +# system_generator.txt +SYSTEM_GENERATOR = """{seed} Your task is to design heuristics that can effectively solve optimization problems. +Your response outputs Python code and nothing else. Format your code as a Python code string: "```python ... ```". +""" + +# user_generator.txt +USER_GENERATOR = """{seed} Your task is to write a {func_name} function for {problem_desc} +{func_desc} +""" + +# seed.txt +SEED = """{seed_func} + +Refer to the format of a trivial design above. Be very creative and give `{func_name}_v2`. Output code only and enclose your code with Python code block: ```python ... ```.""" + +# crossover.txt +CROSSOVER = """{user_generator} + +### Better code +{func_signature_m1} +{code_method1} + +### Worse code +{func_signature_m2} +{code_method2} + +### Analyze & experience +- {analyze} +- {exp} + +Your task is to write an improved function `{func_name}_v2` by COMBINING elements of two above heuristics base Analyze & experience. +Output the code within a Python code block: ```python ... ```, has comment and docstring (<50 words) to description key idea of heuristics design. + +I'm going to tip $999K for a better heuristics! Let's think step by step.""" + +# mutation.txt +MUTATION = """{user_generator} + +Current heuristics: +{func_signature1} +{elitist_code} + +Now, think outside the box write a mutated function `{func_name}_v2` better than current version. +You can use some hints below: +- {reflection} + +Output code only and enclose your code with Python code block: ```python ... ```. +I'm going to tip $999K for a better solution!""" + +# system_reflector.txt +SYSTEM_REFLECTOR = """You are an expert in the domain of optimization heuristics. Your task is to provide useful advice based on analysis to design better heuristics. +""" + +# user_flash_reflection.txt +USER_FLASH_REFLECTION = """### List heuristics +Below is a list of design heuristics ranked from best to worst. +{lst_method} + +### Guide +- Keep in mind, list of design heuristics ranked from best to worst. Meaning the first function in the list is the best and the last function in the list is the worst. +- The response in Markdown style and nothing else has the following structure: +"**Analysis:** +**Experience:**" +In there: ++ Meticulously analyze comments, docstrings and source code of several pairs (Better code - Worse code) in List heuristics to fill values for **Analysis:**. +Example: "Comparing (best) vs (worst), we see ...; (second best) vs (second worst) ...; Comparing (1st) vs (2nd), we see ...; (3rd) vs (4th) ...; Comparing (second worst) vs (worst), we see ...; Overall:" + ++ Self-reflect to extract useful experience for design better heuristics and fill to **Experience:** (<60 words). + +I'm going to tip $999K for a better heuristics! Let's think step by step.""" + +# user_comprehensive_reflection.txt +USER_COMPREHENSIVE_REFLECTION = """Your task is to redefine 'Current self-reflection' paying attention to avoid all things in 'Ineffective self-reflection' in order to come up with ideas to design better heuristics. + +### Current self-reflection +{curr_reflection} +{good_reflection} + +### Ineffective self-reflection +{bad_reflection} + +Response (<100 words) should have 4 bullet points: Keywords, Advice, Avoid, Explanation. +I'm going to tip $999K for a better heuristics! Let's think step by step.""" + +# system_harmony_search.txt +SYSTEM_HARMONY_SEARCH = """You are an expert in code review. Your task extract all threshold, weight or hardcode variable of the function make it become default parameters.""" + +# harmony_search.txt +HARMONY_SEARCH = """[code] +{code_extract} + +Now extract all threshold, weight or hardcode variable of the function make it become default parameters and give me a 'parameter_ranges' dictionary representation. Key of dict is name of variable. Value of key is a tuple in Python MUST include 2 float elements, first element is begin value, second element is end value corresponding with parameter. + +- Output code only and enclose your code with Python code block: ```python ... ```. +- Output 'parameter_ranges' dictionary only and enclose your code with other Python code block: ```python ... ```.""" + + +# ---------------------------------------------------------------------------- +# Scientist personas (verbatim from hsevo.py) used to diversify the initial +# population: each initial individual is generated with a rotating persona. +# ---------------------------------------------------------------------------- +SCIENTISTS = [ + "You are an expert in the domain of optimization heuristics.", + "You are Albert Einstein, relativity theory developer.", + "You are Isaac Newton, the father of physics.", + "You are Marie Curie, pioneer in radioactivity.", + "You are Nikola Tesla, master of electricity.", + "You are Galileo Galilei, champion of heliocentrism.", + "You are Stephen Hawking, black hole theorist.", + "You are Richard Feynman, quantum mechanics genius.", + "You are Rosalind Franklin, DNA structure revealer.", + "You are Ada Lovelace, computer programming pioneer.", +] + + +# ---------------------------------------------------------------------------- +# Problem-specific prompt derivation (replaces HSEvo's per-problem .txt files). +# These build the equivalents of seed_func.txt / func_signature.txt / +# func_desc.txt from the LLM4AD task's template function + task description. +# ---------------------------------------------------------------------------- + + +def make_func_signature(function: Function) -> str: + """Return a signature template with a ``{version}`` placeholder, e.g. + ``def priority_v{version}(item: float, bins: np.ndarray) -> np.ndarray:``. + """ + return_type = f" -> {function.return_type}" if function.return_type else "" + return f"def {function.name}_v{{version}}({function.args}){return_type}:" + + +def make_seed_func(function: Function) -> str: + """Render the template function as ``{name}_v1`` inside a python code block. + + HSEvo's seed_func.txt holds a complete trivial design. We render the LLM4AD + template function (with its docstring + body) as ``_v1`` and wrap it in a + ```python``` fence so that HSEvo's `extract_code_from_generator` extracts the + full body via its regex path (rather than the def->first-return fallback). + """ + f = copy.deepcopy(function) + f.name = f"{function.name}_v1" + return "```python\n" + str(f).rstrip() + "\n```" + + +def make_func_desc(function: Function, task_description: str) -> str: + """Describe the target function's I/O. Prefer the template docstring; fall + back to the task description.""" + if function.docstring: + return function.docstring + return task_description or "" diff --git a/llm4ad/method/hsevo/util.py b/llm4ad/method/hsevo/util.py new file mode 100644 index 00000000..d3fb5849 --- /dev/null +++ b/llm4ad/method/hsevo/util.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import re +from typing import List, Dict + + +def format_messages(system: str, user: str) -> List[Dict[str, str]]: + """Build a chat message list with a system and a user turn. + + This mirrors HSEvo's `format_messages`, but takes the system/user strings + directly (the original took a Hydra `cfg` it never used). LLM4AD's + `HttpsApi`/`OpenAIAPI` backends accept this list directly as the `prompt` + argument of `draw_sample`, which preserves HSEvo's system/user separation. + """ + return [{"role": "system", "content": system}, {"role": "user", "content": user}] + + +def extract_code_from_generator(content): + """Extract code from the response of the code generator (verbatim HSEvo). + + The only deviation from upstream is that the prepended ``scipy``/``torch`` + imports are guarded with ``try/except`` so that environments without those + optional packages (e.g. the default LLM4AD env, which ships without torch) + do not fail to execute every generated heuristic. + """ + pattern_code = r"```python(.*?)```" + code_string = re.search(pattern_code, content, re.DOTALL) + code_string = code_string.group(1).strip() if code_string is not None else None + if code_string is None: + # Find the line that starts with "def" and the line that starts with "return", and extract the code in between + lines = content.split("\n") + start = None + end = None + for i, line in enumerate(lines): + if line.startswith("def"): + start = i + if "return" in line: + end = i + break + if start is not None and end is not None: + code_string = "\n".join(lines[start : end + 1]) + + if code_string is None: + return None + + # --- LLM4AD adaptation ------------------------------------------------- + # Upstream HSEvo prepends: + # "import numpy as np\nimport random\nimport math\nimport scipy\nimport torch\n" + # We keep numpy/random/math unconditional, but guard scipy/torch so a + # missing optional dependency degrades gracefully (only heuristics that + # actually use the missing module fail, instead of the whole search). + global_imports = ( + "import numpy as np\n" + "import random\n" + "import math\n" + "try:\n" + " import scipy\n" + "except Exception:\n" + " pass\n" + "try:\n" + " import torch\n" + "except Exception:\n" + " pass\n" + ) + code_string = global_imports + code_string + return code_string + + +def filter_code(code_string): + """Remove lines containing signature and import statements (verbatim HSEvo).""" + lines = code_string.split("\n") + filtered_lines = [] + for line in lines: + if line.startswith("def"): + continue + elif line.startswith("import"): + continue + elif line.startswith("from"): + continue + elif line.startswith("return"): + filtered_lines.append(line) + break + else: + filtered_lines.append(line) + code_string = "\n".join(filtered_lines) + return code_string + + +def extract_to_hs(input_string: str): + """Parse the harmony-search LLM response (verbatim HSEvo). + + Expects two ```python``` blocks: (1) the parameterised function with + ``{param}`` placeholders in its default values and (2) a ``parameter_ranges`` + dict. Returns ``(parameter_ranges, function_block)`` or ``(None, None)``. + """ + code_blocks = input_string.split("```python\n")[1:] + + try: + parameter_ranges_block = ( + "import numpy as np\n" + code_blocks[1].split("```")[0].strip() + ) + if any( + keyword in parameter_ranges_block for keyword in ["inf", "np.inf", "None"] + ): + return None, None + exec_globals = {} + exec(parameter_ranges_block, exec_globals) + parameter_ranges = exec_globals["parameter_ranges"] + except: + return None, None + + function_block = code_blocks[0].split("```")[0].strip() + + paren_count = 0 + in_signature = False + signature_start_index = None + signature_end_index = None + + # Loop through the function block to find the start and end of the function signature + for i, char in enumerate(function_block): + if char == "d" and function_block[i : i + 3] == "def": + in_signature = True + signature_start_index = i + if in_signature: + if char == "(": + paren_count += 1 + elif char == ")": + paren_count -= 1 + if char == ":" and paren_count == 0: + signature_end_index = i + break + + if signature_start_index is not None and signature_end_index is not None: + function_signature = function_block[ + signature_start_index : signature_end_index + 1 + ] + + # Clean up the function signature from potential default values that might be corrupted (e.g. .eps suffix) + # This regex looks for parameter definitions and cleans any trailing garbage before the next comma or closing paren + function_signature = re.sub( + r"(\w+\s*:\s*\w+\s*=\s*[\d.e-]+)(\.[a-zA-Z]+)", r"\1", function_signature + ) + + for param in parameter_ranges: + pattern = rf"(\b{param}\b[^=]*=)[^,)]+" + replacement = r"\1 {" + param + "}" + function_signature = re.sub( + pattern, replacement, function_signature, flags=re.DOTALL + ) + function_block = ( + function_block[:signature_start_index] + + function_signature + + function_block[signature_end_index + 1 :] + ) + + return parameter_ranges, function_block