-
Notifications
You must be signed in to change notification settings - Fork 367
Expand file tree
/
Copy pathbase.py
More file actions
882 lines (787 loc) · 41.2 KB
/
base.py
File metadata and controls
882 lines (787 loc) · 41.2 KB
1
2
3
4
5
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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
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
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
# Copyright (c) Meta Platforms, Inc. and affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.
import time
import pickle
import warnings
from pathlib import Path
from numbers import Real
from collections import deque
import numpy as np
import nevergrad.common.typing as tp
from nevergrad.parametrization import parameter as p
from nevergrad.common import tools as ngtools
from nevergrad.common import errors as errors
from nevergrad.common.decorators import Registry
from . import utils
from . import multiobjective as mobj
OptCls = tp.Union["ConfiguredOptimizer", tp.Type["Optimizer"]]
registry: Registry[OptCls] = Registry()
_OptimCallBack = tp.Union[
tp.Callable[["Optimizer", "p.Parameter", float], None], tp.Callable[["Optimizer"], None]
]
X = tp.TypeVar("X", bound="Optimizer")
Y = tp.TypeVar("Y")
IntOrParameter = tp.Union[int, p.Parameter]
_PruningCallable = tp.Callable[[utils.Archive[utils.MultiValue]], utils.Archive[utils.MultiValue]]
def _loss(param: p.Parameter) -> float:
"""Returns the loss if available, or inf otherwise.
Used to simplify handling of losses
"""
return param.loss if param.loss is not None else float("inf")
def load(cls: tp.Type[X], filepath: tp.PathLike) -> X:
"""Loads a pickle file and checks that it contains an optimizer.
The optimizer class is not always fully reliable though (e.g.: optimizer families) so the user is responsible for it.
"""
filepath = Path(filepath)
with filepath.open("rb") as f:
opt = pickle.load(f)
assert isinstance(opt, cls), f"You should only load {cls} with this method (found {type(opt)})"
return opt
class Optimizer: # pylint: disable=too-many-instance-attributes
"""Algorithm framework with 3 main functions:
- :code:`ask()` which provides a candidate on which to evaluate the function to optimize.
- :code:`tell(candidate, loss)` which lets you provide the loss associated to points.
- :code:`provide_recommendation()` which provides the best final candidate.
Typically, one would call :code:`ask()` num_workers times, evaluate the
function on these num_workers points in parallel, update with the fitness value when the
evaluations is finished, and iterate until the budget is over. At the very end,
one would call provide_recommendation for the estimated optimum.
This class is abstract, it provides internal equivalents for the 3 main functions,
among which at least :code:`_internal_ask_candidate` has to be overridden.
Each optimizer instance should be used only once, with the initial provided budget
Parameters
----------
parametrization: int or Parameter
either the dimension of the optimization space, or its parametrization
budget: int/None
number of allowed evaluations
num_workers: int
number of evaluations which will be run in parallel at once
"""
# pylint: disable=too-many-locals
# optimizer qualifiers
recast = False # algorithm which were not designed to work with the suggest/update pattern
one_shot = False # algorithm designed to suggest all budget points at once
no_parallelization = False # algorithm which is designed to run sequentially only
def __init__(
self, parametrization: IntOrParameter, budget: tp.Optional[int] = None, num_workers: int = 1
) -> None:
if self.no_parallelization and num_workers > 1:
raise ValueError(f"{self.__class__.__name__} does not support parallelization")
# "seedable" random state: externally setting the seed will provide deterministic behavior
# you can also replace or reinitialize this random state
self.num_workers = int(num_workers)
self.budget = budget
self.optim_curve: tp.List[tp.Any] = []
self.skip_constraints = False
# How do we deal with cheap constraints i.e. constraints which are fast and use low resources and easy ?
# True ==> we penalize them (infinite values for candidates which violate the constraint).
# False ==> we repeat the ask until we solve the problem.
self._constraints_manager = utils.ConstraintManager()
self._penalize_cheap_violations = False
self.parametrization = (
parametrization
if not isinstance(parametrization, (int, np.int_))
else p.Array(shape=(parametrization,))
)
self.parametrization.freeze() # avoids issues!
if not self.dimension:
raise ValueError("No variable to optimize in this parametrization.")
self.name = self.__class__.__name__ # printed name in repr
# keep a record of evaluations, and current bests which are updated at each new evaluation
self.archive: utils.Archive[utils.MultiValue] = (
utils.Archive()
) # dict like structure taking np.ndarray as keys and Value as values
self.current_bests = {
x: utils.MultiValue(self.parametrization, np.inf, reference=self.parametrization)
for x in ["optimistic", "pessimistic", "average", "minimum"]
}
# pruning function, called at each "tell"
# this can be desactivated or modified by each implementation
self.pruning: tp.Optional[_PruningCallable] = utils.Pruning.sensible_default(
num_workers=num_workers, dimension=self.parametrization.dimension
)
# multiobjective
self._MULTIOBJECTIVE_AUTO_BOUND = mobj.AUTO_BOUND
self._hypervolume_pareto: tp.Optional[mobj.HypervolumePareto] = None
# instance state
self._asked: tp.Set[str] = set()
self._num_objectives = 0
self._suggestions: tp.Deque[p.Parameter] = deque()
self._num_ask = 0
self._num_tell = 0 # increases after each successful tell
self._num_tell_not_asked = 0
self._callbacks: tp.Dict[str, tp.List[tp.Any]] = {}
# to make optimize function stoppable halway through
self._running_jobs: tp.List[tp.Tuple[p.Parameter, tp.JobLike[tp.Loss]]] = []
self._finished_jobs: tp.Deque[tp.Tuple[p.Parameter, tp.JobLike[tp.Loss]]] = deque()
self._sent_warnings: tp.Set[tp.Any] = set() # no use printing several time the same
# Most optimizers are designed for single objective and use a float loss.
# To use these in a multi-objective optimization, we provide the negative of
# the hypervolume of the pareto front as the loss.
# If not needed, an optimizer can set this to True.
self._no_hypervolume = False
def _warn(self, msg: str, e: tp.Any) -> None:
"""Warns only once per warning type"""
if e not in self._sent_warnings:
warnings.warn(msg, e)
self._sent_warnings.add(e)
@property
def _rng(self) -> np.random.RandomState:
"""np.random.RandomState: parametrization random state the optimizer must pull from.
It can be seeded or updated directly on the parametrization instance (`optimizer.parametrization.random_state`)
"""
return self.parametrization.random_state
@property
def dimension(self) -> int:
"""int: Dimension of the optimization space."""
return self.parametrization.dimension
@property
def num_objectives(self) -> int:
"""Provides 0 if the number is not known yet, else the number of objectives
to optimize upon.
"""
if (
self._hypervolume_pareto is not None
and self._num_objectives != self._hypervolume_pareto.num_objectives
):
raise RuntimeError("Number of objectives is incorrectly set. Please create a nevergrad issue")
return self._num_objectives
@num_objectives.setter
def num_objectives(self, num: int) -> None:
num = int(num)
if num <= 0:
raise ValueError("Number of objectives must be strictly positive")
if not self._num_objectives:
self._num_objectives = num
self._num_objectives_set_callback()
elif num != self._num_objectives:
raise ValueError(f"Expected {self._num_objectives} loss(es), but received {num}.")
def _num_objectives_set_callback(self) -> None:
"""Callback for when num objectives is first known"""
@property
def num_ask(self) -> int:
"""int: Number of time the `ask` method was called."""
return self._num_ask
@property
def num_tell(self) -> int:
"""int: Number of time the `tell` method was called."""
return self._num_tell
@property
def num_tell_not_asked(self) -> int:
"""int: Number of time the :code:`tell` method was called on candidates that were not asked for by the optimizer
(or were suggested).
"""
return self._num_tell_not_asked
def pareto_front(
self, size: tp.Optional[int] = None, subset: str = "random", subset_tentatives: int = 12
) -> tp.List[p.Parameter]:
"""Pareto front, as a list of Parameter. The losses can be accessed through
parameter.losses
Parameters
------------
size: int (optional)
if provided, selects a subset of the full pareto front with the given maximum size
subset: str
method for selecting the subset ("random, "loss-covering", "domain-covering", "hypervolume")
subset_tentatives: int
number of random tentatives for finding a better subset
Returns
--------
list
the list of Parameter of the pareto front
Note
----
During non-multiobjective optimization, this returns the current pessimistic best
"""
pareto = (
[]
if self._hypervolume_pareto is None
else self._hypervolume_pareto.pareto_front(
size=size, subset=subset, subset_tentatives=subset_tentatives
)
)
return pareto if pareto else [self.provide_recommendation()]
def dump(self, filepath: tp.Union[str, Path]) -> None:
"""Pickles the optimizer into a file."""
filepath = Path(filepath)
with filepath.open("wb") as f:
pickle.dump(self, f)
@classmethod
def load(cls: tp.Type[X], filepath: tp.Union[str, Path]) -> X:
"""Loads a pickle and checks that the class is correct."""
return load(cls, filepath)
def __repr__(self) -> str:
inststr = self.parametrization.name
return f"Instance of {self.name}(parametrization={inststr}, budget={self.budget}, num_workers={self.num_workers})"
def register_callback(self, name: str, callback: _OptimCallBack) -> None:
"""Add a callback method called either when `tell` or `ask` are called, with the same
arguments (including the optimizer / self). This can be useful for custom logging.
Parameters
----------
name: str
name of the method to register the callback for (either :code:`ask` or :code:`tell`)
callback: callable
a callable taking the same parameters as the method it is registered upon (including self)
"""
assert name in ["ask", "tell"], f'Only "ask" and "tell" methods can have callbacks (not {name})'
self._callbacks.setdefault(name, []).append(callback)
def remove_all_callbacks(self) -> None:
"""Removes all registered callables"""
self._callbacks = {}
def suggest(self, *args: tp.Any, **kwargs: tp.Any) -> None:
"""Suggests a new point to ask.
It will be asked at the next call (last in first out).
Parameters
----------
*args: Any
positional arguments matching the parametrization pattern.
*kwargs: Any
keyword arguments matching the parametrization pattern.
Note
----
- This relies on optmizers implementing a way to deal with unasked candidate.
Some optimizers may not support it and will raise a :code:`TellNotAskedNotSupportedError`
at :code:`tell` time.
- LIFO is used so as to be able to suggest and ask straightaway, as an alternative to
creating a new candidate with :code:`optimizer.parametrization.spawn_child(new_value)`
"""
if isinstance(self.parametrization, p.Instrumentation):
new_value: tp.Any = (args, kwargs)
else:
assert len(args) == 1 and not kwargs
new_value = args[0]
self._suggestions.append(self.parametrization.spawn_child(new_value=new_value))
# pylint: disable=too-many-branches
def tell(
self,
candidate: p.Parameter,
loss: tp.Loss,
constraint_violation: tp.Optional[tp.Loss] = None,
penalty_style: tp.Optional[tp.ArrayLike] = None,
) -> None:
"""Provides the optimizer with the evaluation of a fitness value for a candidate.
Parameters
----------
x: np.ndarray
point where the function was evaluated
loss: float/list/np.ndarray
loss of the function (or multi-objective function
constraint_violation: float/list/np.ndarray/None
constraint violation (> 0 means that this is not correct)
penalty_style: ArrayLike/None
to be read as [a,b,c,d,e,f]
with cv the constraint violation vector (above):
penalty = (a + sum(|loss|)) * (f+num_tell)**e * (b * sum(cv**c)) ** d
default: [1e5, 1., .5, 1., .5, 1.]
Note
----
The candidate should generally be one provided by :code:`ask()`, but can be also
a non-asked candidate. To create a p.Parameter instance from args and kwargs,
you can use :code:`candidate = optimizer.parametrization.spawn_child(new_value=your_value)`:
- for an :code:`Array(shape(2,))`: :code:`optimizer.parametrization.spawn_child(new_value=[12, 12])`
- for an :code:`Instrumentation`: :code:`optimizer.parametrization.spawn_child(new_value=(args, kwargs))`
Alternatively, you can provide a suggestion with :code:`optimizer.suggest(*args, **kwargs)`, the next :code:`ask`
will use this suggestion.
"""
# Check loss type
if isinstance(loss, (Real, float)) or (isinstance(loss, np.ndarray) and not loss.shape):
# using "float" along "Real" because mypy does not understand "Real" for now Issue #3186
loss = float(loss)
# Non-sense values including NaNs should not be accepted.
# We do not use max-float as various later transformations could lead to greater values.
if not loss < 5.0e20: # pylint: disable=unneeded-not
self._warn(
f"Clipping very high value {loss} in tell (rescale the cost function?).",
errors.LossTooLargeWarning,
)
loss = 5.0e20 # sys.float_info.max leads to numerical problems so let us do this.
elif isinstance(loss, (tuple, list, np.ndarray)):
loss = np.asarray(loss, dtype=float).ravel() if len(loss) != 1 else loss[0]
elif not isinstance(loss, np.ndarray):
raise TypeError(
f'"tell" method only supports float values but the passed loss was: {loss} (type: {type(loss)}.'
)
if isinstance(loss, float) and (
len(self.optim_curve) == 0 or self.num_tell > self.optim_curve[-1][0] * 1.1
):
self.optim_curve += [(self.num_tell, loss)]
# check Parameter
if not isinstance(candidate, p.Parameter):
raise TypeError(
"'tell' must be provided with the candidate.\n"
"Use optimizer.parametrization.spawn_child(new_value)) if you want to "
"create a candidate that as not been asked for, "
"or optimizer.suggest(*args, **kwargs) to suggest a point that should be used for "
"the next ask"
)
# check loss length
self.num_objectives = 1 if isinstance(loss, float) else loss.size
# checks are done, start processing
candidate.freeze() # make sure it is not modified somewhere
# add reference if provided
if isinstance(candidate, p.MultiobjectiveReference):
if self._hypervolume_pareto is not None:
raise RuntimeError("MultiobjectiveReference can only be provided before the first tell.")
if not isinstance(loss, np.ndarray):
raise RuntimeError("MultiobjectiveReference must only be used for multiobjective losses")
self._hypervolume_pareto = mobj.HypervolumePareto(
upper_bounds=loss, seed=self._rng, no_hypervolume=self._no_hypervolume
)
if candidate.value is None:
return # no value, so stopping processing there
candidate = candidate.value
# preprocess multiobjective loss
if isinstance(loss, np.ndarray):
candidate._losses = loss
if not isinstance(loss, float):
loss = self._preprocess_multiobjective(candidate)
# call callbacks for logging etc...
candidate.loss = loss
assert isinstance(loss, float)
for callback in self._callbacks.get("tell", []):
# multiobjective reference is not handled :s
# but this allows obtaining both scalar and multiobjective loss (through losses)
callback(self, candidate, loss)
no_update = False
if not candidate.satisfies_constraints(self.parametrization) and self.budget is not None:
penalty = self._constraints_manager.penalty(candidate, self.num_ask, self.budget)
no_update = True
loss = loss + penalty
if constraint_violation is not None:
if penalty_style is not None:
a, b, c, d, e, f = penalty_style
else:
a, b, c, d, e, f = (1e5, 1.0, 0.5, 1.0, 0.5, 1.0)
ratio = 1 if self.budget is not None and self._num_tell > self.budget / 2.0 else 0.0
iviolation = np.sum(np.maximum(constraint_violation, 0.0))
if iviolation > 0.0:
no_update = True
violation = float(
(a * ratio + np.sum(np.maximum(loss, 0.0)))
* ((f + self._num_tell) ** e)
* (b * np.sum(np.maximum(constraint_violation, 0.0) ** c) ** d)
)
loss += violation
if (
isinstance(loss, float)
and (self.num_objectives == 1 or self.num_objectives > 1 and not self._no_hypervolume)
and not no_update
):
self._update_archive_and_bests(candidate, loss)
if candidate.uid in self._asked:
self._internal_tell_candidate(candidate, loss)
self._asked.remove(candidate.uid)
else:
self._internal_tell_not_asked(candidate, loss)
self._num_tell_not_asked += 1
self._num_tell += 1
def _preprocess_multiobjective(self, candidate: p.Parameter) -> tp.FloatLoss:
if self._hypervolume_pareto is None:
self._hypervolume_pareto = mobj.HypervolumePareto(
auto_bound=self._MULTIOBJECTIVE_AUTO_BOUND, no_hypervolume=self._no_hypervolume
)
return self._hypervolume_pareto.add(candidate)
def _update_archive_and_bests(self, candidate: p.Parameter, loss: tp.FloatLoss) -> None:
x = candidate.get_standardized_data(reference=self.parametrization)
if not isinstance(
loss, (Real, float)
): # using "float" along "Real" because mypy does not understand "Real" for now Issue #3186
raise TypeError(
f'"tell" method only supports float values but the passed loss was: {loss} (type: {type(loss)}.'
)
if np.isnan(loss) or loss == np.inf:
self._warn(f"Updating fitness with {loss} value", errors.BadLossWarning)
mvalue: tp.Optional[utils.MultiValue] = None
if x not in self.archive:
self.archive[x] = utils.MultiValue(candidate, loss, reference=self.parametrization)
else:
mvalue = self.archive[x]
mvalue.add_evaluation(loss)
# both parameters should be non-None
if mvalue.parameter.loss > candidate.loss: # type: ignore
mvalue.parameter = candidate # keep best candidate
# update current best records
# this may have to be improved if we want to keep more kinds of best losss
for name in self.current_bests:
if mvalue is self.current_bests[name]: # reboot
best = min(self.archive.values(), key=lambda mv, n=name: mv.get_estimation(n)) # type: ignore
# rebuild best point may change, and which value did not track the updated value anyway
self.current_bests[name] = best
else:
if self.archive[x].get_estimation(name) <= self.current_bests[name].get_estimation(name):
self.current_bests[name] = self.archive[x]
# deactivated checks
# if not (np.isnan(loss) or loss == np.inf):
# if not self.current_bests[name].x in self.archive:
# bval = self.current_bests[name].get_estimation(name)
# avals = (min(v.get_estimation(name) for v in self.archive.values()),
# max(v.get_estimation(name) for v in self.archive.values()))
# raise RuntimeError(f"Best value should exist in the archive at num_tell={self.num_tell})\n"
# f"Best value is {bval} and archive is within range {avals} for {name}")
if self.pruning is not None:
self.archive = self.pruning(self.archive)
def ask(self) -> p.Parameter:
"""Provides a point to explore.
This function can be called multiple times to explore several points in parallel
Returns
-------
p.Parameter:
The candidate to try on the objective function. :code:`p.Parameter` have field :code:`args` and :code:`kwargs`
which can be directly used on the function (:code:`objective_function(*candidate.args, **candidate.kwargs)`).
"""
# call callbacks for logging etc...
for callback in self._callbacks.get("ask", []):
callback(self)
current_num_ask = self.num_ask
# tentatives if a cheap constraint is available
max_trials = max(1, self._constraints_manager.max_trials // 2)
# half will be used for sub-optimization --- if the optimization method does not need/use a budget.
# TODO(oteytaud): actually we could do this even when the budget is known, if we are sure that
# exceeding the budget is not a problem.
# Very simple constraint solver:
# - we use a simple algorithm.
# - no memory of previous iterations.
# - just projection to constraint satisfaction.
# We try using the normal tool during half constraint budget, in order to reduce the impact on the normal run.
self.parametrization.tabu_fails = 0
if self.skip_constraints and not self._suggestions:
candidate = self._internal_ask_candidate()
is_suggestion = False
else:
for _ in range(max_trials):
is_suggestion = False
if self._suggestions: # use suggestions if available
is_suggestion = True
candidate = self._suggestions.pop()
else:
try: # Sometimes we have a limited budget so that
candidate = self._internal_ask_candidate()
except AssertionError as e:
assert (
self.parametrization._constraint_checkers
), f"Error: {e}" # This should not happen without constraint issues.
candidate = self.parametrization.spawn_child()
if candidate.satisfies_constraints(self.parametrization):
if self._num_ask % 10 == 0:
if candidate.can_skip_constraints(self.parametrization):
self.skip_constraints = True
break # good to go!
if self._penalize_cheap_violations or self.no_parallelization:
# Warning! This might be a tell not asked.
self._internal_tell_candidate(candidate, float("Inf")) # DE requires a tell
# updating num_ask is necessary for some algorithms which need new num to ask another point
self._num_ask += 1
satisfies = candidate.satisfies_constraints(self.parametrization)
if not satisfies and self.parametrization.tabu_length == 0:
# still not solving, let's run sub-optimization
candidate = _constraint_solver(candidate, budget=max_trials)
if not (satisfies or candidate.satisfies_constraints(self.parametrization, no_tabu=True)):
self._warn(
f"Could not bypass the constraint after {max_trials} tentatives, "
"sending candidate anyway.",
errors.FailedConstraintWarning,
)
if not is_suggestion:
if candidate.uid in self._asked:
raise RuntimeError(
"Cannot submit the same candidate twice: please recreate a new candidate from data.\n"
"This is to make sure that stochastic parametrizations are resampled."
)
self._asked.add(candidate.uid)
self._num_ask = current_num_ask + 1
assert (
candidate is not None
), f"{self.__class__.__name__}._internal_ask method returned None instead of a point."
# make sure to call value getter which may update the value, before we freeze the paremeter
candidate.value # pylint: disable=pointless-statement
candidate.freeze() # make sure it is not modified somewhere
return candidate
def provide_recommendation(self) -> p.Parameter:
"""Provides the best point to use as a minimum, given the budget that was used
Returns
-------
p.Parameter
The candidate with minimal value. p.Parameters have field :code:`args` and :code:`kwargs` which can be directly used
on the function (:code:`objective_function(*candidate.args, **candidate.kwargs)`).
"""
return self.recommend() # duplicate method
def recommend(self) -> p.Parameter:
"""Provides the best candidate to use as a minimum, given the budget that was used.
Returns
-------
p.Parameter
The candidate with minimal loss. :code:`p.Parameters` have field :code:`args` and :code:`kwargs` which can be directly used
on the function (:code:`objective_function(*candidate.args, **candidate.kwargs)`).
"""
if self.num_objectives > 1:
raise RuntimeError(
"No best candidate in MOO. Use optimizer.pareto_front() instead to get the set of all non-dominated candidates."
)
recom_data = self._internal_provide_recommendation() # pylint: disable=assignment-from-none
if recom_data is None or any(np.isnan(recom_data)):
name = "minimum" if self.parametrization.function.deterministic else "pessimistic"
return self.current_bests[name].parameter
out = self.parametrization.spawn_child()
with p.helpers.deterministic_sampling(out):
out.set_standardized_data(recom_data)
return out
def _internal_tell_not_asked(self, candidate: p.Parameter, loss: tp.FloatLoss) -> None:
"""Called whenever calling :code:`tell` on a candidate that was not "asked".
Defaults to the standard tell pipeline.
"""
self._internal_tell_candidate(candidate, loss)
def _internal_tell_candidate(self, candidate: p.Parameter, loss: tp.FloatLoss) -> None:
"""Called whenever calling :code:`tell` on a candidate that was "asked"."""
data = candidate.get_standardized_data(reference=self.parametrization)
self._internal_tell(data, loss)
def _internal_ask_candidate(self) -> p.Parameter:
return self.parametrization.spawn_child().set_standardized_data(self._internal_ask())
# Internal methods which can be overloaded (or must be, in the case of _internal_ask)
def _internal_tell(self, x: tp.ArrayLike, loss: tp.FloatLoss) -> None:
pass
def _internal_ask(self) -> tp.ArrayLike:
raise RuntimeError("Not implemented, should not be called.")
def _internal_provide_recommendation(self) -> tp.Optional[tp.ArrayLike]:
"""Override to provide a recommendation in standardized space"""
return None
def enable_pickling(self) -> None:
"""
Some optimizers are only optionally picklable, because picklability
requires saving the whole history which would be a waste of memory
in general. To tell an optimizer to be picklable, call this function
before any asks.
In this base class, the function is a no-op, but it is overridden
in some optimizers.
"""
def minimize(
self,
objective_function: tp.Callable[..., tp.Loss],
executor: tp.Optional[tp.ExecutorLike] = None,
batch_mode: bool = False,
verbosity: int = 0,
constraint_violation: tp.Any = None,
max_time: tp.Optional[float] = None,
) -> p.Parameter:
"""Optimization (minimization) procedure
Parameters
----------
objective_function: callable
A callable to optimize (minimize)
executor: Executor
An executor object, with method :code:`submit(callable, *args, **kwargs)` and returning a Future-like object
with methods :code:`done() -> bool` and :code:`result() -> float`. The executor role is to dispatch the execution of
the jobs locally/on a cluster/with multithreading depending on the implementation.
Eg: :code:`concurrent.futures.ProcessPoolExecutor`
batch_mode: bool
when :code:`num_workers = n > 1`, whether jobs are executed by batch (:code:`n` function evaluations are launched,
we wait for all results and relaunch n evals) or not (whenever an evaluation is finished, we launch
another one)
verbosity: int
print information about the optimization (0: None, 1: fitness values, 2: fitness values and recommendation)
constraint_violation: list of functions or None
each function in the list returns >0 for a violated constraint.
Returns
-------
ng.p.Parameter
The candidate with minimal value. :code:`ng.p.Parameters` have field :code:`args` and :code:`kwargs` which can
be directly used on the function (:code:`objective_function(*candidate.args, **candidate.kwargs)`).
Note
----
for evaluation purpose and with the current implementation, it is better to use batch_mode=True
"""
# pylint: disable=too-many-branches
if self.budget is None:
raise ValueError("Budget must be specified")
if executor is None:
executor = utils.SequentialExecutor() # defaults to run everything locally and sequentially
if self.num_workers > 1:
self._warn(
f"num_workers = {self.num_workers} > 1 is suboptimal when run sequentially",
errors.InefficientSettingsWarning,
)
assert executor is not None
tmp_runnings: tp.List[tp.Tuple[p.Parameter, tp.JobLike[tp.Loss]]] = []
tmp_finished: tp.Deque[tp.Tuple[p.Parameter, tp.JobLike[tp.Loss]]] = deque()
# go
sleeper = ngtools.Sleeper() # manages waiting time depending on execution time of the jobs
remaining_budget = self.budget - self.num_ask
first_iteration = True
#
t0 = time.time()
while (remaining_budget or self._running_jobs or self._finished_jobs) and (
max_time is None or time.time() < t0 + max_time
):
# # # # # Update optimizer with finished jobs # # # # #
# this is the first thing to do when resuming an existing optimization run
# process finished
if self._finished_jobs:
if (remaining_budget or sleeper._start is not None) and not first_iteration:
# ignore stop if no more suggestion is sent
# this is an ugly hack to avoid warnings at the end of steady mode
sleeper.stop_timer()
while self._finished_jobs:
x, job = self._finished_jobs[0]
result = job.result()
if constraint_violation is not None:
self.tell(
x, result, [f(x.value) for f in constraint_violation]
) # TODO: this is not parallelized, wtf!
else:
self.tell(x, result)
self._finished_jobs.popleft() # remove it after the tell to make sure it was indeed "told" (in case of interruption)
if verbosity:
print(f"Updating fitness with value {job.result()}")
if verbosity:
print(f"{remaining_budget} remaining budget and {len(self._running_jobs)} running jobs")
if verbosity > 1:
print("Current pessimistic best is: {}".format(self.current_bests["pessimistic"]))
elif not first_iteration:
sleeper.sleep()
# # # # # Start new jobs # # # # #
if not batch_mode or not self._running_jobs:
new_sugg = max(0, min(remaining_budget, self.num_workers - len(self._running_jobs)))
if verbosity and new_sugg:
print(f"Launching {new_sugg} jobs with new suggestions")
for _ in range(new_sugg):
try:
args = self.ask()
except errors.NevergradEarlyStopping:
remaining_budget = 0
break
self._running_jobs.append(
(args, executor.submit(objective_function, *args.args, **args.kwargs))
)
if new_sugg:
sleeper.start_timer()
if remaining_budget > 0: # early stopping sets it to 0
remaining_budget = self.budget - self.num_ask
# split (repopulate finished and runnings in only one loop to avoid
# weird effects if job finishes in between two list comprehensions)
tmp_runnings, tmp_finished = [], deque()
for x_job in self._running_jobs:
(tmp_finished if x_job[1].done() else tmp_runnings).append(x_job)
self._running_jobs, self._finished_jobs = tmp_runnings, tmp_finished
first_iteration = False
return self.provide_recommendation() if self.num_objectives == 1 else p.Constant(None)
def _info(self) -> tp.Dict[str, tp.Any]:
"""Easy access to debug/benchmark info"""
return {}
# Adding a comparison-only functionality to an optimizer.
def addCompare(optimizer: Optimizer) -> None:
def compare(self: Optimizer, winners: tp.List[p.Parameter], losers: tp.List[p.Parameter]) -> None:
# This means that for any i and j, winners[i] is better than winners[i+1], and better than losers[j].
# This is for cases in which we do not know fitness values, we just know comparisons.
ref = self.parametrization
# Evaluate the best fitness value among losers.
best_fitness_value = 0.0
for candidate in losers:
data = candidate.get_standardized_data(reference=self.parametrization)
if data in self.archive:
best_fitness_value = min(best_fitness_value, self.archive[data].get_estimation("average"))
# Now let us decide the fitness value of winners.
for i, candidate in enumerate(winners):
self.tell(candidate, best_fitness_value - len(winners) + i)
data = candidate.get_standardized_data(reference=self.parametrization)
self.archive[data] = utils.MultiValue(
candidate, best_fitness_value - len(winners) + i, reference=ref
)
setattr(optimizer.__class__, "compare", compare)
class ConfiguredOptimizer:
"""Creates optimizer-like instances with configuration.
Parameters
----------
OptimizerClass: type
class of the optimizer to configure, or another ConfiguredOptimizer (config will then be ignored
except for the optimizer name/representation)
config: dict
dictionnary of all the configurations
as_config: bool
whether to provide all config as kwargs to the optimizer instantiation (default, see ConfiguredCMA for an example),
or through a config kwarg referencing self. (if True, see EvolutionStrategy for an example)
Note
----
This provides a default repr which can be bypassed through set_name
"""
# optimizer qualifiers
recast = False # algorithm which were not designed to work with the suggest/update pattern
one_shot = False # algorithm designed to suggest all budget points at once
no_parallelization = False # algorithm which is designed to run sequentially only
def __init__(self, OptimizerClass: OptCls, config: tp.Dict[str, tp.Any], as_config: bool = False) -> None:
self._OptimizerClass = OptimizerClass
config.pop("self", None) # self comes from "locals()"
config.pop("__class__", None) # self comes from "locals()"
self._as_config = as_config
self._config = config # keep all, to avoid weird behavior at mismatch between optim and configoptim
diff = ngtools.different_from_defaults(instance=self, instance_dict=config, check_mismatches=True)
params = ", ".join(f"{x}={y!r}" for x, y in sorted(diff.items()))
self.name = f"{self.__class__.__name__}({params})"
if not as_config:
# try instantiating for init checks
# if as_config: check can be done before setting attributes
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore", category=errors.InefficientSettingsWarning
) # this check does not need to be efficient
self(parametrization=4, budget=100)
def config(self) -> tp.Dict[str, tp.Any]:
return dict(self._config)
def __call__(
self, parametrization: IntOrParameter, budget: tp.Optional[int] = None, num_workers: int = 1
) -> Optimizer:
"""Creates an optimizer from the parametrization
Parameters
----------
instrumentation: int or Instrumentation
either the dimension of the optimization space, or its instrumentation
budget: int/None
number of allowed evaluations
num_workers: int
number of evaluations which will be run in parallel at once
"""
config = dict(config=self) if self._as_config else self.config()
if isinstance(self._OptimizerClass, ConfiguredOptimizer):
config = {} # ignore, it's already configured
run = self._OptimizerClass(parametrization=parametrization, budget=budget, num_workers=num_workers, **config) # type: ignore
run.name = self.name
# hacky but convenient to have around:
run._configured_optimizer = self # type: ignore
return run
def __repr__(self) -> str:
return self.name
def set_name(self, name: str, register: bool = False) -> "ConfiguredOptimizer":
"""Set a new representation for the instance"""
self.name = name
if register:
registry.register_name(name, self)
return self
def load(self, filepath: tp.Union[str, Path]) -> "Optimizer":
"""Loads a pickle and checks that it is an Optimizer."""
return self._OptimizerClass.load(filepath)
def __eq__(self, other: tp.Any) -> tp.Any:
if self.__class__ == other.__class__:
if self._config == other._config:
return True
return False
def _constraint_solver(parameter: p.Parameter, budget: int) -> p.Parameter:
"""Runs a suboptimization to solve the parameter constraints"""
parameter_without_constraint = parameter.copy()
parameter_without_constraint._constraint_checkers.clear()
parameter_without_constraint.tabu_length = 0
opt = registry["OnePlusOne"](parameter_without_constraint, num_workers=1, budget=budget)
for _ in range(budget):
cand = opt.ask()
# Our objective function is minimum for the point the closest to
# the original candidate under the constraints.
penalty = sum(utils._float_penalty(func(cand.value)) for func in parameter._constraint_checkers)
# TODO: this may not scale well with dimension
distance = np.tanh(np.sum(cand.get_standardized_data(reference=parameter) ** 2))
# TODO: because of the return whenever constraints are satisfied, the first case never arises
loss = distance if penalty <= 0 else penalty + distance + 1.0
opt.tell(cand, loss)
if penalty <= 0: # constraints are satisfied
break
data = opt.recommend().get_standardized_data(reference=parameter_without_constraint)
return parameter.spawn_child().set_standardized_data(data)