python / moduleshillclimb

Your own modules

Write a policy, an operator or a selector and drop it into the climber like a built-in one.

Every slot of a climber takes a class of your own. The base classes come from hillclimb.sdk, which is the only import a module file needs. Then you pass the class to Climber wherever a name would go.

One step of a search is two decisions, always in this order. The selector reads the history and picks the node the next attempt starts from, or none. Then the policy reads the same history and that pick, and names the operator. The selector never names an operator; the policy never picks a node. Each is a small class on its own.

A policy, which picks the operator

from hillclimb import Climber
from hillclimb.operators import Draft, Improve
from hillclimb.sdk import Action, OperatorPolicy, SearchState, Selection, improves
from hillclimb.selectors import Best


class GiveUpQuickly(OperatorPolicy):
    """Build on the chosen node, unless its last few children failed to beat it.
    Then it is stuck: draft somewhere new instead."""

    DEFAULTS = {"patience": 2}  # every knob and its default

    def propose(self, state: SearchState, selection: Selection | None) -> Action | None:
        if selection is None or selection.combine:
            return super().propose(state, selection)   # the plain mapping: draft, ensemble
        node = state.journal.candidates[selection.target_id]
        if node.status != "passing":
            return super().propose(state, selection)   # a failing tip: debug
        children = [c for c in state.journal.children(node.candidate_id) if c.val_score is not None]
        recent = children[-self.param("patience"):]
        stuck = len(recent) >= self.param("patience") and not any(
            improves(c.val_score, node.val_score, higher_is_better=state.higher_is_better, band=state.accept_band)
            for c in recent
        )
        return self.draft_action(state) if stuck else self.expand_action(state, selection)


climber = Climber(selector_policy=Best(num_drafts=4), operator_policy=GiveUpQuickly(patience=2), operators=[Draft(), Improve()])

propose(state, selection) is the whole contract. The selection is what the selector chose: None for a root step, or a Selection with the node's target_id, any inspiration_ids, and combine=True when those nodes are the inputs of one combined candidate. Return the next Action, or None to wait. The state is everything else a policy gets to see: the journal with every candidate so far (holdout scores hidden), what is in flight, how much budget is left, and which direction counts as better.

The base propose already does the plain mapping. No node means draft, a failing node means debug, several nodes mean ensemble, a scored node means improve. So your policy only has to say where it differs, and draft_action, expand_action(state, selection, operator=...) are there to build the usual actions. observe(state, candidate) is called for every result that lands, and the base class uses it to keep the selector in sync.

An operator, which says how an attempt is made

The built-in Draft, Debug, Improve and Ensemble are classes of exactly this shape. Yours goes next to them in the climber's operators=[...].

from hillclimb.sdk import Attempt, Operator, OperatorContext


class Stride(Operator):
    """Move away from the parent by a chosen stride."""

    name = "stride"        # the name an Action uses
    kind = "refine"        # create, repair, refine or combine
    needs_target = True    # it starts from an existing candidate

    def prepare(self, ctx: OperatorContext) -> Attempt:
        step = self.params.get("step", 0.4)   # this operator's own knob, set with Stride(step=...)
        prompt = (
            f"Candidate {ctx.target.candidate_id} scores {ctx.target.val_score}. Move from it.\n"
            f"toy: step={step}\n\n"
            "{{contract}}"   # the harness fills in the problem's contract here
        )
        return Attempt(prompt=prompt, copy_parent=True, inherit_params=True)

An operator never touches the agent, the disk or the journal itself. It returns an Attempt with the prompt and a description of what the harness should place in the candidate dir: the parent's solution (copy_parent), the parent's tuned parameter values as new defaults (inherit_params), the inspirations (copy_inspirations), or extra files. The ctx has the target, the inspirations, the journal, the problem and the budget, all with holdout scores hidden, plus ctx.render(template, ...) for the built-in prompt templates.

A selector, which picks the node

from hillclimb import Climber
from hillclimb.policies import Greedy
from hillclimb.sdk import Selection, SelectorPolicy, improvable


class LeastExpanded(SelectorPolicy):
    """Of the top few scored candidates, the one with the fewest children."""

    name = "least-expanded"
    DEFAULTS = {"top": 4}

    def select(self, state, *, busy=frozenset()) -> Selection | None:
        journal = state.journal
        scored = [c for c in journal.ranked_candidates(state.higher_is_better, "val") if improvable(c)]
        if not scored:
            return None   # a root step: the policy drafts
        top = scored[: self.param("top")]
        chosen = min(top, key=lambda c: len(journal.children(c.candidate_id)))
        return Selection(chosen.candidate_id, prompt_context=f"The least explored of the top {len(top)}.")


climber = Climber(selector_policy=LeastExpanded(top=4), operator_policy=Greedy())

A selector only has to say which scored node to build on, in select(state, busy=...). It returns a Selection with the parent, optional inspirations and a line for the prompt, or None when there is nothing to build on. busy holds the candidates that an attempt in flight is already working on. The base class wraps select with the schedule every selector shares: a failing tip comes first, nothing is built on until num_drafts roots exist, and the final budget window returns the top candidates to combine. Those knobs (num_drafts, debug, max_debug_depth, ensemble, ensemble_top_k, …) belong to the selector, so you write Best(num_drafts=4) or LeastExpanded(top=4, num_drafts=4).

A few things to keep in mind

  • Import the contracts from hillclimb.sdk: OperatorPolicy, Action, SearchState, Operator, Attempt, OperatorContext, SelectorPolicy, Selection, Tuner, Memory, and the records you read, Candidate, Trial and JournalView.
  • A search imports the file your classes live in, so keep the code that starts a search under if __name__ == "__main__":.
  • hillclimb climber check climber.yaml replays recorded journals through a policy and reports broken contracts before you spend any budget. Things like stalling on an empty journal, naming a candidate that is not in the journal, or writing to disk. climber.write("climber.yaml") gives you the file to check.
  • The built-in modules are good reference code. Look at src/hillclimb/modules/policies/greedy.py, modules/operators/builtin.py and modules/selectors/best.py.