The Kapso orchestrator: running the experiment loop

from kapso import Kapso

kapso = Kapso()
solution = kapso.evolve(
    goal="Build a classifier with accuracy > 0.9",
    max_iterations=10,
)

# Check if goal was achieved
if solution.succeeded:
    print(f"Goal achieved with score: {solution.final_score}")

from kapso.execution.orchestrator import OrchestratorAgent
from kapso.environment.handlers.generic import GenericProblemHandler

# Create problem handler
handler = GenericProblemHandler(
    problem_description="Build a classifier with accuracy > 0.9",
)

# Create orchestrator — config_path=None uses the packaged config
orchestrator = OrchestratorAgent(
    problem_handler=handler,
    mode="GENERIC",
    coding_agent="claude_code",
    goal="Build a classifier with accuracy > 0.9",
)

# Run experimentation — returns a SolveResult
result = orchestrator.solve(
    experiment_max_iter=10,
    time_budget_minutes=60,
    cost_budget=50.0,
)

print(result.stopped_reason)   # "goal_achieved" | "max_iterations" | "budget_exhausted" | "legacy_stop" | "waiting_for_user"
best = result.best_experiment

# Faithful sketch of OrchestratorAgent.solve()
budget = BudgetSpec.resolve(config_block, time_budget_minutes, cost_budget, ...)

for i in range(experiment_max_iter):
    budget_progress = budget.progress(elapsed_seconds, i, ledger.total_cost())
    if budget_progress >= 100:
        break                     # checkpoint stays status="running", last_stop says why

    # Context is the handler's problem statement plus the previous
    # iteration's feedback; experiment history reaches the agent via
    # MCP tools, not the prompt.
    context = problem
    if self.current_feedback:
        context += f"\n\n## Feedback from Previous Iteration\n\n{self.current_feedback}"

    # One search iteration: ideation, implementation, judged feedback
    node = self.search_strategy.run(context, budget_progress)

    # A stop vote ends the campaign only when the handler's task semantics
    # allow it — open-ended maximization has no "goal achieved"
    if node.should_stop and self.problem_handler.honor_agent_stop:
        break

    self._save_run_checkpoint(...)  # atomically, at the iteration boundary

return SolveResult(best_experiment=self.search_strategy.get_deliverable_experiment(), ...)

budget_progress = max(time_progress, iteration_progress, cost_progress) * 100

def _create_search_strategy(self, coding_agent, workspace_dir):
    # Extract config
    search_config = self.mode_config.get('search_strategy', {})
    strategy_type = search_config.get('type', 'generic')
    strategy_params = search_config.get('params', {})

    # Build coding agent config
    coding_agent_config = CodingAgentFactory.build_config(
        agent_type=coding_agent_type,
        model=coding_agent_model,
    )

    # Create via factory
    return SearchStrategyFactory.create(
        strategy_type=strategy_type,
        problem_handler=self.problem_handler,
        llm=self.llm,
        coding_agent_config=coding_agent_config,
        params=strategy_params,
    )

def _create_knowledge_search(self, is_kg_active):
    ks_config = self.mode_config.get('knowledge_search', {})

    if ks_config:
        return KnowledgeSearchFactory.create_from_config(ks_config)

    if is_kg_active:
        return KnowledgeSearchFactory.create("kg_llm_navigation")

    return KnowledgeSearchFactory.create_null()

try:
    # Solve loop
    ...
finally:
    # Close knowledge search (only if the orchestrator created it)
    if self._owns_knowledge_search:
        self.knowledge_search.close()

modes:
  GENERIC:
    search_strategy:
      type: "generic"
      params:
        idea_generation_model: "claude-opus-5"
        implementation_model: "claude-opus-5"
        auth_mode: "oauth"
        gate_failure_policy: "warn"
        ideation_gates: ["research", "experiment_history", "repo_memory", "leeroopedia"]

    retry:
      max_attempts: 2
      initial_delay_seconds: 5
      max_delay_seconds: 60
      multiplier: 2
      jitter: true

    coding_agent:
      type: "claude_code"
      model: "claude-opus-5"
      agent_specific:
        auth_mode: "oauth"

    knowledge_search:
      type: "kg_graph_search"
      enabled: true
      preset: "DEFAULT"