File size: 7,487 Bytes
95c4c89 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """BoFire optimizer backend for physics-informed BO."""
from typing import Callable, Dict, List, Optional, Tuple
import torch
from torch import Tensor
from physics_informed_bo.config import OptimizationConfig
from physics_informed_bo.optimizers.base_optimizer import BaseOptimizer
class BoFireOptimizer(BaseOptimizer):
"""BoFire backend for chemistry/materials-focused Bayesian optimization.
BoFire is designed for real-world experimental design in chemistry
and materials science. It supports:
- Complex parameter spaces (continuous, categorical, molecular)
- Mixture constraints (sum-to-one)
- Multi-objective optimization with Pareto fronts
- Integration with domain-specific descriptors
The physics model is incorporated as a prior mean function in BoFire's
surrogate model specification.
"""
def __init__(self, config: OptimizationConfig):
super().__init__(config)
self._domain = None
self._strategy = None
self._experiments_df = None
def setup_domain(
self,
parameters: Dict[str, Dict],
objectives: Dict[str, Dict],
constraints: Optional[List[Dict]] = None,
) -> None:
"""Set up a BoFire domain with physics-informed features.
Args:
parameters: Dict of parameter specifications.
Example: {"temp": {"type": "continuous", "bounds": (300, 500)}}
objectives: Dict of objective specifications.
Example: {"yield": {"type": "maximize", "weight": 1.0}}
constraints: Optional list of constraint specifications.
Example: [{"type": "linear", "features": ["x1", "x2"], "coeffs": [1, 1], "rhs": 1}]
"""
try:
from bofire.data_models.domain.api import Domain, Inputs, Outputs
from bofire.data_models.features.api import (
ContinuousInput,
ContinuousOutput,
CategoricalInput,
)
from bofire.data_models.objectives.api import MaximizeObjective, MinimizeObjective
from bofire.data_models.constraints.api import (
LinearInequalityConstraint,
LinearEqualityConstraint,
)
except ImportError:
raise ImportError(
"BoFire is required for BoFireOptimizer. "
"Install with: pip install bofire"
)
# Build input features
input_features = []
self._feature_names = []
for name, spec in parameters.items():
self._feature_names.append(name)
if spec["type"] == "continuous":
lb, ub = spec["bounds"]
input_features.append(
ContinuousInput(key=name, bounds=(float(lb), float(ub)))
)
elif spec["type"] == "categorical":
input_features.append(
CategoricalInput(key=name, categories=spec["categories"])
)
# Build output features
output_features = []
for name, spec in objectives.items():
if spec.get("type", "maximize") == "maximize":
obj = MaximizeObjective(w=spec.get("weight", 1.0))
else:
obj = MinimizeObjective(w=spec.get("weight", 1.0))
output_features.append(ContinuousOutput(key=name, objective=obj))
# Build constraints
bofire_constraints = []
if constraints:
for c in constraints:
if c["type"] == "linear_inequality":
bofire_constraints.append(
LinearInequalityConstraint(
features=c["features"],
coefficients=c["coeffs"],
rhs=c["rhs"],
)
)
elif c["type"] == "linear_equality":
bofire_constraints.append(
LinearEqualityConstraint(
features=c["features"],
coefficients=c["coeffs"],
rhs=c["rhs"],
)
)
self._domain = Domain(
inputs=Inputs(features=input_features),
outputs=Outputs(features=output_features),
constraints=bofire_constraints if bofire_constraints else None,
)
def setup_strategy(self, strategy_type: str = "sobo") -> None:
"""Set up the BoFire optimization strategy.
Args:
strategy_type: One of 'sobo' (single-objective), 'mobo' (multi-objective),
'qehvi' (q-Expected Hypervolume Improvement).
"""
try:
from bofire.data_models.strategies.api import SoboStrategy, QehviStrategy
from bofire.data_models.acquisition_functions.api import qEI, qNEI
import bofire.strategies.api as strategies
except ImportError:
raise ImportError("BoFire is required. Install with: pip install bofire")
if self._domain is None:
raise RuntimeError("Call setup_domain() before setup_strategy().")
if strategy_type == "sobo":
strategy_data = SoboStrategy(domain=self._domain, acquisition_function=qEI())
elif strategy_type in ("mobo", "qehvi"):
strategy_data = QehviStrategy(domain=self._domain)
else:
raise ValueError(f"Unsupported strategy type: {strategy_type}")
self._strategy = strategies.map(strategy_data)
def suggest(
self,
n_candidates: int = 1,
X_observed: Optional[Tensor] = None,
y_observed: Optional[Tensor] = None,
) -> Tensor:
"""Suggest next experiments using BoFire."""
if self._strategy is None:
raise RuntimeError("Call setup_domain() and setup_strategy() first.")
import pandas as pd
# Tell strategy about existing experiments
if self._experiments_df is not None:
self._strategy.tell(self._experiments_df)
candidates_df = self._strategy.ask(n_candidates)
candidates = torch.tensor(
candidates_df[self._feature_names].values, dtype=torch.float64
)
# Filter through physics constraints
candidates = self._filter_feasible(candidates)
return candidates[:n_candidates]
def update(self, X_new: Tensor, y_new: Tensor) -> None:
"""Update BoFire with new observations."""
import pandas as pd
data = {}
for i, name in enumerate(self._feature_names):
data[name] = X_new[:, i].numpy()
# Assume single objective for now
output_keys = [f.key for f in self._domain.outputs.features]
for i, key in enumerate(output_keys):
if y_new.dim() > 1 and y_new.shape[1] > i:
data[key] = y_new[:, i].numpy()
else:
data[key] = y_new.squeeze().numpy()
new_df = pd.DataFrame(data)
if self._experiments_df is None:
self._experiments_df = new_df
else:
self._experiments_df = pd.concat(
[self._experiments_df, new_df], ignore_index=True
)
|