Write a Custom Optimization Callback

This guide shows how to add a custom callback that records optimization outcomes while an agent runs.

Use a callback when you want to observe optimization progress or send progress somewhere else, such as a file, dashboard, log service, or operator display. If you want background on how callbacks fit into Blop’s optimization workflow, see Optimization Callbacks.

Create a callback class

Subclass bluesky.callbacks.CallbackBase and implement the document methods you need. A common pattern is to use descriptor to identify fields of interest and event to record their values.

from bluesky.callbacks import CallbackBase


class OutcomeHistory(CallbackBase):
    def __init__(self):
        self.outcome_keys = []
        self.rows = []

    def descriptor(self, doc):
        self.outcome_keys = [
            name
            for name, data_key in doc.get("data_keys", {}).items()
            if data_key.get("source") == "optimization-outcome"
        ]

    def event(self, doc):
        data = doc.get("data", {})
        row = {key: data[key] for key in self.outcome_keys if key in data}
        if row:
            self.rows.append(row)

Subscribe the callback

Configure an agent, then create one callback instance and subscribe it before running the optimization.

from blop.ax import Agent, Objective, RangeDOF


agent = Agent(
    sensors=[signal],
    dofs=[RangeDOF(actuator=motor_x, bounds=(-1, 1), parameter_type="float")],
    objectives=[Objective(name="signal", minimize=False)],
    evaluation_function=evaluation_function,
)

outcome_history = OutcomeHistory()
agent.subscribe(outcome_history)

RE(agent.optimize(iterations=2))

After the run, inspect the data collected by the callback.

print(len(outcome_history.rows) == 2)

True

Keep or remove the default logger

Agents include the default console logger unless you remove it. Keep it if you want normal console output alongside your custom callback.

To run only your callback, clear the existing callbacks before subscribing.

agent.callbacks.clear()
agent.subscribe(outcome_history)

To remove a specific callback later, keep a reference to it and unsubscribe it.

agent.unsubscribe(outcome_history)

Use callback data carefully

Use callbacks for side effects: logging, notifications, dashboards, summaries, or lightweight record keeping. Do not use a callback to compute objective values, move hardware, or update the optimizer state. Put those tasks in the evaluation function, acquisition plan, or optimizer instead.