Agent#

class blop.ax.Agent(sensors, dofs, objectives, evaluation_function, acquisition_plan=None, dof_constraints=None, outcome_constraints=None, checkpoint_path=None, **kwargs)[source]#

Bases: _AxAgentMixin

An interface that uses Ax as the backend for optimization and experiment tracking.

The Agent is the main entry point for setting up and running Bayesian optimization using Blop. It coordinates the DOFs, objectives, evaluation function, and optimizer to perform intelligent exploration of the parameter space.

Parameters:

sensorsSequence[Sensor]: The sensors to use for acquisition. These should be the minimal set of sensors that are needed to compute the objectives.
dofsSequence[DOF]: The degrees of freedom that the agent can control, which determine the search space.
objectivesSequence[Objective]: The objectives which the agent will try to optimize.
evaluation_functionEvaluationFunction: The function to evaluate acquired data and produce outcomes.
acquisition_planAcquisitionPlan | None, optional: The acquisition plan to use for acquiring data from the beamline. If not provided, blop.plans.default_acquire() will be used.
dof_constraintsSequence[DOFConstraint] | None, optional: Constraints on DOFs to refine the search space.
outcome_constraintsSequence[OutcomeConstraint] | None, optional: Constraints on outcomes to be satisfied during optimization.
checkpoint_pathstr | None, optional: The path to the checkpoint file to save the optimizer’s state to.
**kwargsAny: Additional keyword arguments to configure the Ax experiment.

See also

blop.protocols.Sensor: The protocol for sensors.
blop.ax.dof.RangeDOF: For continuous parameters.
blop.ax.dof.ChoiceDOF: For discrete parameters.
blop.ax.objective.Objective: For defining objectives.
blop.ax.optimizer.AxOptimizer: The optimizer used internally.
blop.plans.optimize: Bluesky plan for running optimization.

Notes

For more complex setups, you can configure the Ax client directly via self.ax_client.

For complete working examples of creating and using an Agent, see the tutorial documentation, particularly Your first Bayesian optimization with Blop.

classmethod from_checkpoint(checkpoint_path, actuators, sensors, evaluation_function, acquisition_plan=None)[source]#

Load an agent from the optimizer’s checkpoint file.

Note

Only the optimizer state is saved during a checkpoint, so we cannot reliably validate the remaining state against the optimizer configuration.

Parameters:

checkpoint_pathstr: The checkpoint path to load the agent from.
actuators: Sequence[Actuator]: Objects that can be moved to control the beamline using the Bluesky RunEngine. A subset of the actuators’ names must match the names of suggested parameterizations.
sensors: Sequence[Sensor]: Objects that can produce data to acquire data from the beamline using the Bluesky RunEngine.
evaluation_function: EvaluationFunction: A callable to evaluate data from a Bluesky run and produce outcomes.
acquisition_plan: AcquisitionPlan, optional: A Bluesky plan to acquire data from the beamline. If not provided, a default plan will be used.

property callbacks: list[CallbackBase]#

The list of active optimization callbacks.

Callbacks in this list receive documents from "optimize" and "sample_suggestions" runs. The default list contains an OptimizationLogger.

The list can be mutated directly, or use subscribe() / unsubscribe() for convenience.

subscribe(callback)[source]#

Subscribe a callback to receive optimization run documents.

Parameters:

callbackCallbackBase: A Bluesky callback instance.

Raises:

ValueError: If callback is already subscribed.

unsubscribe(callback)[source]#

Unsubscribe a previously subscribed callback.

Parameters:

callbackCallbackBase: The callback instance to remove.

Raises:

ValueError: If callback is not subscribed.

property sensors: Sequence[Readable | EventCollectable | EventPageCollectable]#: The sensors used for data acquisition.

property actuators: Sequence[MovableHasName | Flyable]#: The actuators that control the degrees of freedom.

property evaluation_function: EvaluationFunction#: The function used to evaluate acquired data and produce outcomes.

