Customizing Beacon#

Replace Beacon’s default Bayesian optimizer by overriding _create_optimizer. Beacon retains AI sampling, fallback, expert insights, journaling, runtime controls, and metadata persistence. The inner optimizer implements the Optimizers contract.

Integration Rules#

  • Set custom constructor attributes before calling super().__init__(). Beacon invokes _create_optimizer(inputs, outputs, constraints) during initialization.

  • The inner sample and report methods must be synchronous. Beacon runs them in a worker thread.

  • Accept every valid measured point in report, including points proposed by AI.

  • Handle batch sampling and reports that arrive out of order. Completed results are replayed on resume.

  • Keep custom parameter names distinct from Beacon settings. Declare UI fields on the Beacon subclass and implement runtime getters and setters on the inner optimizer.

A Complete Grid-Search Example#

This example searches the multiplication protocol’s discrete inputs and minimizes score_multiplication.loss, computed as abs(product - 1024). It uses no Bayesian acquisition function. The grid contains 36 combinations, with a minimum loss of 0. One optimum is number=4 with both factors equal to 16.

The implementation supports discrete inputs, one minimization objective, and no constraints. It tracks pending and completed points to avoid repeating them. Exhausting the grid raises an error, so limit a grid-only campaign to at most 36 runs.

Download the complete example and save it as user/example/protocols/optimize_multiplication/optimizer.py. Back up the existing file first, then reload the protocol before submitting a new campaign.

Grid Optimizer#

The algorithm records all reported results and returns the best observed rows. It reconstructs its completed-point set when EOS replays results after a resume.

class GridSearchOptimizer(AbstractSequentialOptimizer):
    def __init__(self, inputs, outputs, constraints, descending=False):
        if constraints or not all(isinstance(feature, DiscreteInput) for feature in inputs):
            raise ValueError("This example supports discrete inputs without constraints")
        if len(outputs) != 1 or not isinstance(outputs[0].objective, MinimizeObjective):
            raise ValueError("This example requires one minimization objective")

        self._input_names = [feature.key for feature in inputs]
        self._output_names = [feature.key for feature in outputs]
        self._grid = list(product(*(feature.values for feature in inputs)))
        self._pending = set()
        self._completed = set()
        self._results = []
        self.set_runtime_params({"descending": descending})

    def sample(self, num_protocol_runs=1):
        available = [
            point
            for point in sorted(self._grid, reverse=self._descending)
            if point not in self._completed and point not in self._pending
        ]
        selected = available[:num_protocol_runs]
        if len(selected) != num_protocol_runs:
            raise ValueError("Grid exhausted")

        self._pending.update(selected)
        return pd.DataFrame(selected, columns=self._input_names)

    def report(self, inputs_df, outputs_df):
        rows = pd.concat([inputs_df.reset_index(drop=True), outputs_df.reset_index(drop=True)], axis=1)
        self._results.extend(rows.to_dict(orient="records"))

        for point in inputs_df[self._input_names].itertuples(index=False, name=None):
            self._completed.add(point)
            self._pending.discard(point)

    def get_optimal_solutions(self):
        results = pd.DataFrame(self._results, columns=self._input_names + self._output_names)
        if results.empty:
            return results

        objective = results[self._output_names[0]]
        return results.loc[objective == objective.min()].copy()

    def get_input_names(self):
        return self._input_names

    def get_output_names(self):
        return self._output_names

    def get_num_samples_reported(self):
        return len(self._results)

    def get_runtime_params(self):
        return {"descending": self._descending}

    def set_runtime_params(self, params):
        if "descending" in params:
            if not isinstance(params["descending"], bool):
                raise ValueError("descending must be a boolean")
            self._descending = params["descending"]

Beacon Subclass#

descending changes the grid traversal order. Its schema creates a checkbox, and the inner optimizer’s runtime methods let Beacon forward and persist changes.

class GridBeacon(BeaconOptimizer):
    def __init__(self, descending=False, **kwargs):
        self._descending = descending
        super().__init__(**kwargs)

    def _create_optimizer(self, inputs, outputs, constraints):
        return GridSearchOptimizer(inputs, outputs, constraints, descending=self._descending)

    @classmethod
    def eos_param_schema(cls):
        return [{"key": "descending", "type": "checkbox", "default": False, "runtime": True}]

Factory#

The factory defaults to grid-only sampling so the example runs without an AI provider. To enable AI, configure a provider and use probabilities such as p_bayesian=0.5 and p_ai=0.5. Despite its name, p_bayesian selects the custom grid optimizer.

def eos_create_campaign_optimizer():
    return {
        "inputs": [
            DiscreteInput(key="mult_1.number", values=[4, 8, 16, 32]),
            DiscreteInput(key="mult_1.factor", values=[4, 8, 16]),
            DiscreteInput(key="mult_2.factor", values=[4, 8, 16]),
        ],
        "outputs": [
            ContinuousOutput(key="score_multiplication.loss", objective=MinimizeObjective()),
        ],
        "constraints": [],
        "p_bayesian": 1.0,
        "p_ai": 0.0,
        "ai_additional_parameters": ["mult_2.product"],
    }, GridBeacon

Run and Tune a Campaign#

Load the multiplication lab and protocol, then submit a campaign through the web UI or REST API:

curl -X POST http://localhost:8070/api/campaigns \
  -H "Content-Type: application/json" \
  -d '{
    "name": "custom_grid",
    "protocol": "optimize_multiplication",
    "owner": "example",
    "max_protocol_runs": 12,
    "optimize": true,
    "meta": {"optimizer_overrides": {"descending": true}}
  }'

Change traversal order while it runs:

curl -X PUT http://localhost:8070/api/campaigns/custom_grid/optimizer/params \
  -H "Content-Type: application/json" \
  -d '{"descending": false}'

For the full parameter schema, see Parameter Schema and Runtime Controls. API requests need a bearer token when Authentication & Authorization is enabled.