Skip to content

Dynamics

Dynamics (base class)

Dynamics

Bases: ABC

Base class for solver dynamics that control agent-solver interaction.

Subclasses must implement :meth:reset, :meth:step, and :meth:close.

Source code in gyozas/dynamics/dynamics.py 
 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
class Dynamics(ABC):
    """Base class for solver dynamics that control agent-solver interaction.

    Subclasses must implement :meth:`reset`, :meth:`step`, and :meth:`close`.
    """

    min_seed = 0
    max_seed = 2**31 - 1

    def __init__(self) -> None:
        self._rng = random.Random()

    def seed(self, value: int) -> None:
        """Seed the dynamics' internal RNG for reproducible SCIP randomization."""
        self._rng.seed(value)

    def set_seed_on_model(self, model: Model) -> None:
        def draw() -> int:
            return self._rng.randint(self.min_seed, self.max_seed)

        model.setParams(
            {
                "randomization/permuteconss": True,
                "randomization/permutevars": True,
                "randomization/permutationseed": draw(),
                "randomization/randomseedshift": draw(),
                "randomization/lpseed": draw(),
            }
        )

    @abstractmethod
    def reset(self, model: Model) -> tuple[bool, NDArray[np.int64] | None]:
        """Reset the dynamics for a new episode and return (done, action_set)."""
        ...

    @abstractmethod
    def step(self, action) -> tuple[bool, NDArray[np.int64] | None]:
        """Apply an action and return (done, action_set)."""
        ...

    @abstractmethod
    def close(self) -> None:
        """Release resources held by the dynamics."""
        ...

    @abstractmethod
    def add_action_reward_to_branching_tree(self, _branching_tree, _action, _reward) -> None:
        """Record an action and its reward in the branching tree visualisation."""
        ...

seed(value)

Seed the dynamics' internal RNG for reproducible SCIP randomization.

Source code in gyozas/dynamics/dynamics.py 
21
22
23
def seed(self, value: int) -> None:
    """Seed the dynamics' internal RNG for reproducible SCIP randomization."""
    self._rng.seed(value)

reset(model) abstractmethod

Reset the dynamics for a new episode and return (done, action_set).

Source code in gyozas/dynamics/dynamics.py 
39
40
41
42
@abstractmethod
def reset(self, model: Model) -> tuple[bool, NDArray[np.int64] | None]:
    """Reset the dynamics for a new episode and return (done, action_set)."""
    ...

step(action) abstractmethod

Apply an action and return (done, action_set).

Source code in gyozas/dynamics/dynamics.py 
44
45
46
47
@abstractmethod
def step(self, action) -> tuple[bool, NDArray[np.int64] | None]:
    """Apply an action and return (done, action_set)."""
    ...

close() abstractmethod

Release resources held by the dynamics.

Source code in gyozas/dynamics/dynamics.py 
49
50
51
52
@abstractmethod
def close(self) -> None:
    """Release resources held by the dynamics."""
    ...

add_action_reward_to_branching_tree(_branching_tree, _action, _reward) abstractmethod

Record an action and its reward in the branching tree visualisation.

Source code in gyozas/dynamics/dynamics.py 
54
55
56
57
@abstractmethod
def add_action_reward_to_branching_tree(self, _branching_tree, _action, _reward) -> None:
    """Record an action and its reward in the branching tree visualisation."""
    ...

BranchingDynamics

BranchingDynamics

Bases: ThreadedDynamics

Dynamics for variable branching decisions in the branch-and-bound solver.

At each step, the agent selects which fractional variable to branch on. The action set contains variable indices from SCIP's LP branching candidates.

Parameters:

Name Type Description Default
with_extra_actions list[int] | list[ExtraBranchingActions] | list[int | ExtraBranchingActions] | None

Optional list of extra action IDs (e.g. ExtraBranchingActions.SKIP) to prepend to the action set at each step.

 None