property acquisition_plan: AcquisitionPlan | None#: The acquisition plan for acquiring data, or None if using the default.

to_optimization_problem()[source]#

Construct an optimization problem from the agent.

Creates an immutable blop.protocols.OptimizationProblem that encapsulates all components needed for optimization. This is typically used internally by optimization plans.

Returns:

OptimizationProblem: An immutable optimization problem that can be deployed via Bluesky.

See also

blop.protocols.OptimizationProblem: The optimization problem dataclass.
blop.plans.optimize: Uses the optimization problem to run optimization.

acquire_baseline(parameterization=None)[source]#

Acquire a baseline reading for reference.

Acquires data at a specific parameterization (or current positions) to establish a baseline for comparison. Useful for relative outcome constraints.

Parameters:

parameterizationdict[str, Any] | None, optional: The DOF values to move to before acquiring baseline. If None, acquires at current positions.

Yields:

Msg: Bluesky messages for the run engine.

See also

blop.plans.acquire_baseline: The underlying Bluesky plan.

optimize(iterations=1, n_points=1)[source]#

Run Bayesian optimization.

Performs iterative optimization by suggesting points, acquiring data, evaluating outcomes, and updating the model. This is the main method for running optimization with an agent.

Parameters:

iterationsint, optional: The number of optimization iterations to run. Default is 1. Each iteration suggests, evaluates, and learns from n_points.
n_pointsint, optional: The number of points to evaluate per iteration. Default is 1. Higher values enable batch optimization but may reduce optimization efficiency per iteration.

Yields:

Msg: Bluesky messages for the run engine.

See also

blop.plans.optimize: The underlying Bluesky optimization plan.
suggest: Get point suggestions without running acquisition.
ingest: Manually ingest evaluation results.

Notes

This is the primary method for running optimization. It handles the full loop of suggesting points, acquiring data, evaluating outcomes, and updating the model.

For complete examples, see Your first Bayesian optimization with Blop.

sample_suggestions(suggestions)[source]#

Evaluate specific parameter combinations.

Acquires data for given suggestions and ingests results. Supports both optimizer suggestions and manual points.

Parameters:

suggestionslist[dict]: Either optimizer suggestions (with “_id”) or manual points (without “_id”).

Returns:

tuple[str, list[dict], list[dict]]: Bluesky run UID, suggestions with “_id”, and outcomes.

See also

suggest: Get optimizer suggestions.
optimize: Run full optimization loop.

navigate_to_best(parameterization=None)[source]#

Move actuators to the best point found during optimization.

If no explicit parameterization is provided, queries the optimizer for its best point(s). For multi-objective optimizers that return multiple Pareto-optimal points, an explicit parameterization must be provided.

Parameters:

parameterizationMapping | None, optional: Explicit parameterization to navigate to. If None, queries the optimizer’s best point. For multi-objective problems, call get_best_points() to inspect the Pareto set and select one.

Raises:

ValueError: If the optimizer returns multiple Pareto-optimal points and no explicit parameterization is provided.

See also

get_best_points: Query the optimizer for best points.

property ax_client: Client#

checkpoint()#

Save the agent’s state to a JSON file.

property checkpoint_path: str | None#

property fixed_dofs: dict[str, Any] | None#

get_best_points()#

Get a list of the optimal points found during optimization.

For single-objective optimization, returns a single best point. For multi-objective optimization, returns the Pareto-optimal set.

Returns:

list[tuple[int, TParameterization, TOutcome]]

Each element in the list is a tuple of:

trial index (int)
parameter values (dict)
metric values (dict, where values may be (value, sem) tuples)

See also

navigate_to_best: Plan stub to move actuators to a best point.

ingest(points)#

Ingest evaluation results into the optimizer.

Updates the optimizer’s model with new data. Can ingest both suggested points (with “_id” key) and external data (without “_id” key).

Parameters:

pointslist[dict]: A list of dictionaries, each containing outcomes for a trial. For suggested points, include the “_id” key. For external data, include DOF names and objective values, and omit “_id”.

