BayesianOptimizer

The optimizer API supports:

• Local and remote evaluation backends.
• Explicit inequality constraints using <= and >=.
• Trial-history exports as explicit records with parameters and objectives.
• Typed failures for configuration, transport, contract, and execution errors.

mdo_framework.optimization.optimizer.BayesianOptimizer

Bayesian Optimizer using Ax Platform.

Parameters:

Name	Type	Description	Default
evaluator	Evaluator	Local or Remote implementation of Evaluator protocol.	required
parameters	list[dict[str, Any]]	Dict defining the variables bounds, choices, and types.	required
objectives	list[dict[str, Any]]	Dict defining the targeted metrics and their directions.	required
constraints	list[dict[str, Any]] | None	Dict defining boundaries mapped out of GEMSEO evaluations.	None
fidelity_parameter	str | None	Name of variable designating multi-fidelity.	None
use_bonsai	bool	Toggle for experimental algorithmic execution.	False
parameter_constraints	list[str] | None	List of string-based constraints on the search space parameters.	None

Source code in src/mdo_framework/optimization/optimizer.py

class BayesianOptimizer:
    """Bayesian Optimizer using Ax Platform.

    Args:
        evaluator: Local or Remote implementation of Evaluator protocol.
        parameters: Dict defining the variables bounds, choices, and types.
        objectives: Dict defining the targeted metrics and their directions.
        constraints: Dict defining boundaries mapped out of GEMSEO evaluations.
        fidelity_parameter: Name of variable designating multi-fidelity.
        use_bonsai: Toggle for experimental algorithmic execution.
        parameter_constraints: List of string-based constraints on the search space parameters.

    """

    def __init__(
        self,
        evaluator: Evaluator,
        parameters: list[dict[str, Any]],
        objectives: list[dict[str, Any]],
        constraints: list[dict[str, Any]] | None = None,
        fidelity_parameter: str | None = None,
        use_bonsai: bool = False,
        parameter_constraints: list[str] | None = None,
    ) -> None:
        self.evaluator = evaluator
        self.parameters = parameters
        self.objectives = objectives
        self.constraints = constraints or []
        self.fidelity_parameter = fidelity_parameter
        self.use_bonsai = use_bonsai
        self.parameter_constraints = parameter_constraints

    def _build_output_names(self) -> list[str]:
        return [objective["name"] for objective in self.objectives] + [
            constraint["name"] for constraint in self.constraints
        ]

    def _build_discipline(self) -> Discipline:
        if hasattr(self.evaluator, "problem"):
            return self.evaluator.problem
        return RemoteDiscipline(
            self.evaluator,
            self.parameters,
            self._build_output_names(),
        )

    def _prepare_scenario_context(self) -> tuple[Discipline, DesignSpace, list[str]]:
        return (
            self._build_discipline(),
            _build_design_space(self.parameters),
            [objective["name"] for objective in self.objectives],
        )

    def _create_scenario(
        self,
        *,
        discipline: Discipline,
        design_space: DesignSpace,
        objective_names: list[str],
        scenario_type: str | None = None,
        maximize_objective: bool | list[bool] | None = None,
        name: str | None = None,
    ) -> Any:
        scenario_kwargs: dict[str, Any] = {
            "formulation_name": "MDF",
            "objective_name": objective_names,
            "design_space": design_space,
        }
        if scenario_type is not None:
            scenario_kwargs["scenario_type"] = scenario_type
        if maximize_objective is not None:
            scenario_kwargs["maximize_objective"] = maximize_objective
        if name is not None:
            scenario_kwargs["name"] = name

        scenario = create_scenario([discipline], **scenario_kwargs)
        _add_constraints_to_scenario(scenario, self.constraints)
        return scenario

    def explore(self, n_samples: int = 10, n_processes: int = 1) -> dict[str, Any]:
        """Runs a Design of Experiments (DOE) exploration using GEMSEO DOEScenario.

        Args:
            n_samples: Number of samples to evaluate.
            n_processes: Number of concurrent processes.

        Returns:
            A dictionary containing the exploration history.
        """
        discipline, design_space, objective_names = self._prepare_scenario_context()

        scenario = self._create_scenario(
            discipline=discipline,
            design_space=design_space,
            objective_names=objective_names,
            scenario_type="DOE",
        )

        try:
            scenario.execute(
                algo_name="Sobol",
                n_samples=n_samples,
                n_processes=n_processes,
            )

            # Post-process
            try:
                from gemseo.settings.post import ScatterPlotMatrix_Settings

                scenario.post_process(
                    "ScatterPlotMatrix",
                    settings_model=ScatterPlotMatrix_Settings(save=True, show=False),
                )
            except Exception as pp_err:
                logger.warning(f"Failed to post-process DOE: {pp_err}")

            return {
                "history": scenario.to_dataset(),
            }
        except (
            OptimizationConfigurationError,
            OptimizationExecutionError,
            RemoteEvaluationContractError,
            RemoteEvaluationTransportError,
        ):
            raise
        except Exception as e:
            logger.error(f"Exploration failed: {e}")
            raise OptimizationExecutionError(f"Exploration failed: {str(e)}") from e

    def optimize(self, n_steps: int = 5, n_init: int = 5) -> dict[str, Any]:
        """Runs the optimization loop using GEMSEO MDOScenario."""
        if self.fidelity_parameter is not None:
            warnings.warn("fidelity_parameter is ignored.")
        discipline, design_space, objective_names = self._prepare_scenario_context()

        # Explicitly configure maximize_objective per user request.
        # GEMSEO maximize_objective expects a single boolean or a list of booleans
        maximize_objective = [not o.get("minimize", True) for o in self.objectives]
        if len(maximize_objective) == 1:
            maximize_objective = maximize_objective[0]

        scenario = self._create_scenario(
            discipline=discipline,
            design_space=design_space,
            objective_names=objective_names,
            maximize_objective=maximize_objective,
            name="MDOScenario_Ax",
        )

        algo = None
        try:
            from mdo_framework.optimization.ax_algo_lib import AxOptimizationLibrary

            problem = scenario.formulation.optimization_problem

            algo = AxOptimizationLibrary()
            algo.execute(
                problem,
                max_iter=n_steps,
                n_init=n_init,
                use_bonsai=self.use_bonsai,
                ax_parameters=self.parameters,
                ax_objectives=self.objectives,
                ax_parameter_constraints=self.parameter_constraints,
            )

            # Generate XDSM diagram
            scenario.xdsmize(show_html=False)
            # Generate Post-Processing
            try:
                from gemseo.settings.post import OptHistoryView_Settings

                scenario.post_process(
                    settings_model=OptHistoryView_Settings(save=True, show=False)
                )
            except Exception as pp_err:
                logger.warning(f"Failed to post-process: {pp_err}")

            optimum = problem.optimum
            if optimum is None:
                raise OptimizationExecutionError(
                    "Optimization completed without a valid GEMSEO optimum."
                )

            best_params = _extract_best_parameters(
                optimum,
                design_space,
                self.parameters,
            )
            best_objectives = _extract_best_objectives(
                optimum,
                objective_names,
                getattr(algo, "best_objectives", None),
            )

            return {
                "best_parameters": best_params,
                "best_objectives": best_objectives,
                "history": _get_optimization_history(scenario, algo),
            }
        except (
            OptimizationConfigurationError,
            OptimizationExecutionError,
            RemoteEvaluationContractError,
            RemoteEvaluationTransportError,
        ):
            raise
        except Exception as e:
            import traceback

            logger.error(f"Optimization failed: {e}\n{traceback.format_exc()}")
            raise OptimizationExecutionError(f"Optimization failed: {str(e)}") from e