Source code in gyozas/dynamics/branching.py 
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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
class BranchingDynamics(ThreadedDynamics):
    """Dynamics for variable branching decisions in the branch-and-bound solver.

    At each step, the agent selects which fractional variable to branch on.
    The action set contains variable indices from SCIP's LP branching candidates.

    Parameters
    ----------
    with_extra_actions
        Optional list of extra action IDs (e.g. ``ExtraBranchingActions.SKIP``)
        to prepend to the action set at each step.
    """

    def __init__(
        self,
        with_extra_actions: list[int] | list[ExtraBranchingActions] | list[int | ExtraBranchingActions] | None = None,
    ) -> None:
        super().__init__()
        self.action = None
        self.branch_rule: BranchingOracle
        self.model: Model
        self.infeasible_nodes: list = []
        self.feasible_nodes: list = []
        self.current_node_id = None
        self._last_node_id = None
        self.with_extra_actions: NDArray[np.int64] | None = (
            np.array(sorted(int(x) for x in with_extra_actions), dtype=np.int64)
            if with_extra_actions is not None
            else None
        )
        self._action_set: NDArray[np.int64] | None = None

    def reset(self, model) -> tuple[bool, NDArray[np.int64] | None]:
        self._stop_thread()
        # Drop caught events and null plugin refs so the old model can be GC'd.
        # catchEvent() calls Py_INCREF(model); dropEvent() balances each one.
        self._release_plugins()
        self.done = False
        self.model = model
        self._last_node_id = None
        self.current_node_id = None
        self.obs_event.clear()
        self.action_event.clear()
        self.die_event.clear()
        self.infeasible_nodes = []
        self.feasible_nodes = []
        self.branch_rule = BranchingOracle(model, self.obs_event, self.action_event, self.die_event)
        model.includeBranchrule(
            self.branch_rule,
            "python-mostinf",
            "custom most infeasible branching rule",
            priority=10000000,
            maxdepth=-1,
            maxbounddist=1,
        )
        self._node_event_handler = _NodeEventHandler(self)
        model.includeEventhdlr(self._node_event_handler, "branching-node-events", "tracks node feasibility events")

        self._start_solve_thread(model)
        self.obs_event.wait()
        if self.done:
            return self.done, None
        action_set = self.branch_rule.obs
        if self.with_extra_actions is not None and action_set is not None and len(action_set) > 0:
            action_set = np.concatenate((self.with_extra_actions, action_set))
        self.obs_event.clear()
        self._action_set = action_set
        current_node = self.model.getCurrentNode()
        self.current_node_id = current_node.getNumber()
        return self.done, action_set

    def step(self, action) -> tuple[bool, NDArray[np.int64] | None]:
        if self._action_set is None:
            raise RuntimeError("No action set available. Call reset() first.")
        if action not in self._action_set:
            raise ValueError(f"Action {action} not in action set {self._action_set}")
        self.branch_rule.action = action
        self._last_node_id = self.current_node_id
        self.action_event.set()
        self.obs_event.wait()
        if self.done:
            self._action_set = None
            return self.done, None
        action_set = self.branch_rule.obs
        self._action_set = action_set
        self.obs_event.clear()
        current_node = self.model.getCurrentNode()
        self.current_node_id = current_node.getNumber()
        return self.done, action_set

    def close(self) -> None:
        super().close()  # _stop_thread() — joins thread so SCIP is no longer running
        self._release_plugins()

    def _release_plugins(self) -> None:
        """Drop caught events and null plugin refs so the SCIP model can be freed."""
        if hasattr(self, "_node_event_handler"):
            handler = self._node_event_handler
            if handler.model is not None:
                try:
                    handler.model.dropEvent(SCIP_EVENTTYPE.NODEINFEASIBLE, handler)
                    handler.model.dropEvent(SCIP_EVENTTYPE.NODEFEASIBLE, handler)
                except Exception:
                    pass
            handler.model = None
            handler.dynamics = None  # ty: ignore[invalid-assignment]
        if hasattr(self, "branch_rule"):
            self.branch_rule.scip = None  # ty: ignore[invalid-assignment]
            self.branch_rule.model = None

    def add_action_reward_to_branching_tree(self, _branching_tree, _action, _reward) -> None:
        node_id = self._last_node_id
        data = _branching_tree.get_node_data(node_id)
        if data is None:
            logging.error(f"Node {node_id} not found in branching tree.")
            return
        data.update({"action": _action, "reward": _reward})

NodeSelectionDynamics

NodeSelectionDynamics

Bases: ThreadedDynamics

Dynamics for node selection decisions in the branch-and-bound solver.

At each step, the agent selects which open node to explore next. The action set contains node IDs (leaves, children, and siblings).

Source code in gyozas/dynamics/node_selection.py 
 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
