"""ALE-Agent — the algorithm-engineering agent introduced with ALE-Bench. This is a paper-specification port: the public ALE-Bench repository contains the benchmark and generic evaluation harness, while the agent's experimental implementation is not included there. The NeurIPS paper and supplement define the mechanisms reproduced by these components. """ from __future__ import annotations from ...config import GalapagosConfig from ...models import GalapagosModel, Prompt from ..base_scaffold import GalapagosScaffold from ..registry import register_scaffold from .memory import ALEAgentMemory from .population import ALEAgentPopulation from .prompt_builder import ALEAgentPromptBuilder from .proposer import ALEAgentProposer from .selection_policy import ALEAgentSelectionPolicy @register_scaffold("ale_agent") class ALEAgentScaffold(GalapagosScaffold): name = "ale_agent" @classmethod def build_components(cls, config: GalapagosConfig, model: GalapagosModel | None) -> dict: selection = config.selection_policy prompt = config.prompt_builder return { "population": ALEAgentPopulation(), "selection_policy": ALEAgentSelectionPolicy( seed=int(config.seed), children_per_parent=int(selection.ale_children_per_parent), refinement_turns=int(selection.ale_refinement_turns), ), "prompt_builder": ALEAgentPromptBuilder( domain_guidance=bool(prompt.ale_domain_guidance), include_history=bool(prompt.ale_include_history), ), "proposer": ALEAgentProposer(), "memory": ALEAgentMemory(max_entries=int(config.memory.ale_history_size)), } def setup(self, task) -> None: super().setup(task) # Galapagos tasks always provide a reproducible initial program. It is the search-tree root; # the bundled ALE-Bench tasks use a minimal valid baseline rather than a hand-optimized seed. if self.ctx.best is not None: self.ctx.best.metadata["ale_frontier"] = True self.ctx.best.metadata["ale_expanded"] = False self.ctx.best.metadata["ale_branch_winner"] = False seed_result = getattr(self, "_seed_eval_result", None) if seed_result is not None: self.ctx.best.metadata["valid"] = bool(seed_result.valid) self._emit_event( event_type="population", action="update_frontier", status="completed", component_role="population", evolution_track="solution", caused_by_event_ids=[ self._candidate_last_event_ids.get(self.ctx.best.id) ], outputs=[ self._event_ref("candidate", self.ctx.best.id, "collection_member") ], details={ "frontier_type": "ale_frontier", "collection_id": "unexpanded_frontier", "added_candidate_ids": [self.ctx.best.id], "removed_candidate_ids": [], "reason": "seed_initialization", }, ) def after_step(self, child, result) -> None: # Admitted evaluator-valid candidates advance in the base loop's SelectionPolicy.observe # call. Unevaluable and evaluator-invalid attempts are not population states, but still spend # one of this branch's implementation turns so the deterministic branch schedule progresses. rejected = result is None or not bool(result.valid) if rejected: policy_state_before = self._selection_policy_state() self.selection_policy.observe_rejected(child, self.ctx) self._emit_policy_update( child, policy_state_before, self._selection_policy_state(), reason="rejected_candidate_observation", ) branch = child.metadata.get("ale_branch_index", "?") turn = child.metadata.get("ale_refinement_turn", "?") if result is None: outcome = "not evaluable" else: outcome = f"score={child.fitness:g}, valid={bool(result.valid)}" change = str(child.metadata.get("changes") or "unspecified change").replace("\n", " ")[:240] memory_entry = f"branch {branch}, turn {turn}: {outcome}; {change}" self.memory.write(memory_entry) self._emit_event( event_type="memory", action="write", status="completed", component_role="memory", evolution_track="solution", caused_by_event_ids=[self._candidate_last_event_ids.get(child.id)], inputs=[self._event_ref("candidate", child.id, "observed_candidate")], details={ "memory_name": "search_history", "entry_type": "branch_result", "content": memory_entry, }, ) # The base controller tracks ctx.best by combined_score alone. ALE-Agent's winner and future # historical-best context use acceptance-before-score priority, so restore that invariant # after every evaluated or rejected turn and before the next selection/checkpoint. priority_best = self.population.best() if priority_best is not None: self.ctx.best = priority_best if child.metadata.get("ale_branch_complete"): winner_id = self.ctx.blackboard.get("ale_agent", {}).get("branch_winner_id") current_frontier_ids = [ genome.id for genome in self.population.query({"frontier": True}) ] self._emit_event( event_type="population", action="update_frontier", status="completed", component_role="selection_policy", evolution_track="solution", inputs=[ self._event_ref( "candidate", child.metadata.get("ale_outer_parent_id"), "expanded_parent" ), self._event_ref("candidate", child.id, "completed_branch_candidate"), ], outputs=[ self._event_ref("candidate", candidate_id, "collection_member") for candidate_id in current_frontier_ids ], details={ "frontier_type": "ale_frontier", "collection_id": "unexpanded_frontier", "branch_index": child.metadata.get("ale_branch_index"), "refinement_turns": child.metadata.get("ale_refinement_turns"), "winner_candidate_id": winner_id, "added_candidate_ids": [winner_id] if winner_id else [], "removed_candidate_ids": [], "reason": "branch_completed", }, ) def _should_stop(self) -> bool: if super()._should_stop(): return True # All K branches can fail before producing a frontier state. Stop cleanly instead of violating # ALE-Agent's tabu rule by re-expanding the exhausted parent. if self._history and not self.selection_policy.can_select(self.population): return True return False def _stop_reason(self) -> str: if self.ctx.iteration >= self.general.max_iterations: return "max_iterations" if self._history and not self.selection_policy.can_select(self.population): return "frontier_exhausted" return super()._stop_reason() def _record(self, genome, result, **kwargs) -> None: """Record the actual implementation prompt and preserve the turn-one strategy call. Turn one performs a strategy-only call followed by an implementation call. The base trajectory has one prompt/response slot per resulting genome, so that slot is assigned to the implementation response; the preceding strategy call is retained in row metadata. """ actual = (genome.artifacts or {}).get("implementation_prompt") if kwargs.get("prompt") is not None and isinstance(actual, dict): kwargs["prompt"] = Prompt( system=str(actual.get("system") or ""), user=str(actual.get("user") or ""), ) strategy_response = (genome.artifacts or {}).get("strategy_response") if strategy_response is not None: metadata = dict(kwargs.get("metadata") or {}) strategy_prompt = (genome.artifacts or {}).get("strategy_prompt") or {} calls = (genome.artifacts or {}).get("model_calls") or [] metadata["ale_agent_strategy_call"] = { "prompt": { "system": str(strategy_prompt.get("system") or ""), "user": str(strategy_prompt.get("user") or ""), }, "response": str(strategy_response), "reasoning": str((genome.artifacts or {}).get("strategy_reasoning") or ""), "call": calls[0] if calls else {}, } kwargs["metadata"] = metadata super()._record(genome, result, **kwargs)