explore(n_samples=10, n_processes=1)

Runs a Design of Experiments (DOE) exploration using GEMSEO DOEScenario.

Parameters:

Name	Type	Description	Default
n_samples	int	Number of samples to evaluate.	10
n_processes	int	Number of concurrent processes.	1

Returns:

Type	Description
dict[str, Any]	A dictionary containing the exploration history.

Source code in src/mdo_framework/optimization/optimizer.py

def explore(self, n_samples: int = 10, n_processes: int = 1) -> dict[str, Any]:
    """Runs a Design of Experiments (DOE) exploration using GEMSEO DOEScenario.

    Args:
        n_samples: Number of samples to evaluate.
        n_processes: Number of concurrent processes.

    Returns:
        A dictionary containing the exploration history.
    """
    discipline, design_space, objective_names = self._prepare_scenario_context()

    scenario = self._create_scenario(
        discipline=discipline,
        design_space=design_space,
        objective_names=objective_names,
        scenario_type="DOE",
    )

    try:
        scenario.execute(
            algo_name="Sobol",
            n_samples=n_samples,
            n_processes=n_processes,
        )

        # Post-process
        try:
            from gemseo.settings.post import ScatterPlotMatrix_Settings

            scenario.post_process(
                "ScatterPlotMatrix",
                settings_model=ScatterPlotMatrix_Settings(save=True, show=False),
            )
        except Exception as pp_err:
            logger.warning(f"Failed to post-process DOE: {pp_err}")

        return {
            "history": scenario.to_dataset(),
        }
    except (
        OptimizationConfigurationError,
        OptimizationExecutionError,
        RemoteEvaluationContractError,
        RemoteEvaluationTransportError,
    ):
        raise
    except Exception as e:
        logger.error(f"Exploration failed: {e}")
        raise OptimizationExecutionError(f"Exploration failed: {str(e)}") from e