Notes

This method is typically called automatically by optimize(). Manual usage is only needed for custom workflows or when ingesting external data.

For complete examples, see Attach external data to experiments.

plot_objective(x_dof_name, y_dof_name, objective_name, *args, **kwargs)#

Plot the predicted objective as a function of two DOFs.

Creates a contour plot showing the model’s prediction of an objective across the space defined by two DOFs. Useful for visualizing the optimization landscape.

Parameters:

x_dof_namestr: The name of the DOF to plot on the x-axis.
y_dof_namestr: The name of the DOF to plot on the y-axis.
objective_namestr: The name of the objective to plot.
*argsAny: Additional positional arguments passed to Ax’s compute_analyses.
**kwargsAny: Additional keyword arguments passed to Ax’s compute_analyses.

Returns:

list[AnalysisCard]: The computed analysis cards containing the plot data.

See also

ax.analysis.ContourPlot: Pre-built analysis for plotting objectives.
ax.analysis.AnalysisCard: Contains the raw and computed data.

reconfigure_search_space(dof_mappings)#

Update bounds or values of existing DOFs for future optimizations.

Parameters:

dof_mappingsdict[DOF, tuple[float, float] | list[float] | list[int] | list[str] | list[bool]]: Mapping of DOFs to their new search space.

suggest(num_points=1)#

Get the next point(s) to evaluate in the search space.

Uses the Bayesian optimization algorithm to suggest promising points based on all previously acquired data. Each suggestion includes an “_id” key for tracking.

Parameters:

num_pointsint, optional: The number of points to suggest. Default is 1. Higher values enable batch optimization but may reduce optimization efficiency per iteration.

Returns:

list[dict]: A list of dictionaries, each containing a parameterization of a point to evaluate next. Each dictionary includes an “_id” key for identification.

class blop.ax.QueueserverAgent(re_manager_api, zmq_consumer_addr, sensors, dofs, objectives, evaluation_function, acquisition_plan=None, dof_constraints=None, outcome_constraints=None, checkpoint_path=None, acquisition_plan_kwargs=None, **kwargs)[source]#

Bases: _AxAgentMixin

An asynchronous interface that uses Ax as the backend for optimization and experiment tracking and the bluesky-queueserver-api for scheduling plan execution.

Parameters:

re_manager_apiREManagerAPI: The manager API for interaction with Bluesky queueserver.
zmq_consumer_addrstr: A ZMQ address to consume Bluesky messages from, to react to plan execution on the remote server.
sensorsSequence[str]: The sensors to use for acquisition. These should be the minimal set of sensors that are needed to compute the objectives.
dofsSequence[DOF]: The degrees of freedom that the agent can control, which determine the search space.
objectivesSequence[Objective]: The objectives which the agent will try to optimize.
evaluation_functionEvaluationFunction: The function to evaluate acquired data and produce outcomes.
acquisition_planstr | None, optional: The acquisition plan to use for acquiring data from the beamline. If not provided, blop.plans.default_acquire() will be assumed.
dof_constraintsSequence[DOFConstraint] | None, optional: Constraints on DOFs to refine the search space.
outcome_constraintsSequence[OutcomeConstraint] | None, optional: Constraints on outcomes to be satisfied during optimization.
checkpoint_pathstr | None, optional: The path to the checkpoint file to save the optimizer’s state to.
**kwargsAny: Additional keyword arguments to configure the Ax experiment.

See also

blop.protocols.Sensor: The protocol for sensors.
blop.ax.dof.RangeDOF: For continuous parameters.
blop.ax.dof.ChoiceDOF: For discrete parameters.
blop.ax.objective.Objective: For defining objectives.
blop.ax.optimizer.AxOptimizer: The optimizer used internally.
blop.queueserver.QueueserverOptimizatonRunner: Runner that handles interaction with bluesky-queueserver.

property evaluation_function: EvaluationFunction#

property actuators: Sequence[str]#

property sensors: Sequence[str]#

