> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/skydiscover-ai/skydiscover/llms.txt
> Use this file to discover all available pages before exploring further.

# Custom Search Algorithms

> Implement your own search strategies by extending ProgramDatabase and DiscoveryController

SkyDiscover's architecture lets you plug in custom search algorithms without rewriting the generate-evaluate loop. You only implement what changes—everything else is inherited.

## Two Levels of Customization

There are two levels depending on what you need:

<CardGroup cols={2}>
  <Card title="Database Only" icon="database">
    Customize `add()` and `sample()` for different parent selection or storage logic
  </Card>

  <Card title="Database + Controller" icon="gears">
    Override `run_discovery()` for cross-iteration behavior like stagnation response or acceptance gating
  </Card>
</CardGroup>

## Level 1: Database Only

Subclass `ProgramDatabase` and implement two abstract methods. The default controller runs the loop unchanged.

### Complete Implementation Example

Here's the full implementation of TopK search (56 lines):

```python skydiscover/search/topk/database.py theme={null}
import logging
from typing import List, Optional, Tuple

from skydiscover.config import DatabaseConfig
from skydiscover.search.base_database import Program, ProgramDatabase

logger = logging.getLogger(__name__)

class TopKDatabase(ProgramDatabase):
    """Database for top-k programs"""

    def __init__(self, name: str, config: DatabaseConfig):
        super().__init__(name, config)
        self.initial_program = None

    def add(self, program: Program, iteration: Optional[int] = None, **kwargs) -> str:
        """Add a program to the database."""
        # Store the initial program
        if iteration == 0 or program.iteration_found == 0:
            self.initial_program = program

        # Store the program
        self.programs[program.id] = program

        # Track last iteration
        if iteration is not None:
            self.last_iteration = max(self.last_iteration, iteration)

        # Save to disk if configured
        if self.config.db_path:
            self._save_program(program)

        # Update best program tracking (required)
        self._update_best_program(program)

        logger.debug(f"Added program {program.id} to top-k database")
        return program.id

    def sample(
        self, num_context_programs: Optional[int] = 4, **kwargs
    ) -> Tuple[Program, List[Program]]:
        """
        Sample a program and context programs for the next discovery step.

        Top-K sampling strategy:
        - Parent: Top 1 program (best program)
        - Context programs: Next K programs (ranks 2 to K+1)
        """
        if not self.programs:
            raise ValueError("Cannot sample: no programs in database")

        # Get top (K+1) programs: top 1 for parent, next K for context
        total_needed = num_context_programs + 1
        top_programs = self.get_top_programs(total_needed)

        if len(top_programs) < 2:
            parent = top_programs[0]
            context_programs = [top_programs[0]]
        else:
            parent = top_programs[0]
            context_programs = top_programs[1:min(len(top_programs), num_context_programs + 1)]

        return parent, context_programs
```

### Key Points

<AccordionGroup>
  <Accordion title="What add() must do">
    1. Store the program: `self.programs[program.id] = program`
    2. Update iteration tracking: `self.last_iteration = max(self.last_iteration, iteration)`
    3. Persist to disk: `self._save_program(program)` (if `config.db_path` is set)
    4. **Call `self._update_best_program(program)`** — required for tracking global best
  </Accordion>

  <Accordion title="What sample() must return">
    Returns `(parent, context_programs)` where:

    * `parent` is a single `Program` to mutate
    * `context_programs` is a `List[Program]` shown as examples to the LLM

    Optionally, you can return dict-wrapped results to add metadata:

    * `({"island_3": parent}, {"top_performers": context_programs})`
  </Accordion>

  <Accordion title="Inherited helper methods">
    * `self.programs` — `dict[str, Program]` storing all programs
    * `self._update_best_program(program)` — updates global best (call in `add()`)
    * `self._save_program(program)` — persists to disk
    * `self.get_top_programs(n)` — returns top N by score
    * `self.get_best_program()` — returns highest-scoring program
  </Accordion>
</AccordionGroup>

### Program Dataclass

