Active Learning

Showcase of active learning in bofire. Active learning per definition focusses on fitting the model to the experimental observations best possible in an iterative manner reducing some kind of uncertainty. The ActiveLearningStrategy proposes a set of evaluation points that will gain the most information about the problem each iteration. Thus, an unknown black-box-function can be approximated without optimization. It represents an exploration-only strategy.

import os

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from matplotlib import cm

import bofire.strategies.api as strategies
from bofire.benchmarks.api import GenericBenchmark
from bofire.benchmarks.single import Himmelblau
from bofire.data_models.api import Domain, Inputs, Outputs
from bofire.data_models.features.api import ContinuousInput, ContinuousOutput
from bofire.data_models.objectives.api import MinimizeObjective
from bofire.data_models.strategies.api import RandomStrategy
from bofire.runners.api import run


SMOKE_TEST = os.environ.get("SMOKE_TEST")

1-D Objective Function

For a 1-D objective function. The Himmelblau benchmark is used. \[\begin{equation} f: \mathbb{R}^2 \rightarrow \mathbb{R} \quad | \quad f(x_1, x_2) = (x_1^2 + x_2 - 11)^2 + (x_1 + x_2^2) ^2 \end{equation}\] To start the active learning strategy we need to supply some initial data points to set up the Gaussian Regression Model in the background.

himmelblau = Himmelblau()


def sample(domain: Domain):
    datamodel = RandomStrategy(domain=domain)
    sampler = strategies.map(data_model=datamodel)
    sampled = sampler.ask(10)
    return sampled


initial_points = sample(domain=himmelblau.domain)
initial_experiments = pd.concat([initial_points, himmelblau.f(initial_points)], axis=1)
display(initial_experiments)

	x_1	x_2	y	valid_y
0	5.401121	2.675580	465.537862	1
1	0.576342	1.897418	84.892051	1
2	-5.453175	-1.796953	372.054067	1
3	1.778934	5.777340	797.029422	1
4	-4.981100	4.589033	420.986575	1
5	-4.193914	-5.356077	307.547464	1
6	-1.943967	5.154563	314.930112	1
7	-5.617293	0.570447	597.331664	1
8	0.622289	-0.861274	163.416950	1
9	-4.774761	4.802698	403.084476	1

ActiveLearningStrategy

The ActiveLearningstrategy can be set up just as other BO strategies implemented in bofire. Just take a look into the other tutorials. Basic calls are ask() to retrieve new evaluation candidates from the acquisition function and tell() to train the model with a new observation.

Currently, the default active learning acquisition function implemented is qNegIntegratedPosteriorVariance. It focuses on minimizing the overall posterior variance by choosing a new candidate.

The ActiveLearningStrategy uses Monte-Carlo-integration to evaluate the acquisition function. The number of integration nodes significantly influences the speed of the integration. These can be adjusted by changing the parameter data_model.num_sobol_samples. Note that a sample size representing a power of $2$ increases performance.

# Manual set up of ActiveLearning
from bofire.data_models.acquisition_functions.api import qNegIntPosVar
from bofire.data_models.strategies.api import ActiveLearningStrategy
from bofire.data_models.surrogates.api import BotorchSurrogates, SingleTaskGPSurrogate


af = qNegIntPosVar(
    n_mc_samples=64,  # lower the number of monte carlo samples to improve speed
)

data_model = ActiveLearningStrategy(domain=himmelblau.domain, acquisition_function=af)
recommender = strategies.map(data_model=data_model)
recommender.tell(experiments=initial_experiments)
candidates = recommender.ask(candidate_count=1)

display(candidates)

	x_1	x_2	y_pred	y_sd	y_des
0	2.185312	-3.779142	370.752805	191.830065	-370.752805

# Running the active learning strategy
n_iter = 1 if SMOKE_TEST else 20
results = initial_experiments

for _ in range(n_iter):
    # run active learning strategy
    X = recommender.ask(candidate_count=1)[himmelblau.domain.inputs.get_keys()]
    Y = himmelblau.f(X)
    XY = pd.concat([X, Y], axis=1)
    recommender.tell(experiments=XY)  # pass new experimental data
    results = pd.concat([results, XY], axis=0, ignore_index=True)

# Running a random strategy for comparison
def strategy_factory(domain: Domain):
    data_model = RandomStrategy(domain=domain)
    return strategies.map(data_model)


random_results = run(
    himmelblau,
    strategy_factory=strategy_factory,
    n_iterations=n_iter,
    metric=lambda domain, experiments: 1.0,
    initial_sampler=sample,
    n_runs=1,
    n_procs=1,
)

Plotting