class NodeSelectionDynamics(ThreadedDynamics):
    """Dynamics for node selection decisions in the branch-and-bound solver.

    At each step, the agent selects which open node to explore next.
    The action set contains node IDs (leaves, children, and siblings).
    """

    def __init__(self) -> None:
        super().__init__()
        self.action = None
        self.node_selection_rule: NodeSelectionOracle
        self.model = None
        self.infeasible_nodes = []
        self.feasible_nodes = []
        self.nsteps = 0

    def reset(self, model: Model) -> tuple[bool, NDArray[np.int64] | None]:
        self._stop_thread()
        # Drop caught events and null plugin refs so the old model can be GC'd.
        # catchEvent() calls Py_INCREF(model); dropEvent() balances each one.
        self._release_plugins()
        self.done = False
        self.model = model
        self.nsteps = 0
        self.obs_event.clear()
        self.action_event.clear()
        self.die_event.clear()
        self.infeasible_nodes = []
        self.feasible_nodes = []
        self.node_selection_rule = NodeSelectionOracle(model, self.obs_event, self.action_event, self.die_event)
        model.includeNodesel(
            self.node_selection_rule,
            "python-nodesel",
            "custom node selection rule",
            stdpriority=1_000_000,
            memsavepriority=1_000_000,
        )
        self._node_event_handler = _NodeEventHandler(self)
        model.includeEventhdlr(self._node_event_handler, "nodesel-node-events", "tracks node feasibility events")

        self._start_solve_thread(model)
        self.obs_event.wait()
        if self.done:
            return self.done, None
        action_set = self.node_selection_rule.obs
        if action_set is not None and len(action_set) == 0:
            # No open nodes yet (e.g. SCIP is still in presolving); advance
            # the solver one step by sending a sentinel action.
            self.step(-1)
        self.obs_event.clear()
        return self.done, action_set

    def step(self, action) -> tuple[bool, NDArray[np.int64] | None]:
        self.node_selection_rule.action = action
        self.action_event.set()
        self.obs_event.wait()
        if self.done:
            return self.done, None
        action_set = self.node_selection_rule.obs
        self.obs_event.clear()
        self.nsteps += 1
        return self.done, action_set

    def close(self) -> None:
        super().close()  # _stop_thread() — joins thread so SCIP is no longer running
        self._release_plugins()

    def _release_plugins(self) -> None:
        """Drop caught events and null plugin refs so the SCIP model can be freed."""
        if hasattr(self, "_node_event_handler"):
            handler = self._node_event_handler
            if handler.model is not None:
                try:
                    handler.model.dropEvent(SCIP_EVENTTYPE.NODEINFEASIBLE, handler)
                    handler.model.dropEvent(SCIP_EVENTTYPE.NODEFEASIBLE, handler)
                except Exception:
                    pass
            handler.model = None
            handler.dynamics = None  # ty: ignore[invalid-assignment]
        if hasattr(self, "node_selection_rule"):
            self.node_selection_rule.scip = None  # ty: ignore[invalid-assignment]
            self.node_selection_rule.model = None

    def add_action_reward_to_branching_tree(self, _branching_tree, _action, _reward) -> None:
        node_id = _action
        data = _branching_tree.get_node_data(node_id)
        if data is None:
            return
        data.update({"order": self.nsteps, "reward": _reward})

ConfiguringDynamics

ConfiguringDynamics

Bases: Dynamics

Dynamics for algorithm configuration: set SCIP parameters then solve.

Inspired by ecole.dynamics.ConfiguringDynamics.

The episode has a single step: 1. reset(model) returns (False, None) — the agent may now choose params. 2. step(param_dict) sets each {name: value} pair on the model, calls model.optimize(), and returns (True, None).

The action is a dict[str, Any] mapping SCIP parameter names to values, e.g. {"limits/time": 60.0, "lp/threads": 4}. The action set is always None (unconstrained).

Source code in gyozas/dynamics/configuring.py 
 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
