darwinian-evolver

1"""2Template: a custom darwinian-evolver problem.3 4Copy this file, fill in the THREE marked spots (Organism, Evaluator, Mutator),5then run it as a driver script. The skeleton handles all the wiring so you only6write the domain-specific logic.7 8To run:9    cd ~/.hermes/cache/darwinian-evolver/darwinian_evolver10    OPENROUTER_API_KEY=... uv run --with openai python /path/to/this_file.py \11        --num_iterations 3 --num_parents_per_iteration 2 \12        --output_dir /tmp/my_problem13 14The pattern mirrors `scripts/parrot_openrouter.py` (the working reference).15"""16from __future__ import annotations17 18import argparse19import os20import sys21from pathlib import Path22 23from openai import OpenAI24 25# Upstream types (AGPL — invoked via subprocess in production; importing here26# is fine for skill-side driver scripts the user owns).27from darwinian_evolver.cli_common import (28    build_hyperparameter_config_from_args,29    parse_learning_log_view_type,30    register_hyperparameter_args,31)32from darwinian_evolver.evolve_problem_loop import EvolveProblemLoop33from darwinian_evolver.learning_log import LearningLogEntry34from darwinian_evolver.problem import (35    EvaluationFailureCase,36    EvaluationResult,37    Evaluator,38    Mutator,39    Organism,40    Problem,41)42 43DEFAULT_MODEL = os.environ.get("EVOLVER_MODEL", "openai/gpt-4o-mini")44 45 46def _client() -> OpenAI:47    key = os.environ.get("OPENROUTER_API_KEY")48    if not key:49        sys.exit("OPENROUTER_API_KEY is not set")50    return OpenAI(api_key=key, base_url="https://openrouter.ai/api/v1")51 52 53def _prompt_llm(prompt: str, max_tokens: int = 1024) -> str:54    try:55        r = _client().chat.completions.create(56            model=DEFAULT_MODEL,57            max_tokens=max_tokens,58            messages=[{"role": "user", "content": prompt}],59        )60        return r.choices[0].message.content or ""61    except Exception as e:62        # Never let one bad LLM response kill the run.63        return f"<LLM_ERROR: {type(e).__name__}: {e}>"64 65 66# ---------------------------------------------------------------------------67# 1. ORGANISM — what you are evolving.68# ---------------------------------------------------------------------------69class MyOrganism(Organism):70    # TODO: replace with your artifact field. Common shapes:71    #   prompt_template: str72    #   regex_pattern: str73    #   sql_query: str74    #   code_block: str75    artifact: str76 77    def run(self, *inputs) -> str:78        """Exercise the organism on a test input. Return whatever your79        evaluator wants to score."""80        # TODO: implement. For prompt evolution this typically calls _prompt_llm81        # with the artifact rendered against the input. For regex/SQL it would82        # call `re.findall(self.artifact, input)` / execute SQL / etc.83        raise NotImplementedError84 85 86# ---------------------------------------------------------------------------87# 2. EVALUATOR — score organisms and surface failures the mutator can learn from.88# ---------------------------------------------------------------------------89class MyFailureCase(EvaluationFailureCase):90    # TODO: include enough context for the LLM to diagnose the failure.91    input: str92    expected: str93    actual: str94 95 96class MyEvaluator(Evaluator[MyOrganism, EvaluationResult, MyFailureCase]):97    # Split your dataset. Mutator only sees trainable; holdout detects overfitting.98    TRAINABLE = [99        # TODO: list of (input, expected) tuples100        # ("input1", "expected1"),101    ]102    HOLDOUT = [103        # TODO: separate set the mutator never sees104    ]105 106    def evaluate(self, organism: MyOrganism) -> EvaluationResult:107        train_fails: list[MyFailureCase] = []108        hold_fails: list[MyFailureCase] = []109        for i, (inp, expected) in enumerate(self.TRAINABLE):110            actual = organism.run(inp)111            if actual != expected:112                train_fails.append(MyFailureCase(113                    input=inp, expected=expected, actual=actual,114                    data_point_id=f"trainable_{i}",115                ))116        for i, (inp, expected) in enumerate(self.HOLDOUT):117            actual = organism.run(inp)118            if actual != expected:119                hold_fails.append(MyFailureCase(120                    input=inp, expected=expected, actual=actual,121                    data_point_id=f"holdout_{i}",122                ))123        n_total = len(self.TRAINABLE) + len(self.HOLDOUT)124        n_ok = n_total - len(train_fails) - len(hold_fails)125        return EvaluationResult(126            score=n_ok / n_total if n_total else 0.0,127            trainable_failure_cases=train_fails,128            holdout_failure_cases=hold_fails,129            # Always-viable. The evolver only blocks completely-broken organisms;130            # a 0-score organism is fine and will simply be sampled less often.131            is_viable=True,132        )133 134 135# ---------------------------------------------------------------------------136# 3. MUTATOR — LLM proposes an improved organism from a failure case.137# ---------------------------------------------------------------------------138class MyMutator(Mutator[MyOrganism, MyFailureCase]):139    PROMPT = """140The current artifact is:141```142{artifact}143```144 145On this input:146```147{input}148```149it produced:150```151{actual}152```153but we wanted:154```155{expected}156```157 158Diagnose what went wrong, then propose an improved version of the artifact.159Put the new version in the LAST triple-backtick block of your response.160""".strip()161 162    def mutate(163        self,164        organism: MyOrganism,165        failure_cases: list[MyFailureCase],166        learning_log_entries: list[LearningLogEntry],167    ) -> list[MyOrganism]:168        fc = failure_cases[0]169        prompt = self.PROMPT.format(170            artifact=organism.artifact,171            input=fc.input,172            actual=fc.actual,173            expected=fc.expected,174        )175        resp = _prompt_llm(prompt)176        parts = resp.split("```")177        if len(parts) < 3:178            return []179        new_artifact = parts[-2].strip()180        # Strip an opening language tag like "python\n" or "sql\n"181        if "\n" in new_artifact:182            first_line, rest = new_artifact.split("\n", 1)183            if first_line and not first_line.startswith(" ") and len(first_line) < 20:184                new_artifact = rest185        return [MyOrganism(artifact=new_artifact)]186 187 188# ---------------------------------------------------------------------------189# Driver — fills in the EvolveProblemLoop boilerplate. You shouldn't need to190# touch anything below this line for a typical run.191# ---------------------------------------------------------------------------192def make_problem() -> Problem:193    initial = MyOrganism(artifact="TODO: starting artifact here")  # TODO194    return Problem[MyOrganism, EvaluationResult, MyFailureCase](195        evaluator=MyEvaluator(),196        mutators=[MyMutator()],197        initial_organism=initial,198    )199 200 201def main() -> int:202    ap = argparse.ArgumentParser()203    register_hyperparameter_args(ap.add_argument_group("hyperparameters"))204    ap.add_argument("--num_iterations", type=int, default=3)205    ap.add_argument("--mutator_concurrency", type=int, default=2)206    ap.add_argument("--evaluator_concurrency", type=int, default=2)207    ap.add_argument("--output_dir", type=str, required=True)208    args = ap.parse_args()209 210    out = Path(args.output_dir)211    out.mkdir(parents=True, exist_ok=True)212    (out / "snapshots").mkdir(exist_ok=True)213 214    hp = build_hyperparameter_config_from_args(args)215    loop = EvolveProblemLoop(216        problem=make_problem(),217        learning_log_view_type=parse_learning_log_view_type(hp.learning_log_view_type),218        num_parents_per_iteration=hp.num_parents_per_iteration,219        mutator_concurrency=args.mutator_concurrency,220        evaluator_concurrency=args.evaluator_concurrency,221        fixed_midpoint_score=hp.fixed_midpoint_score,222        midpoint_score_percentile=hp.midpoint_score_percentile,223        sharpness=hp.sharpness,224        novelty_weight=hp.novelty_weight,225        batch_size=hp.batch_size,226        should_verify_mutations=hp.verify_mutations,227    )228 229    print("Evaluating initial organism...")230    for snap in loop.run(num_iterations=args.num_iterations):231        (out / "snapshots" / f"iteration_{snap.iteration}.pkl").write_bytes(snap.snapshot)232        _, best = snap.best_organism_result233        print(f"iter={snap.iteration} pop={snap.population_size} best_score={best.score:.3f}")234 235    print(f"\nDone. Results in: {out}")236    return 0237 238 239if __name__ == "__main__":240    sys.exit(main())241