if not SMOKE_TEST:
    plt.rcParams["figure.figsize"] = (10, 4)

    fig, ax = plt.subplots(1, 2)

# contour plot of himmelblau
    def f(grid):
        return (grid[0] ** 2 + grid[1] - 11) ** 2 + (grid[0] + grid[1] ** 2) ** 2


    X_grid = np.arange(-7, 7, 0.01)
    Y_grid = np.arange(-7, 7, 0.01)
    mesh = np.meshgrid(X_grid, Y_grid)
    Z = f(grid=mesh)
    levels = np.linspace(Z.min(), Z.max(), 6)


    ax[0].contourf(X_grid, Y_grid, Z, cmap=cm.viridis)
    ax[0].scatter(random_results[0][0].x_1, random_results[0][0].x_2, c="white")
    ax[1].contourf(X_grid, Y_grid, Z, cmap=cm.viridis)
    ax[1].scatter(results.x_1, results.x_2, c="white")

    ax[0].axis([-7, 7, -7, 7])
    ax[0].set_xlabel("$x_1$")
    ax[1].set_xlabel("$x_1$")
    ax[0].set_ylabel("$x_2$")
    ax[0].set_title("random strategy")
    ax[1].set_title("active learning strategy")
    fig.show()

The plot shows the exploratory behavior of the ActiveLearningStrategy.

2-D (n-D) Objective Function

Now, we want to actively learn an objective function with a multi-dimensional output space. This is shown by an example function with $2$ output variables. For this, we again utilize the Himmelblau benchmark function and the Ackley function.

\[\begin{equation} f: \mathbb{R}^2 \rightarrow \mathbb{R}^2 \quad | \quad f(x_1, x_2) = \begin{pmatrix} (x_1^2 + x_2 - 11)^2 + (x_1 + x_2^2) ^2 \\ -20\exp \left[-0.2{\sqrt {0.5\left(x_1^{2}+x_2^{2}\right)}}\right] -\exp \left[0.5\left(\cos 2\pi x_1+\cos 2\pi x_2\right)\right]+e+20 \end{pmatrix} \end{equation}\]

inputs = Inputs(
    features=[
        ContinuousInput(key="x_1", bounds=(-6, 6)),
        ContinuousInput(key="x_2", bounds=(-6, 6)),
    ],
)
outputs = Outputs(
    features=[
        ContinuousOutput(key="f_0", objective=MinimizeObjective()),
        ContinuousOutput(key="f_1", objective=MinimizeObjective()),
    ],
)
domain = Domain(inputs=inputs, outputs=outputs)


def benchmark_function(candidates):
    f0 = (candidates["x_1"] ** 2 + candidates["x_2"] - 11) ** 2 + (
        candidates["x_1"] + candidates["x_2"] ** 2
    ) ** 2
    f1 = -20 * np.exp(
        -0.2 * np.sqrt(0.5 * (candidates["x_1"] ** 2 + candidates["x_2"] ** 2)),
    ) + (
        -np.exp(
            0.5
            * (
                np.cos(2 * np.pi * candidates["x_1"])
                + np.cos(2 * np.pi * candidates["x_2"])
            ),
        )
        + np.exp(1)
        + 20
    )
    return pd.DataFrame({"f_0": f0, "f_1": f1})


function = GenericBenchmark(domain=domain, func=benchmark_function)
initial_experiments = pd.concat(
    [initial_points, function.f(candidates=initial_points)],
    axis=1,
)

For the multi-objective function we need to pass two models to the strategy as each individual output is represented by a separate model. By default, the ActiveLearningStrategy focusses on minimizing the negative integrated posterior variance of each model equally. To minimize the variances in a more specific way certain weights can be provided for each output feature. This can be done by passing a dictionary containing the individual weights for each output feature with its corresponding key to the parameter weights.

# Manual set up of ActiveLearning
weights = {
    "f_0": 0.4,
    "f_1": 0.6,
}
# create an instance of the acquisition function with distinct weights
af = qNegIntPosVar(weights=weights, n_mc_samples=16)

data_model = ActiveLearningStrategy(
    domain=domain,
    surrogate_specs=BotorchSurrogates(
        surrogates=[
            SingleTaskGPSurrogate(
                inputs=domain.inputs,
                outputs=Outputs(features=[domain.outputs[0]]),
            ),
            SingleTaskGPSurrogate(
                inputs=domain.inputs,
                outputs=Outputs(features=[domain.outputs[1]]),
            ),
        ],
    ),
    acquisition_function=af,
)
recommender = strategies.map(data_model=data_model)
recommender.tell(experiments=initial_experiments)
candidates = recommender.ask(candidate_count=1)