class ConfiguringDynamics(Dynamics):
    """Dynamics for algorithm configuration: set SCIP parameters then solve.

    Inspired by ``ecole.dynamics.ConfiguringDynamics``.

    The episode has a single step:
    1. ``reset(model)`` returns ``(False, None)`` — the agent may now choose params.
    2. ``step(param_dict)`` sets each ``{name: value}`` pair on the model, calls
       ``model.optimize()``, and returns ``(True, None)``.

    The action is a ``dict[str, Any]`` mapping SCIP parameter names to values,
    e.g. ``{"limits/time": 60.0, "lp/threads": 4}``.
    The action set is always ``None`` (unconstrained).
    """

    def __init__(self) -> None:
        super().__init__()
        self.model: Model | None = None
        self.done: bool = False
        self.infeasible_nodes: list = []
        self.feasible_nodes: list = []

    def __del__(self) -> None:
        self.close()

    def reset(self, model: Model) -> tuple[bool, None]:
        self.model = model
        self.done = False
        return False, None

    def step(self, action: dict) -> tuple[bool, None]:
        """Set SCIP parameters from *action* and solve the instance.

        Parameters
        ----------
        action
            Mapping of SCIP parameter names to values.
            Pass an empty dict to solve with default parameters.
        """
        if self.model is None:
            raise RuntimeError("No model available. Call reset() first.")
        if self.done:
            raise RuntimeError("Episode is already done. Call reset() first.")
        if action:
            self.model.setParams(action)
        self.model.optimize()
        self.done = True
        return True, None

    def add_action_reward_to_branching_tree(self, _branching_tree, _action, _reward) -> None:
        # Configuration is a single global action — there is no per-node decision to annotate.
        pass

    def close(self) -> None:
        self.model = None

step(action)

Set SCIP parameters from action and solve the instance.

Parameters:

Name Type Description Default
action dict

Mapping of SCIP parameter names to values. Pass an empty dict to solve with default parameters.

 required
Source code in gyozas/dynamics/configuring.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def step(self, action: dict) -> tuple[bool, None]:
    """Set SCIP parameters from *action* and solve the instance.

    Parameters
    ----------
    action
        Mapping of SCIP parameter names to values.
        Pass an empty dict to solve with default parameters.
    """
    if self.model is None:
        raise RuntimeError("No model available. Call reset() first.")
    if self.done:
        raise RuntimeError("Episode is already done. Call reset() first.")
    if action:
        self.model.setParams(action)
    self.model.optimize()
    self.done = True
    return True, None

PrimalSearchDynamics

PrimalSearchDynamics

Bases: ThreadedDynamics

Dynamics for primal solution search in the branch-and-bound solver.

Inspired by ecole.dynamics.PrimalSearchDynamics. At each step the agent provides a partial assignment (var_indices, vals) over the current pseudo-branching candidates. The dynamics tries to complete it into a feasible solution via LP probing.

Parameters:

Name Type Description Default
trials_per_node int

Number of agent interactions (probing trials) per heuristic call. -1 means unlimited (run until SCIP stops the solve).

 1
depth_freq int

Heuristic frequency: called every depth_freq nodes in depth.

 1
depth_start int

Minimum depth at which the heuristic is called.

 0
depth_stop int

Maximum depth at which the heuristic is called (-1 = no limit).

 -1
