Search Strategies

search_strategy:
  type: "generic"
  params:
    idea_generation_model: "us.anthropic.claude-opus-4-5-20251101-v1:0"
    implementation_model: "us.anthropic.claude-opus-4-5-20251101-v1:0"
    auth_mode: "bedrock"
    parent_policy: "best"
    aws_region: "us-east-1"
    gate_failure_policy: "warn"
    ideation_timeout: 300
    implementation_timeout: 600
    ideation_gates: ["research", "experiment_history", "repo_memory", "leeroopedia"]
    implementation_gates: ["research", "repo_memory", "leeroopedia"]

from kapso.execution.search_strategies import SearchStrategyFactory

strategy = SearchStrategyFactory.create(
    strategy_type="generic",
    problem_handler=handler,
    llm=llm,
    coding_agent_config=config,
    params={"ideation_timeout": 300},
)

strategy.run(context, budget_progress=0.0)
best = strategy.get_best_experiment()

def run(self, context, budget_progress):
    # Prune after initial exploration
    if budget_progress >= 20:
        self.prune_bad_solutions(context)

    # Expand nodes with new solutions
    self.expand(context, budget_progress)

    # Select best nodes to experiment
    best_nodes = self.select(
        context,
        top_k=experiments_count,
        exclude_experimented_nodes=True
    )

    # Run experiments in parallel
    with ThreadPoolExecutor() as executor:
        for node in best_nodes:
            executor.submit(self._run_for_node, node, context, branch_name)

search_strategy:
  type: "benchmark_tree_search"
  params:
    reasoning_effort: "medium"
    code_debug_tries: 5
    node_expansion_limit: 2
    node_expansion_new_childs_count: 5
    idea_generation_steps: 1
    first_experiment_factor: 1
    experimentation_per_run: 1
    per_step_maximum_solution_count: 10
    exploration_budget_percent: 30
    idea_generation_model: "gpt-4o-mini"

@dataclass
class SearchNode:
    node_id: int
    parent_node_id: Optional[int] = None
    
    # Step 1: Solution generation
    solution: str = ""
    
    # Step 2: Implementation
    branch_name: str = ""
    code_changes_summary: str = ""
    
    # Step 3: Evaluation
    evaluation_script_path: str = ""
    evaluation_output: str = ""
    
    # Step 4: Feedback
    feedback: str = ""
    score: Optional[float] = None
    should_stop: bool = False
    evaluation_valid: bool = True
    
    # Metadata
    had_error: bool = False
    error_message: str = ""
    workspace_dir: str = ""

from kapso.execution.search_strategies.base import SearchStrategy
from kapso.execution.search_strategies.factory import register_strategy

@register_strategy("my_custom_search")
class MyCustomSearch(SearchStrategy):
    def __init__(self, config, workspace_dir=None):
        super().__init__(config, workspace_dir)
        # Custom initialization

    def run(self, context, budget_progress=0.0):
        # Generate and run experiments
        solution = self.generate_solution(context)
        result = self._implement(
            solution, context,
            branch_name="experiment_0",
        )
        self.node_history.append(result)

    def get_experiment_history(self, best_last=False):
        if best_last:
            return sorted(self.node_history, key=lambda x: x.score)
        return self.node_history

    def get_best_experiment(self):
        valid = [e for e in self.node_history if not e.had_error]
        return max(valid, key=lambda x: x.score) if valid else None

    def checkout_to_best_experiment_branch(self):
        best = self.get_best_experiment()
        if best:
            self.workspace.switch_branch(best.branch_name)