display(candidates)

	x_1	x_2	f_0_pred	f_1_pred	f_0_sd	f_1_sd	f_0_des	f_1_des
0	3.841079	-1.367371	392.08179	8.849437	276.17034	1.971168	-392.08179	-8.849437

# Running the active learning strategy
n_iter = 1 if SMOKE_TEST else 20
results = initial_experiments

for _ in range(n_iter):
    # run active learning strategy
    X = recommender.ask(candidate_count=1)[domain.inputs.get_keys()]
    Y = function.f(candidates=X)
    XY = pd.concat([X, Y], axis=1)
    recommender.tell(experiments=XY)  # pass new experimental data
    results = pd.concat([results, XY], axis=0, ignore_index=True)

random_results = run(
    function,
    strategy_factory=strategy_factory,
    n_iterations=n_iter,
    metric=lambda domain, experiments: 1.0,
    initial_sampler=sample,
    n_runs=1,
    n_procs=1,
)

Plotting

if not SMOKE_TEST:
    plt.rcParams["figure.figsize"] = (10, 8)
    fig, ax = plt.subplots(2, 2)


    def f1(grid):
        return (
            -20 * np.exp(-0.2 * np.sqrt(0.5 * (grid[0] ** 2 + grid[1] ** 2)))
            - np.exp(0.5 * (np.cos(2 * np.pi * grid[0]) + np.cos(2 * np.pi * grid[1])))
            + np.exp(1)
            + 20
        )


    Z1 = f1(mesh)
    levels = np.linspace(Z1.min(), Z1.max(), 10)
    ax[0, 0].contourf(
        X_grid,
        Y_grid,
        Z,
        cmap=cm.viridis,
    )
    ax[0, 0].scatter(random_results[0][0].x_1, random_results[0][0].x_2, c="white")
    ax[0, 1].contourf(
        X_grid,
        Y_grid,
        Z,
        cmap=cm.viridis,
    )
    ax[0, 1].scatter(results.x_1, results.x_2, c="white")
    ax[1, 0].contourf(X_grid, Y_grid, Z1, cmap=cm.viridis, levels=levels)
    ax[1, 0].scatter(
        random_results[0][0].x_1,
        random_results[0][0].x_2,
        c="white",
        edgecolors="black",
    )
    ax[1, 1].contourf(X_grid, Y_grid, Z1, cmap=cm.viridis, levels=levels)
    ax[1, 1].scatter(results.x_1, results.x_2, c="white", edgecolors="black")

    ax[0, 0].axis([-7, 7, -7, 7])
    ax[1, 0].set_xlabel("$x_1$")
    ax[1, 1].set_xlabel("$x_1$")
    ax[0, 0].set_ylabel("$x_2$")
    ax[1, 0].set_ylabel("$x_2$")
    ax[0, 0].set_title("random strategy")
    ax[0, 1].set_title("active learning strategy")
    fig.show()