property acquisition_plan: str | None#

stop()[source]#

property current_iteration: int#

to_optimization_problem()[source]#

property ax_client: Client#

checkpoint()#

Save the agent’s state to a JSON file.

property checkpoint_path: str | None#

property fixed_dofs: dict[str, Any] | None#

get_best_points()#

Get a list of the optimal points found during optimization.

For single-objective optimization, returns a single best point. For multi-objective optimization, returns the Pareto-optimal set.

Returns:

list[tuple[int, TParameterization, TOutcome]]

Each element in the list is a tuple of:

trial index (int)
parameter values (dict)
metric values (dict, where values may be (value, sem) tuples)

See also

navigate_to_best: Plan stub to move actuators to a best point.

ingest(points)#

Ingest evaluation results into the optimizer.

Updates the optimizer’s model with new data. Can ingest both suggested points (with “_id” key) and external data (without “_id” key).

Parameters:

pointslist[dict]: A list of dictionaries, each containing outcomes for a trial. For suggested points, include the “_id” key. For external data, include DOF names and objective values, and omit “_id”.

Notes

This method is typically called automatically by optimize(). Manual usage is only needed for custom workflows or when ingesting external data.

For complete examples, see Attach external data to experiments.

plot_objective(x_dof_name, y_dof_name, objective_name, *args, **kwargs)#

Plot the predicted objective as a function of two DOFs.

Creates a contour plot showing the model’s prediction of an objective across the space defined by two DOFs. Useful for visualizing the optimization landscape.

Parameters:

x_dof_namestr: The name of the DOF to plot on the x-axis.
y_dof_namestr: The name of the DOF to plot on the y-axis.
objective_namestr: The name of the objective to plot.
*argsAny: Additional positional arguments passed to Ax’s compute_analyses.
**kwargsAny: Additional keyword arguments passed to Ax’s compute_analyses.

Returns:

list[AnalysisCard]: The computed analysis cards containing the plot data.

See also

ax.analysis.ContourPlot: Pre-built analysis for plotting objectives.
ax.analysis.AnalysisCard: Contains the raw and computed data.

reconfigure_search_space(dof_mappings)#

Update bounds or values of existing DOFs for future optimizations.

Parameters:

dof_mappingsdict[DOF, tuple[float, float] | list[float] | list[int] | list[str] | list[bool]]: Mapping of DOFs to their new search space.

run(iterations=1, n_points=1)[source]#

Start the optimization loop.

Validates the queueserver state, then begins the suggest -> acquire -> ingest cycle. This method returns immediately; the optimization runs asynchronously via callbacks.

Parameters:

iterationsint: Number of optimization iterations to run.
n_pointsint: Number of points to suggest per iteration.

Returns:

concurrent.futures.Future[OptimizationResult]: A future that resolves to an OptimizationResult when all iterations complete or when stop() is called. If an unhandled exception occurs the future will hold it and re-raise on .result().

Raises:

RuntimeError: If the queueserver environment is not ready.
ValueError: If required devices or plans are not available.

suggest(num_points=1)#

Get the next point(s) to evaluate in the search space.

Uses the Bayesian optimization algorithm to suggest promising points based on all previously acquired data. Each suggestion includes an “_id” key for tracking.

Parameters:

num_pointsint, optional: The number of points to suggest. Default is 1. Higher values enable batch optimization but may reduce optimization efficiency per iteration.

Returns:

list[dict]: A list of dictionaries, each containing a parameterization of a point to evaluate next. Each dictionary includes an “_id” key for identification.

submit_suggestions(suggestions)[source]#

Evaluate specific parameter combinations.

Acquires data for given suggestions and ingests results. Supports both optimizer suggestions and manual points.

Parameters:

suggestionslist[dict]: Either optimizer suggestions (with “_id”) or manual points (without “_id”).

Returns:

concurrent.futures.Future[OptimizationResult]: A future that resolves to an OptimizationResult when the acquisition completes.

See also

run: Run the full optimization loop.