Source code in gyozas/dynamics/primal_search.py 
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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
class PrimalSearchDynamics(ThreadedDynamics):
    """Dynamics for primal solution search in the branch-and-bound solver.

    Inspired by ``ecole.dynamics.PrimalSearchDynamics``.  At each step the agent
    provides a *partial assignment* ``(var_indices, vals)`` over the current
    pseudo-branching candidates.  The dynamics tries to complete it into a feasible
    solution via LP probing.

    Parameters
    ----------
    trials_per_node
        Number of agent interactions (probing trials) per heuristic call.
        -1 means unlimited (run until SCIP stops the solve).
    depth_freq
        Heuristic frequency: called every ``depth_freq`` nodes in depth.
    depth_start
        Minimum depth at which the heuristic is called.
    depth_stop
        Maximum depth at which the heuristic is called (-1 = no limit).
    """

    def __init__(
        self,
        trials_per_node: int = 1,
        depth_freq: int = 1,
        depth_start: int = 0,
        depth_stop: int = -1,
    ) -> None:
        if trials_per_node < -1:
            raise ValueError(f"trials_per_node must be >= -1, got {trials_per_node}")
        super().__init__()
        self.trials_per_node = trials_per_node
        self.depth_freq = depth_freq
        self.depth_start = depth_start
        self.depth_stop = depth_stop

        self.model: Model
        self._oracle: PrimalSearchOracle
        self._action_set: NDArray[np.int64] | None = None
        self.infeasible_nodes: list = []
        self.feasible_nodes: list = []

    def close(self) -> None:
        super().close()  # _stop_thread() — joins thread so SCIP is no longer running
        self._release_plugins()

    def _release_plugins(self) -> None:
        """Null oracle plugin refs so the SCIP model can be freed."""
        if hasattr(self, "_oracle"):
            self._oracle.scip = None  # ty: ignore[invalid-assignment]
            self._oracle.model = None

    def reset(self, model: Model) -> tuple[bool, NDArray[np.int64] | None]:
        self._stop_thread()
        self._release_plugins()
        self.done = False
        self.model = model
        self.obs_event.clear()
        self.action_event.clear()
        self.die_event.clear()
        self._action_set = None

        if self.trials_per_node == 0:
            model.optimize()
            return True, None

        self._oracle = PrimalSearchOracle(
            model, self.obs_event, self.action_event, self.die_event, self.trials_per_node
        )
        model.includeHeur(
            self._oracle,
            name="primal-search",
            desc="agent-driven primal solution search via LP probing",
            dispchar="P",
            priority=100_000,
            freq=self.depth_freq,
            freqofs=self.depth_start,
            maxdepth=self.depth_stop,
            timingmask=_HEURTIMING_AFTERLPLOOP,
            usessubscip=False,
        )

        self._start_solve_thread(model)
        self.obs_event.wait()

        if self.done:
            return self.done, None

        action_set = self._oracle.obs
        self.obs_event.clear()
        self._action_set = action_set
        return self.done, action_set

    def step(self, action: tuple[NDArray[np.int64], NDArray[np.float64]]) -> tuple[bool, NDArray[np.int64] | None]:
        """Apply a partial assignment and advance the solve.

        Parameters
        ----------
        action
            ``(var_indices, vals)`` — lists of variable indices and their proposed
            values.  Pass ``([], [])`` to skip without fixing any variable.
        """
        if self._action_set is None:
            raise RuntimeError("No action set available. Call reset() first.")
        var_indices, vals = action
        if len(var_indices) != len(vals):
            raise ValueError(f"var_indices and vals must have the same length, got {len(var_indices)} and {len(vals)}")

        self._oracle.action = (var_indices, vals)
        self.action_event.set()
        self.obs_event.wait()

        if self.done:
            self._action_set = None
            return self.done, None

        action_set = self._oracle.obs
        self.obs_event.clear()
        self._action_set = action_set
        return self.done, action_set

    def add_action_reward_to_branching_tree(self, _branching_tree, _action, _reward) -> None:
        # PrimalSearch actions are partial variable assignments, not node decisions,
        # so there is no natural node to annotate in the branching tree.
        pass

step(action)

Apply a partial assignment and advance the solve.

Parameters:

Name Type Description Default
action tuple[NDArray[int64], NDArray[float64]]

(var_indices, vals) — lists of variable indices and their proposed values. Pass ([], []) to skip without fixing any variable.

 required
Source code in gyozas/dynamics/primal_search.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def step(self, action: tuple[NDArray[np.int64], NDArray[np.float64]]) -> tuple[bool, NDArray[np.int64] | None]:
    """Apply a partial assignment and advance the solve.

    Parameters
    ----------
    action
        ``(var_indices, vals)`` — lists of variable indices and their proposed
        values.  Pass ``([], [])`` to skip without fixing any variable.
    """
    if self._action_set is None:
        raise RuntimeError("No action set available. Call reset() first.")
    var_indices, vals = action
    if len(var_indices) != len(vals):
        raise ValueError(f"var_indices and vals must have the same length, got {len(var_indices)} and {len(vals)}")

    self._oracle.action = (var_indices, vals)
    self.action_event.set()
    self.obs_event.wait()

    if self.done:
        self._action_set = None
        return self.done, None

    action_set = self._oracle.obs
    self.obs_event.clear()
    self._action_set = action_set
    return self.done, action_set

ExtraBranchingActions

ExtraBranchingActions

Bases: IntEnum

Source code in gyozas/dynamics/branching.py 
13
14
15
16
class ExtraBranchingActions(IntEnum):
    SKIP = -1
    CUT_OFF = -2
    REDUCE_DOMAIN = -3