Every program has these fields:

```python theme={null}
@dataclass
class Program:
    # Identification
    id: str                          # UUID
    solution: str                    # Source code or prompt text
    language: str = "python"

    # Performance
    metrics: Dict[str, Any]          # Evaluation results (includes combined_score)

    # Lineage
    parent_id: Optional[str] = None  # ID of parent program
    iteration_found: int = 0         # Iteration that produced this

    # Metadata
    metadata: Dict[str, Any]         # Arbitrary extra data
    artifacts: Dict[str, Any]        # Evaluation artifacts
    timestamp: float
```

### Registration

Add your algorithm to `skydiscover/search/route.py`:

```python route.py theme={null}
from skydiscover.search.my_algo.database import MyDatabase
from skydiscover.search.registry import register_database

register_database("my_algo", MyDatabase)
```

That's all. Now `--search my_algo` works:

```bash theme={null}
skydiscover-run initial.py evaluator.py --search my_algo -i 100
```

<Note>
  **Simple algorithms at this level**: `topk/` (56 lines), `best_of_n/` (85 lines), `beam_search/` (527 lines)
</Note>

***

## Level 2: Database + Controller

Use this when you need behavior that spans across iterations:

* Tracking improvement history
* Reacting to stagnation
* Filtering results before they enter the population
* Multi-island search with migration
* Acceptance gating

The key point: **you do not rewrite generate-evaluate logic**. You call `_run_iteration()`, which runs the full `sample → prompt → LLM → evaluate` cycle, then decide what to do with the result.

### Controller Template

```python theme={null}
from skydiscover.search.default_discovery_controller import (
    DiscoveryController,
    DiscoveryControllerInput
)

class MyController(DiscoveryController):

    def __init__(self, controller_input: DiscoveryControllerInput):
        super().__init__(controller_input)
        # Custom initialization here
        self.stagnation_count = 0

    async def run_discovery(
        self,
        start_iteration: int,
        max_iterations: int,
        checkpoint_callback=None,
        post_process_result=True,
        retry_times=3,
    ):
        """
        Custom discovery loop with cross-iteration logic.
        """
        for iteration in range(start_iteration, start_iteration + max_iterations):
            if self.shutdown_event.is_set():
                break

            # Run full generate-evaluate cycle
            result = await self._run_iteration(iteration, retry_times=retry_times)

            if result.error:
                continue

            # Optional: filter before storing
            if self._should_accept(result):
                self._process_iteration_result(
                    result,
                    iteration,
                    checkpoint_callback
                )

            # Cross-iteration logic
            if self._is_stagnating():
                self._trigger_diversification()

        return self.database.get_best_program()

    def _should_accept(self, result):
        """Custom acceptance criterion."""
        child = result.child_program_dict
        parent = self.database.get(result.parent_id)
        return child["metrics"]["combined_score"] > parent.metrics["combined_score"]

    def _is_stagnating(self):
        """Detect stagnation."""
        return self.stagnation_count > 10

    def _trigger_diversification(self):
        """Custom diversification strategy."""
        self.stagnation_count = 0
        # ... your logic here
```

### Controller Primitives

<ResponseField name="_run_iteration(iteration)" type="async">
  Runs the full `sample → prompt → LLM → evaluate` cycle for one iteration. Returns `SerializableResult`.
</ResponseField>

<ResponseField name="_process_iteration_result(result, iteration, cb)" type="method">
  Stores result to database, logs metrics, and triggers checkpoint callback.
</ResponseField>

<ResponseField name="database.get_best_program()" type="method">
  Returns the best program seen so far.
</ResponseField>

<ResponseField name="shutdown_event.is_set()" type="property">
  Returns `True` when graceful shutdown is requested.
</ResponseField>

### SerializableResult Fields

