BoFire Candidates API

An experimental FastAPI based application that can be used to generate candidates via http(s) using BoFire. It makes use of pydantic data_models in BoFire, which allows for an easy fastAPI integration including Swagger based documentation which can be found when visiting /docs of the running web application.

Candidates can be generated via two different ways, either directly at request which can lead to http timeouts or using an asynchronous worker based procedure.

Usage

Direct Candidate Generation

In the following it is shown how to generate candidates in the direct way using a post request.

import json
from typing import Optional

import requests
from bofire.benchmarks.api import Himmelblau
from bofire.data_models.dataframes.api import Candidates, Experiments
from bofire.data_models.strategies.api import (
    AlwaysTrueCondition,
    AnyStrategy,
    NumberOfExperimentsCondition,
    RandomStrategy,
    SoboStrategy,
    Step,
    StepwiseStrategy,
)
from pydantic import BaseModel


class CandidateRequest(BaseModel):
    strategy_data: AnyStrategy
    n_candidates: int
    experiments: Optional[Experiments]
    pendings: Optional[Candidates]

# generate experimental data from the Himmelblau benchmark
bench = Himmelblau()
experiments = bench.f(bench.domain.inputs.sample(10), return_complete=True)

# setup the strategy for which candidates should be generated
strategy_data = StepwiseStrategy(
    domain=bench.domain,
    steps=[
        Step(
            condition=NumberOfExperimentsCondition(
                n_experiments=10
            ),
            strategy_data=RandomStrategy(domain=bench.domain)
        ),
        Step(
            condition=AlwaysTrueCondition(),
            strategy_data=SoboStrategy(domain=bench.domain)
        )
    ]
)

# create the payload
payload = CandidateRequest(
    strategy_data=strategy_data,
    n_candidates=1,
    experiments=Experiments.from_pandas(experiments, bench.domain),
    pendings=None
)


URL = "http://127.0.0.1:8000/candidates"
HEADERS = {'accept': 'application/json', 'Content-Type': 'application/json'}

# request candidates
response = requests.post(url=f"{URL}/generate", data=payload.model_dump_json(), headers=HEADERS)

# convert response to a pandas dataframe
df_candidates =Candidates(**json.loads(response.content)).to_pandas()

Worker Based Candidate Generation

The following snippet shows how to use the worker based candidate generation using the same payload as above. The API is storing all necessary information regarding the proposals in a TinyDB database. Note that concurrent worker access using multiple users has not been tested yet.

import time

# create the proposal in the database
response = requests.post(url=f"{URL}/proposals", json=payload.model_dump(), headers=HEADERS)
id = json.loads(response.content)["id"]

# poll the state of the proposal
def get_state(id:int):
    return requests.get(url=f"{URL}proposals/{id}/state", headers=HEADERS).json()

state = get_state(id)

while state in ["CREATED", "CLAIMED"]:
    state = get_state(id)
    time.sleep(5)

# get the candidates when the worker is finished
if state=="FINISHED":
    response = requests.get(url=f"{URL}proposals/{id}/candidates", headers=HEADERS)
    candidates = Candidates(**response.json()).to_pandas()
else:
    print(state) # candidate generation was not successful.

Installation

Use the following command to set and run the API locally as well as run the unit tests.

Setup

pip install -r requirements.txt

Run

uvicorn --app-dir=app app:app --reload

If you also want to use the asynchronous worker based candidate generation, use the following snippet to start at least one worker:

python worker

Run unit tests

pytest