diff --git a/docs/source/how-to-guides.rst b/docs/source/how-to-guides.rst index e4a1780b..68e06f35 100644 --- a/docs/source/how-to-guides.rst +++ b/docs/source/how-to-guides.rst @@ -7,6 +7,7 @@ How-to Guides how-to-guides/use-ophyd-devices.rst how-to-guides/attach-data-to-experiments.rst how-to-guides/custom-generation-strategies.rst + how-to-guides/interactive-optimization.rst how-to-guides/set-dof-constraints.rst how-to-guides/set-outcome-constraints.rst how-to-guides/acquire-baseline.rst diff --git a/docs/source/how-to-guides/interactive-optimization.rst b/docs/source/how-to-guides/interactive-optimization.rst new file mode 100644 index 00000000..16b1fc80 --- /dev/null +++ b/docs/source/how-to-guides/interactive-optimization.rst @@ -0,0 +1,206 @@ +Interactive Optimization +======================== + +This guide explains how to use Blop's interactive optimization mode, which gives you fine-grained control over the optimization process through user prompts and manual interventions. + +Overview +-------- + +Interactive optimization allows you to: + +- Manually approve suggested points before they are evaluated +- Suggest custom points with known objective values during optimization +- Decide when to continue or stop the optimization + +The Interactive Optimization Flow +---------------------------------- + +When you run an optimization, the system follows this workflow: + +.. code-block:: text + + Would you like to go interactively? + ├─ No: Run optimize_normal (automatic mode) + │ └─ Completes all iterations automatically + │ + └─ Yes: Interactive mode + │ + 1. Manually approve suggestions? + | ├─ No: Use optimize_step (automated suggestions) + | │ └─ Go to step 2 (post-iteration options) + | │ + | └─ Yes: Manual approval mode + | │ + | a. How many steps before next approval? (x) + | │ + | b. For each iteration: + | | - Suggest point + | | - Ask: "Do you approve this point?" + | | - If Yes → evaluate point + | | - If No → abandon point, suggest new one + | │ + | c. After x iterations complete + | └─ Go to step 2 + | + 2. What would you like to do? + ├─ c: Continue optimization (no manual suggestions) + │ └─ Return to step 1 + │ + ├─ s: Suggest points manually + │ └─ Enter DOF values and objective values + │ └─ Ingest into model + │ └─ Return to step 1 + │ + └─ q: Quit optimization + +Starting an Interactive Optimization +------------------------------------- + +To start an interactive optimization, simply run the ``optimize`` method after defining your agent: + +.. code-block:: python + + RE(agent.optimize(iterations=10, n_points=1)) + +Initial Prompt +~~~~~~~~~~~~~~ + +When you start the optimization, you'll see: + +.. code-block:: text + + +----------------------------------------------------------+ + | Would you like to run the optimization in interactive | + | mode? | + +----------------------------------------------------------+ + y: Yes + n: No + + Enter choice [y,n]: + +- Choose ``y`` for interactive mode with full control +- Choose ``n`` for automatic mode (runs all iterations without prompts) + +Automatic Mode (Non-Interactive) +--------------------------------- + +If you choose ``n`` (No) at the initial prompt, the optimization runs in automatic mode: + +- All iterations execute without user intervention +- Points are suggested, evaluated, and ingested automatically + +Manual Approval Mode +-------------------- + +If you choose ``y`` (Yes) for interactive mode, you'll be asked: + +.. code-block:: text + + +----------------------------------------------------------+ + | Would you like to manually approve suggestions? | + +----------------------------------------------------------+ + y: Yes + n: No + + Enter choice [y,n]: + +Choosing Manual Approval +~~~~~~~~~~~~~~~~~~~~~~~~~ + +If you choose ``y`` (Yes), you'll then be asked: + +.. code-block:: text + + +----------------------------------------------------------+ + | Number of steps before next approval | + +----------------------------------------------------------+ + > + +Enter how often you would like to be able to manually approve a suggested point. If for a suggested point, manual approval is given, you'll see: + +.. code-block:: text + + Do you approve this point {'x1': 2.34, 'x2': -1.56, '_id': 5}? (y/n): + +**Note:** The first 5 points are automatically approved to build an initial model. + +- Enter ``y`` to evaluate this point +- Enter ``n`` to abandon this point (it won't be evaluated, and will be marked as ``abandoned``) + +Automated Suggestions (No Manual Approval) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +If you choose ``n`` (No) for manual approval, the optimizer will: + +1. Generate suggestions automatically +2. Evaluate all points without asking for approval +3. Proceed to post-iteration options (see below) + +Post-Iteration Options +---------------------- + +After completing the iteration(s), you'll be prompted: + +.. code-block:: text + + +----------------------------------------------------------+ + | What would you like to do? | + +----------------------------------------------------------+ + c: continue optimization without suggestion + s: suggest points manually + q: quit optimization + + Enter choice [c,s,q]: + +Option c: Continue Optimization +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Continues with another round of optimization iterations. The system will ask again if you want manual approval for the next round. + +Use this when: + +- You want to run more iterations +- You're satisfied with the current model and want to let it explore more + +Option s: Suggest Points Manually +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Allows you to manually input points with known objective values. This is useful when: + +- You have external data to add to the model +- You want to guide the optimization to specific regions +- You've performed experiments outside of Blop and want to incorporate the results + +When you choose this option, you'll be prompted to enter values for each DOF and objective: + +.. code-block:: text + + Enter value for x1 (float): 2.5 + Enter value for x2 (float): 1.3 + Enter value for my_objective (float): 42.7 + + +----------------------------------------------------------+ + | Do you want to suggest another point? | + +----------------------------------------------------------+ + y: Yes + n: No, finish suggestions + + Enter choice [y,n]: + +If you enter an invalid number (like "abc"), you'll see: + +.. code-block:: text + + Invalid input. Please enter a valid number for x1. + +And you'll be asked to try again for that specific parameter. + +Option q: Quit Optimization +~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Ends the optimization immediately + +See Also +-------- + +- :doc:`/tutorials/simple-experiment` - Basic optimization tutorial diff --git a/src/blop/ax/agent.py b/src/blop/ax/agent.py index 687e382e..e1bf5b8b 100644 --- a/src/blop/ax/agent.py +++ b/src/blop/ax/agent.py @@ -215,7 +215,7 @@ def ingest(self, points: list[dict]) -> None: points : list[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". + objective values, and omit "_id".difference between master and main in github Notes ----- diff --git a/src/blop/plans/__init__.py b/src/blop/plans/__init__.py index fc196814..ce465ed5 100644 --- a/src/blop/plans/__init__.py +++ b/src/blop/plans/__init__.py @@ -4,11 +4,12 @@ default_acquire, optimize, optimize_step, + optimize_step_with_approval, per_step_background_read, read, take_reading_with_background, ) -from .utils import get_route_index, route_suggestions +from .utils import get_route_index, retrieve_suggestions_from_user, route_suggestions __all__ = [ "acquire_baseline", @@ -17,6 +18,8 @@ "get_route_index", "optimize", "optimize_step", + "optimize_step_with_approval", + "retrieve_suggestions_from_user", "per_step_background_read", "read", "route_suggestions", diff --git a/src/blop/plans/plans.py b/src/blop/plans/plans.py index a1efa157..cdbd97ac 100644 --- a/src/blop/plans/plans.py +++ b/src/blop/plans/plans.py @@ -1,5 +1,7 @@ import functools import logging +import os +import sys from collections.abc import Callable, Mapping, Sequence from typing import Any, cast @@ -9,7 +11,7 @@ from bluesky.utils import MsgGenerator, plan from ..protocols import ID_KEY, Actuator, Checkpointable, OptimizationProblem, Sensor -from .utils import route_suggestions +from .utils import ask_user_for_input, retrieve_suggestions_from_user, route_suggestions logger = logging.getLogger(__name__) @@ -122,7 +124,62 @@ def optimize_step( @plan -def optimize( +def optimize_step_with_approval( + optimization_problem: OptimizationProblem, + n_points: int = 1, + n_steps: int = 1, + *args: Any, + **kwargs: Any, +) -> MsgGenerator[None]: + """ + A single step of the optimization loop with manual approval of suggestions. + + Parameters + ---------- + optimization_problem : OptimizationProblem + The optimization problem to solve. + n_points : int, optional + The number of points to suggest. + n_steps : int, optional + The number of steps before next manual approval. + """ + if optimization_problem.acquisition_plan is None: + acquisition_plan = default_acquire + else: + acquisition_plan = optimization_problem.acquisition_plan + optimizer = optimization_problem.optimizer + actuators = optimization_problem.actuators + suggestions = optimizer.suggest(n_points) + + approved_suggestions = [] + for suggestion in suggestions: + trial_id = suggestion.get(ID_KEY) + + # Auto-approve first 5 points to build initial model + if trial_id is not None and trial_id < 5: + approved_suggestions.append(suggestion) + continue + + # Ask for manual approval for subsequent points + if trial_id is not None and trial_id % n_steps == 0: + if input(f"Do you approve this point: {suggestion}? (y/n): ").strip().lower() == "y": + approved_suggestions.append(suggestion) + else: + optimizer.ax_client._experiment.trials[trial_id].mark_abandoned() # type: ignore[attr-defined] + else: + approved_suggestions.append(suggestion) + + if len(approved_suggestions) == 0: + print("No suggestions approved. Skipping this optimization step.") + return + + uid = yield from acquisition_plan(approved_suggestions, actuators, optimization_problem.sensors, *args, **kwargs) + outcomes = optimization_problem.evaluation_function(uid, approved_suggestions) + optimizer.ingest(outcomes) + + +@plan +def optimize_manual( optimization_problem: OptimizationProblem, iterations: int = 1, n_points: int = 1, @@ -131,7 +188,8 @@ def optimize( **kwargs: Any, ) -> MsgGenerator[None]: """ - A plan to solve the optimization problem. + A plan to solve the optimization problem that allows for manual iteraction with the optimization process + through manual approval of suggestions and the ability to manually suggest points. Parameters ---------- @@ -149,14 +207,66 @@ def optimize( Additional positional arguments to pass to the :func:`optimize_step` plan. **kwargs : Any Additional keyword arguments to pass to the :func:`optimize_step` plan. + """ + while True: + # Ask once per cycle if user wants manual approval + use_manual_approval = ask_user_for_input( + "Would you like to manually approve suggestions?", options={"y": "Yes", "n": "No"} + ) + n_steps = int(ask_user_for_input("Number of steps before next approval")) if use_manual_approval == "y" else n_points + for i in range(iterations): + if use_manual_approval == "y": + yield from optimize_step_with_approval(optimization_problem, n_points, n_steps, *args, **kwargs) + else: + yield from optimize_step(optimization_problem, n_points, *args, **kwargs) + + if checkpoint_interval and (i + 1) % checkpoint_interval == 0: + if not isinstance(optimization_problem.optimizer, Checkpointable): + raise ValueError( + "The optimizer is not checkpointable. Please review your optimizer configuration or implementation." + ) + optimization_problem.optimizer.checkpoint() + result = ask_user_for_input( + "What would you like to do?", + options={ + "c": "continue optimization without suggestion", + "s": "suggest points manually", + "q": "quit optimization", + }, + ) + if result.lower().strip() == "s": + suggestions = retrieve_suggestions_from_user( + actuators=optimization_problem.actuators, # type: ignore[attr-defined] + optimizer=optimization_problem.optimizer, # type: ignore[attr-defined] + ) + if suggestions is not None: + optimization_problem.optimizer.ingest(suggestions) + elif result.lower().strip() == "q": + break + # If result == 'c', just continue the loop (do nothing) - See Also - -------- - blop.protocols.OptimizationProblem : The problem to solve. - blop.protocols.Checkpointable : The protocol for checkpointable objects. - optimize_step : The plan to execute a single step of the optimization. + +@plan +def optimize_normal( + optimization_problem: OptimizationProblem, + iterations: int = 1, + n_points: int = 1, + checkpoint_interval: int | None = None, + *args: Any, + **kwargs: Any, +) -> MsgGenerator[None]: """ + A plan to solve the optimization problem that completes the loop automatically. + Parameters + ---------- + optimization_problem : OptimizationProblem + The optimization problem to solve. + iterations : int, optional + The number of optimization iterations to run. + n_points : int, optional + The number of points to suggest per iteration. + """ for i in range(iterations): yield from optimize_step(optimization_problem, n_points, *args, **kwargs) if checkpoint_interval and (i + 1) % checkpoint_interval == 0: @@ -167,6 +277,81 @@ def optimize( optimization_problem.optimizer.checkpoint() +def optimize( + optimization_problem: OptimizationProblem, + iterations: int = 1, + n_points: int = 1, + checkpoint_interval: int | None = None, + *args: Any, + **kwargs: Any, +) -> MsgGenerator[None]: + """ + A plan to solve the optimization problem. + + Parameters + ---------- + optimization_problem : OptimizationProblem + The optimization problem to solve. + iterations : int, optional + The number of optimization iterations to run. + n_points : int, optional + The number of points to suggest per iteration. + checkpoint_interval : int | None, optional + The number of iterations between optimizer checkpoints. If None, checkpoints + will not be saved. Optimizer must implement the + :class:`blop.protocols.Checkpointable` protocol. + *args : Any + Additional positional arguments to pass to the :func:`optimize_step` plan. + **kwargs : Any + Additional keyword arguments to pass to the :func:`optimize_step` plan. + + See Also + -------- + blop.protocols.OptimizationProblem : The problem to solve. + blop.protocols.Checkpointable : The protocol for checkpointable objects. + optimize_step : The plan to execute a single step of the optimization. + """ + + # Check if stdin is available for user input + stdin_available = True + + # Check if running under pytest or doctest + if ( + "pytest" in sys.modules + or "PYTEST_CURRENT_TEST" in os.environ + or "doctest" in sys.modules + or "_pytest.doctest" in sys.modules + ): + stdin_available = False + # if: + # Check if running in a Jupyter environment without stdin support + try: + from IPython.core.getipython import get_ipython + + ipython = get_ipython() + if ipython is not None: + # The kernel attribute only exists in IPyKernel environments + kernel = getattr(ipython, "kernel", None) + if kernel is not None and not getattr(kernel, "_allow_stdin", True): + stdin_available = False + except (ImportError, AttributeError): + pass + + # Only prompt for interactive mode if stdin is available + if stdin_available: + use_interactive = ask_user_for_input( + "Would you like to run the optimization in interactive mode?", options={"y": "Yes", "n": "No"} + ) + else: + # stdin not available - use non-interactive mode + use_interactive = False + + if use_interactive == "y": + yield from optimize_manual(optimization_problem, iterations, n_points, *args, **kwargs) + else: + yield from optimize_normal(optimization_problem, iterations, n_points, checkpoint_interval, *args, **kwargs) + + @plan def read(readables: Sequence[Readable], **kwargs: Any) -> MsgGenerator[dict[str, Any]]: """ diff --git a/src/blop/plans/utils.py b/src/blop/plans/utils.py index f33a06f2..5de89bdd 100644 --- a/src/blop/plans/utils.py +++ b/src/blop/plans/utils.py @@ -1,3 +1,5 @@ +from typing import Any + import networkx as nx import numpy as np @@ -35,3 +37,77 @@ def route_suggestions(suggestions: list[dict], starting_position: dict | None = starting_point = np.array([starting_position[dim] for dim in dims_to_route]) if starting_position else None return [suggestions[i] for i in get_route_index(points=points, starting_point=starting_point)] + + +def ask_user_for_input(prompt: str, options: dict | None = None) -> Any: + BOLD = "\033[1m" + BLUE = "\033[94m" + RESET = "\033[0m" + + print() + print(f"{BOLD}{BLUE}") + print("+" + "-" * max((len(prompt) + 2 * 2), 58) + "+") + print(f"| {prompt}".ljust(max((len(prompt) + 2 * 2), 58)) + " |") + print("+" + "-" * max((len(prompt) + 2 * 2), 58) + "+") + print(RESET, end="") + if options is not None: + for key, value in options.items(): + print(f" {BOLD}{key}{RESET}: {value}") + + while True: + # Build the prompt string with keys dynamically + valid_keys = list(options.keys()) + keys_prompt = ",".join(valid_keys) + choice = input(f"\nEnter choice [{keys_prompt}]: ").lower().strip() + if choice in options: + user_input = choice + break + print("Invalid selection.") + else: + user_input = input("> ") + return user_input + + +def retrieve_suggestions_from_user(actuators: list, optimizer) -> list[dict] | None: + """ + Retrieve manual point suggestions from the user. + + Parameters + ---------- + actuators : list + The actuators to suggest values for. + optimizer : Optimizer + The optimizer containing parameter and objective configurations. + + Returns + ------- + list[dict] | None + A list of suggested points as dictionaries with DOF and objective values, or None if user declined. + """ + # Get parameter configs and objectives from the optimizer + objectives = optimizer.ax_client._experiment.optimization_config.objective.metrics + model_components = actuators + objectives + + suggestions = [] + while True: + # Build a single suggestion point with all DOF and objective values + new_suggestion = {} + for item in model_components: + while True: + try: + value = float(input(f"Enter value for {item.name} (float): ")) + new_suggestion[item.name] = value + break + except ValueError: + print(f"Invalid input. Please enter a valid number for {item.name}.") + + suggestions.append(new_suggestion) + + # Ask if user wants to add more points + if ( + ask_user_for_input("Do you want to suggest another point?", options={"y": "Yes", "n": "No, finish suggestions"}) + == "n" + ): + break + + return suggestions