optimize(n_steps=5, n_init=5)

Runs the optimization loop using GEMSEO MDOScenario.

Source code in src/mdo_framework/optimization/optimizer.py

def optimize(self, n_steps: int = 5, n_init: int = 5) -> dict[str, Any]:
    """Runs the optimization loop using GEMSEO MDOScenario."""
    if self.fidelity_parameter is not None:
        warnings.warn("fidelity_parameter is ignored.")
    discipline, design_space, objective_names = self._prepare_scenario_context()

    # Explicitly configure maximize_objective per user request.
    # GEMSEO maximize_objective expects a single boolean or a list of booleans
    maximize_objective = [not o.get("minimize", True) for o in self.objectives]
    if len(maximize_objective) == 1:
        maximize_objective = maximize_objective[0]

    scenario = self._create_scenario(
        discipline=discipline,
        design_space=design_space,
        objective_names=objective_names,
        maximize_objective=maximize_objective,
        name="MDOScenario_Ax",
    )

    algo = None
    try:
        from mdo_framework.optimization.ax_algo_lib import AxOptimizationLibrary

        problem = scenario.formulation.optimization_problem

        algo = AxOptimizationLibrary()
        algo.execute(
            problem,
            max_iter=n_steps,
            n_init=n_init,
            use_bonsai=self.use_bonsai,
            ax_parameters=self.parameters,
            ax_objectives=self.objectives,
            ax_parameter_constraints=self.parameter_constraints,
        )

        # Generate XDSM diagram
        scenario.xdsmize(show_html=False)
        # Generate Post-Processing
        try:
            from gemseo.settings.post import OptHistoryView_Settings

            scenario.post_process(
                settings_model=OptHistoryView_Settings(save=True, show=False)
            )
        except Exception as pp_err:
            logger.warning(f"Failed to post-process: {pp_err}")

        optimum = problem.optimum
        if optimum is None:
            raise OptimizationExecutionError(
                "Optimization completed without a valid GEMSEO optimum."
            )

        best_params = _extract_best_parameters(
            optimum,
            design_space,
            self.parameters,
        )
        best_objectives = _extract_best_objectives(
            optimum,
            objective_names,
            getattr(algo, "best_objectives", None),
        )

        return {
            "best_parameters": best_params,
            "best_objectives": best_objectives,
            "history": _get_optimization_history(scenario, algo),
        }
    except (
        OptimizationConfigurationError,
        OptimizationExecutionError,
        RemoteEvaluationContractError,
        RemoteEvaluationTransportError,
    ):
        raise
    except Exception as e:
        import traceback

        logger.error(f"Optimization failed: {e}\n{traceback.format_exc()}")
        raise OptimizationExecutionError(f"Optimization failed: {str(e)}") from e

Evaluator

mdo_framework.optimization.optimizer.Evaluator

Bases: Protocol

Source code in src/mdo_framework/optimization/optimizer.py

class Evaluator(Protocol):
    def evaluate(
        self,
        parameters: dict[str, Any],
        objectives: list[str],
    ) -> dict[str, float]:
        """Evaluates the requested objectives given the design parameters."""
        ...

evaluate(parameters, objectives)

Evaluates the requested objectives given the design parameters.

Source code in src/mdo_framework/optimization/optimizer.py