```python theme={null}
@dataclass
class SerializableResult:
    error: Optional[str] = None                # Error message if failed
    child_program_dict: Optional[dict] = None  # Generated program as dict
    parent_id: Optional[str] = None            # Parent program ID
    other_context_ids: List[str] = []          # Context program IDs
    iteration_time: float = 0.0                # Execution time
    prompt: Optional[dict] = None              # System and user prompts
    llm_response: Optional[str] = None         # Raw LLM output
    iteration: int = 0                         # Iteration number
```

### Registration

Register both database and controller:

```python route.py theme={null}
from skydiscover.search.my_algo.database import MyDatabase
from skydiscover.search.my_algo.controller import MyController
from skydiscover.search.registry import register_database, register_controller

register_database("my_algo", MyDatabase)
register_controller("my_algo", MyController)
```

<Note>
  **Complex algorithms at this level**: `adaevolve/` (multi-island UCB search), `gepa_native/` (acceptance gating + merge), `evox/` (co-evolves the search algorithm itself)
</Note>

***

## Custom Configuration

If your algorithm has custom settings, add a dataclass in `skydiscover/config.py`:

```python config.py theme={null}
from dataclasses import dataclass

@dataclass
class MyDatabaseConfig(DatabaseConfig):
    my_param: float = 1.0
    population_size: int = 100
```

Add it to the config type mapping:

```python config.py theme={null}
_DB_CONFIG_BY_TYPE = {
    "topk": DatabaseConfig,
    "beam_search": DatabaseConfig,
    "my_algo": MyDatabaseConfig,  # <-- Add this
}
```

Now users can configure it in `config.yaml`:

```yaml config.yaml theme={null}
search:
  type: my_algo
  database:
    my_param: 2.0
    population_size: 200
```

Access in your database:

```python theme={null}
def __init__(self, name: str, config: MyDatabaseConfig):
    super().__init__(name, config)
    self.my_param = config.my_param
    self.population_size = config.population_size
```

***

## Complete Reference

### ProgramDatabase API

| Method                          | Abstract? | Purpose                              |
| ------------------------------- | --------- | ------------------------------------ |
| `add(program, iteration)`       | ✅ Yes     | Store a scored program               |
| `sample(num_context_programs)`  | ✅ Yes     | Select parent and context            |
| `save(path, iteration)`         | No        | Checkpoint to disk                   |
| `load(path)`                    | No        | Restore from checkpoint              |
| `_update_best_program(program)` | No        | Track best program (call from `add`) |
| `get_best_program()`            | No        | Return highest-scoring program       |
| `get_top_programs(n)`           | No        | Return top N by score                |
| `get(program_id)`               | No        | Retrieve by ID                       |
| `log_status()`                  | No        | Log database summary                 |
| `get_statistics()`              | No        | Return stats dict for prompt context |

### DiscoveryController API

| Method                           | Abstract? | Purpose                        |
| -------------------------------- | --------- | ------------------------------ |
| `run_discovery(...)`             | ✅ Yes     | Main discovery loop            |
| `_run_iteration(iteration)`      | No        | Single generate-evaluate cycle |
| `_process_iteration_result(...)` | No        | Store and log result           |
| `request_shutdown()`             | No        | Trigger graceful shutdown      |

### Complete Registration Example

```python route.py theme={null}
# Level 1: Database only (uses default DiscoveryController)
register_database("topk", TopKDatabase)
register_database("best_of_n", BestOfNDatabase)
register_database("beam_search", BeamSearchDatabase)

# Level 2: Database + Controller
register_database("adaevolve", AdaEvolveDatabase)
register_controller("adaevolve", AdaEvolveController)

register_database("gepa_native", GEPANativeDatabase)
register_controller("gepa_native", GEPANativeController)

# Level 2: Controller + dynamic database
register_controller("evox", CoEvolutionController)
register_database("evox_meta", SearchStrategyDatabase)
```

***

## Next Steps

<CardGroup cols={2}>
  <Card title="Custom Benchmarks" icon="flask" href="/extending/custom-benchmarks">
    Add your own optimization tasks
  </Card>

  <Card title="Context Builders" icon="message" href="/extending/context-builders">
    Customize prompt generation
  </Card>
</CardGroup>