--- title: Active Learning jupyter: python3 --- Showcase of active learning in bofire. Active learning per definition focusses on fitting the model to the experimental observations best possible in an iterative manner reducing some kind of uncertainty. The ActiveLearningStrategy proposes a set of evaluation points that will gain the most information about the problem each iteration. Thus, an unknown black-box-function can be approximated without optimization. It represents an exploration-only strategy. ```{python} import os import matplotlib.pyplot as plt import numpy as np import pandas as pd from matplotlib import cm import bofire.strategies.api as strategies from bofire.benchmarks.api import GenericBenchmark from bofire.benchmarks.single import Himmelblau from bofire.data_models.api import Domain, Inputs, Outputs from bofire.data_models.features.api import ContinuousInput, ContinuousOutput from bofire.data_models.objectives.api import MinimizeObjective from bofire.data_models.strategies.api import RandomStrategy from bofire.runners.api import run SMOKE_TEST = os.environ.get("SMOKE_TEST") ``` ### 1-D Objective Function For a 1-D objective function. The Himmelblau benchmark is used. \begin{equation} f: \mathbb{R}^2 \rightarrow \mathbb{R} \quad | \quad f(x_1, x_2) = (x_1^2 + x_2 - 11)^2 + (x_1 + x_2^2) ^2 \end{equation} To start the active learning strategy we need to supply some initial data points to set up the Gaussian Regression Model in the background. ```{python} himmelblau = Himmelblau() def sample(domain: Domain): datamodel = RandomStrategy(domain=domain) sampler = strategies.map(data_model=datamodel) sampled = sampler.ask(10) return sampled initial_points = sample(domain=himmelblau.domain) initial_experiments = pd.concat([initial_points, himmelblau.f(initial_points)], axis=1) display(initial_experiments) ``` #### ActiveLearningStrategy The `ActiveLearningstrategy` can be set up just as other BO strategies implemented in bofire. Just take a look into the other tutorials. Basic calls are `ask()` to retrieve new evaluation candidates from the acquisition function and `tell()` to train the model with a new observation. Currently, the default active learning acquisition function implemented is `qNegIntegratedPosteriorVariance`. It focuses on minimizing the overall posterior variance by choosing a new candidate. The `ActiveLearningStrategy` uses Monte-Carlo-integration to evaluate the acquisition function. The number of integration nodes significantly influences the speed of the integration. These can be adjusted by changing the parameter `data_model.num_sobol_samples`. Note that a sample size representing a power of $2$ increases performance. ```{python} # | output: false # Manual set up of ActiveLearning from bofire.data_models.acquisition_functions.api import qNegIntPosVar from bofire.data_models.strategies.api import ActiveLearningStrategy from bofire.data_models.surrogates.api import BotorchSurrogates, SingleTaskGPSurrogate af = qNegIntPosVar( n_mc_samples=64, # lower the number of monte carlo samples to improve speed ) data_model = ActiveLearningStrategy(domain=himmelblau.domain, acquisition_function=af) recommender = strategies.map(data_model=data_model) recommender.tell(experiments=initial_experiments) candidates = recommender.ask(candidate_count=1) ``` ```{python} display(candidates) ``` ```{python} # | output: false # Running the active learning strategy n_iter = 1 if SMOKE_TEST else 20 results = initial_experiments for _ in range(n_iter): # run active learning strategy X = recommender.ask(candidate_count=1)[himmelblau.domain.inputs.get_keys()] Y = himmelblau.f(X) XY = pd.concat([X, Y], axis=1) recommender.tell(experiments=XY) # pass new experimental data results = pd.concat([results, XY], axis=0, ignore_index=True) ``` ```{python} # | output: false # Running a random strategy for comparison def strategy_factory(domain: Domain): data_model = RandomStrategy(domain=domain) return strategies.map(data_model) random_results = run( himmelblau, strategy_factory=strategy_factory, n_iterations=n_iter, metric=lambda domain, experiments: 1.0, initial_sampler=sample, n_runs=1, n_procs=1, ) ``` #### Plotting ```{python} if not SMOKE_TEST: plt.rcParams["figure.figsize"] = (10, 4) fig, ax = plt.subplots(1, 2) # contour plot of himmelblau def f(grid): return (grid[0] ** 2 + grid[1] - 11) ** 2 + (grid[0] + grid[1] ** 2) ** 2 X_grid = np.arange(-7, 7, 0.01) Y_grid = np.arange(-7, 7, 0.01) mesh = np.meshgrid(X_grid, Y_grid) Z = f(grid=mesh) levels = np.linspace(Z.min(), Z.max(), 6) ax[0].contourf(X_grid, Y_grid, Z, cmap=cm.viridis) ax[0].scatter(random_results[0][0].x_1, random_results[0][0].x_2, c="white") ax[1].contourf(X_grid, Y_grid, Z, cmap=cm.viridis) ax[1].scatter(results.x_1, results.x_2, c="white") ax[0].axis([-7, 7, -7, 7]) ax[0].set_xlabel("$x_1$") ax[1].set_xlabel("$x_1$") ax[0].set_ylabel("$x_2$") ax[0].set_title("random strategy") ax[1].set_title("active learning strategy") fig.show() ``` The plot shows the exploratory behavior of the `ActiveLearningStrategy`. ### 2-D (n-D) Objective Function Now, we want to actively learn an objective function with a multi-dimensional output space. This is shown by an example function with $2$ output variables. For this, we again utilize the Himmelblau benchmark function and the Ackley function. \begin{equation} f: \mathbb{R}^2 \rightarrow \mathbb{R}^2 \quad | \quad f(x_1, x_2) = \begin{pmatrix} (x_1^2 + x_2 - 11)^2 + (x_1 + x_2^2) ^2 \\ -20\exp \left[-0.2{\sqrt {0.5\left(x_1^{2}+x_2^{2}\right)}}\right] -\exp \left[0.5\left(\cos 2\pi x_1+\cos 2\pi x_2\right)\right]+e+20 \end{pmatrix} \end{equation} ```{python} inputs = Inputs( features=[ ContinuousInput(key="x_1", bounds=(-6, 6)), ContinuousInput(key="x_2", bounds=(-6, 6)), ], ) outputs = Outputs( features=[ ContinuousOutput(key="f_0", objective=MinimizeObjective()), ContinuousOutput(key="f_1", objective=MinimizeObjective()), ], ) domain = Domain(inputs=inputs, outputs=outputs) def benchmark_function(candidates): f0 = (candidates["x_1"] ** 2 + candidates["x_2"] - 11) ** 2 + ( candidates["x_1"] + candidates["x_2"] ** 2 ) ** 2 f1 = -20 * np.exp( -0.2 * np.sqrt(0.5 * (candidates["x_1"] ** 2 + candidates["x_2"] ** 2)), ) + ( -np.exp( 0.5 * ( np.cos(2 * np.pi * candidates["x_1"]) + np.cos(2 * np.pi * candidates["x_2"]) ), ) + np.exp(1) + 20 ) return pd.DataFrame({"f_0": f0, "f_1": f1}) function = GenericBenchmark(domain=domain, func=benchmark_function) initial_experiments = pd.concat( [initial_points, function.f(candidates=initial_points)], axis=1, ) ``` For the multi-objective function we need to pass two models to the strategy as each individual output is represented by a separate model. By default, the `ActiveLearningStrategy` focusses on minimizing the negative integrated posterior variance of each model equally. To minimize the variances in a more specific way certain weights can be provided for each output feature. This can be done by passing a `dictionary` containing the individual weights for each output feature with its corresponding key to the parameter `weights`. ```{python} # | output: false # Manual set up of ActiveLearning weights = { "f_0": 0.4, "f_1": 0.6, } # create an instance of the acquisition function with distinct weights af = qNegIntPosVar(weights=weights, n_mc_samples=16) data_model = ActiveLearningStrategy( domain=domain, surrogate_specs=BotorchSurrogates( surrogates=[ SingleTaskGPSurrogate( inputs=domain.inputs, outputs=Outputs(features=[domain.outputs[0]]), ), SingleTaskGPSurrogate( inputs=domain.inputs, outputs=Outputs(features=[domain.outputs[1]]), ), ], ), acquisition_function=af, ) recommender = strategies.map(data_model=data_model) recommender.tell(experiments=initial_experiments) candidates = recommender.ask(candidate_count=1) ``` ```{python} display(candidates) ``` ```{python} # | output: false # Running the active learning strategy n_iter = 1 if SMOKE_TEST else 20 results = initial_experiments for _ in range(n_iter): # run active learning strategy X = recommender.ask(candidate_count=1)[domain.inputs.get_keys()] Y = function.f(candidates=X) XY = pd.concat([X, Y], axis=1) recommender.tell(experiments=XY) # pass new experimental data results = pd.concat([results, XY], axis=0, ignore_index=True) ``` ```{python} # | output: false random_results = run( function, strategy_factory=strategy_factory, n_iterations=n_iter, metric=lambda domain, experiments: 1.0, initial_sampler=sample, n_runs=1, n_procs=1, ) ``` #### Plotting ```{python} if not SMOKE_TEST: plt.rcParams["figure.figsize"] = (10, 8) fig, ax = plt.subplots(2, 2) def f1(grid): return ( -20 * np.exp(-0.2 * np.sqrt(0.5 * (grid[0] ** 2 + grid[1] ** 2))) - np.exp(0.5 * (np.cos(2 * np.pi * grid[0]) + np.cos(2 * np.pi * grid[1]))) + np.exp(1) + 20 ) Z1 = f1(mesh) levels = np.linspace(Z1.min(), Z1.max(), 10) ax[0, 0].contourf( X_grid, Y_grid, Z, cmap=cm.viridis, ) ax[0, 0].scatter(random_results[0][0].x_1, random_results[0][0].x_2, c="white") ax[0, 1].contourf( X_grid, Y_grid, Z, cmap=cm.viridis, ) ax[0, 1].scatter(results.x_1, results.x_2, c="white") ax[1, 0].contourf(X_grid, Y_grid, Z1, cmap=cm.viridis, levels=levels) ax[1, 0].scatter( random_results[0][0].x_1, random_results[0][0].x_2, c="white", edgecolors="black", ) ax[1, 1].contourf(X_grid, Y_grid, Z1, cmap=cm.viridis, levels=levels) ax[1, 1].scatter(results.x_1, results.x_2, c="white", edgecolors="black") ax[0, 0].axis([-7, 7, -7, 7]) ax[1, 0].set_xlabel("$x_1$") ax[1, 1].set_xlabel("$x_1$") ax[0, 0].set_ylabel("$x_2$") ax[1, 0].set_ylabel("$x_2$") ax[0, 0].set_title("random strategy") ax[0, 1].set_title("active learning strategy") fig.show() ```