def evaluate(
    self,
    parameters: dict[str, Any],
    objectives: list[str],
) -> dict[str, float]:
    """Evaluates the requested objectives given the design parameters."""
    ...

LocalEvaluator

mdo_framework.core.evaluators.LocalEvaluator

Evaluates the design parameters locally using a GEMSEO MDA instance.

Parameters:

Name	Type	Description	Default
problem	Discipline	An instantiated GEMSEO MDA (or Discipline) object.	required

Source code in src/mdo_framework/core/evaluators.py

class LocalEvaluator:
    """Evaluates the design parameters locally using a GEMSEO MDA instance.

    Args:
        problem: An instantiated GEMSEO MDA (or Discipline) object.
    """

    def __init__(self, problem: Discipline):
        self.problem = problem

    def evaluate(
        self,
        parameters: dict[str, Any],
        objectives: list[str],
    ) -> dict[str, float]:
        # GEMSEO uses a dictionary with string keys and numpy array values for local_data
        input_data = {name: np.atleast_1d(val) for name, val in parameters.items()}

        # We need to provide all required inputs for the MDA, not just parameters.
        # This will be passed and merged internally by execute.
        output_data = self.problem.execute(input_data)

        results = {}
        for obj in objectives:
            val = output_data.get(obj)
            if val is None:
                raise KeyError(f"Missing objective or constraint output: {obj}")
            results[obj] = (
                float(val[0])
                if isinstance(val, np.ndarray) and val.size > 0
                else float(val)
            )
        return results

RemoteEvaluator

RemoteEvaluator distinguishes transport failures from invalid execution-service responses so service layers can map them to different HTTP statuses.

mdo_framework.optimization.optimizer.RemoteEvaluator

Evaluates the design parameters remotely by communicating with the Execution microservice.

Parameters:

Name	Type	Description	Default
service_url	str	The URL of the execution service.	required

Source code in src/mdo_framework/optimization/optimizer.py

class RemoteEvaluator:
    """Evaluates the design parameters remotely by communicating with the Execution microservice.

    Args:
        service_url: The URL of the execution service.

    """

    def __init__(
        self,
        service_url: str,
        client: httpx.Client | None = None,
        timeout: httpx.Timeout | None = None,
    ):
        self.service_url = service_url.rstrip("/")
        self._owns_client = client is None
        self.client = client or httpx.Client(
            base_url=self.service_url,
            timeout=timeout or httpx.Timeout(30.0, connect=5.0),
        )

    def close(self) -> None:
        if self._owns_client:
            self.client.close()

    def evaluate(
        self,
        parameters: dict[str, Any],
        objectives: list[str],
    ) -> dict[str, float]:
        payload = {
            "inputs": parameters,
            "objectives": objectives,
        }
        try:
            response = self.client.post(f"{self.service_url}/evaluate", json=payload)
            response.raise_for_status()
        except httpx.TimeoutException as exc:
            raise RemoteEvaluationTransportError(
                "Execution service request timed out."
            ) from exc
        except httpx.HTTPStatusError as exc:
            if exc.response.status_code >= 500:
                raise RemoteEvaluationTransportError(
                    f"Execution service returned HTTP {exc.response.status_code}."
                ) from exc
            raise RemoteEvaluationContractError(
                f"Execution service rejected the evaluation request with HTTP {exc.response.status_code}."
            ) from exc
        except httpx.RequestError as exc:
            raise RemoteEvaluationTransportError(
                "Execution service request failed."
            ) from exc

        try:
            data = response.json()
        except ValueError as exc:
            raise RemoteEvaluationContractError(
                "Execution service returned invalid JSON."
            ) from exc

        results = data.get("results")
        if not isinstance(results, dict):
            raise RemoteEvaluationContractError(
                "Execution service response is missing a 'results' object."
            )

        missing_objectives = [name for name in objectives if name not in results]
        if missing_objectives:
            raise RemoteEvaluationContractError(
                "Execution service response is missing requested objectives: "
                + ", ".join(missing_objectives)
                + "."
            )

        normalized_results: dict[str, float] = {}
        for objective_name in objectives:
            try:
                normalized_results[objective_name] = float(results[objective_name])
            except (TypeError, ValueError) as exc:
                raise RemoteEvaluationContractError(
                    f"Execution service returned a non-numeric value for {objective_name}."
                ) from exc
        return normalized_results