EVO X

HPO mit benutzerdefinierten Algorithmen bereitstellen

HPO mit benutzerdefinierten Algorithmen bereitstellen

In diesem Kapitel konzentrieren wir uns auf die Bereitstellung von HPO mit benutzerdefinierten Algorithmen, wobei wir die Details statt den Gesamtworkflow betonen. Eine kurze Einführung in die HPO-Bereitstellung finden Sie im Tutorial, und vorheriges Lesen wird dringend empfohlen.

Algorithmen parallelisierbar machen

Da wir den inneren Algorithmus in das Problem transformieren müssen, ist es entscheidend, dass der innere Algorithmus parallelisierbar ist. Daher können einige Modifikationen am Algorithmus notwendig sein.

  1. Der Algorithmus sollte keine Methoden mit In-Place-Operationen auf den Attributen des Algorithmus selbst haben.
class ExampleAlgorithm(Algorithm):
    def __init__(self,...):
        self.pop = torch.rand(10,10) #attribute of the algorithm itself

    def step_in_place(self): # method with in-place operations
        self.pop.copy_(pop)

    def step_out_of_place(self): # method without in-place operations
        self.pop = pop
  1. Die Codelogik darf nicht auf Python-Kontrollfluss basieren.
class ExampleAlgorithm(Algorithm):
    def __init__(self,...):
        self.pop = rand(10,10) #attribute of the algorithm itself
        pass

    def plus(self, y):
        return self.pop + y

    def minus(self, y):
        return self.pop - y

    def step_with_python_control_flow(self, y): # function with python control flow
        x = rand()
        if x > 0.5:
            self.pop = self.plus(y)
        else:
            self.pop = self.minus(y)

    def step_without_python_control_flow(self, y): # function without python control flow
        x = rand()
        cond = x > 0.5
        self.pop = torch.cond(cond, self.plus, self.minus, y)

Verwendung des HPOMonitor

Bei der HPO-Aufgabe sollten wir den HPOMonitor verwenden, um die Metriken jedes inneren Algorithmus zu verfolgen. Der HPOMonitor fügt im Vergleich zum Standard-monitor nur eine Methode hinzu, tell_fitness. Diese Ergänzung ist darauf ausgelegt, mehr Flexibilität bei der Bewertung von Metriken zu bieten, da HPO-Aufgaben oft mehrdimensionale und komplexe Metriken beinhalten.

Benutzer müssen nur eine Unterklasse von HPOMonitor erstellen und die tell_fitness-Methode überschreiben, um benutzerdefinierte Bewertungsmetriken zu definieren.

Wir bieten auch einen einfachen HPOFitnessMonitor, der die Berechnung der ‘IGD’- und ‘HV’-Metriken für Mehrzielprobleme sowie den Minimalwert für Einzielprobleme unterstützt.

Ein einfaches Beispiel

Hier demonstrieren wir ein einfaches Beispiel, wie HPO mit EvoX verwendet wird. Wir verwenden den PSO-Algorithmus, um die optimalen Hyperparameter eines einfachen Algorithmus zur Lösung des Sphere-Problems zu suchen.

Zunächst importieren wir die notwendigen Module.

import torch

from evox.algorithms.pso_variants.pso import PSO
from evox.core import Algorithm, Mutable, Parameter, Problem
from evox.problems.hpo_wrapper import HPOFitnessMonitor, HPOProblemWrapper
from evox.workflows import EvalMonitor, StdWorkflow

Als Nächstes definieren wir ein einfaches Sphere-Problem. Beachten Sie, dass dies keinen Unterschied zu den üblichen problems hat.

class Sphere(Problem):
    def __init__(self):
        super().__init__()

    def evaluate(self, x: torch.Tensor):
        return (x * x).sum(-1)

Als Nächstes definieren wir den Algorithmus, wir verwenden die torch.cond-Funktion und stellen sicher, dass er parallelisierbar ist. Konkret modifizieren wir In-Place-Operationen und passen den Python-Kontrollfluss an.

class ExampleAlgorithm(Algorithm):
    def __init__(self, pop_size: int, lb: torch.Tensor, ub: torch.Tensor):
        super().__init__()
        assert lb.ndim == 1 and ub.ndim == 1, f"Lower and upper bounds shall have ndim of 1, got {lb.ndim} and {ub.ndim}"
        assert lb.shape == ub.shape, f"Lower and upper bounds shall have same shape, got {lb.ndim} and {ub.ndim}"
        self.pop_size = pop_size
        self.hp = Parameter([1.0, 2.0, 3.0, 4.0])  # the hyperparameters to be optimized
        self.lb = lb
        self.ub = ub
        self.dim = lb.shape[0]
        self.pop = Mutable(torch.empty(self.pop_size, lb.shape[0], dtype=lb.dtype, device=lb.device))
        self.fit = Mutable(torch.empty(self.pop_size, dtype=lb.dtype, device=lb.device))

    def strategy_1(self, pop):  # one update strategy
        pop = pop * (self.hp[0] + self.hp[1])
        self.pop = pop

    def strategy_2(self, pop):  #  the other update strategy
        pop = pop * (self.hp[2] + self.hp[3])
        self.pop = pop

    def step(self):
        pop = torch.rand(self.pop_size, self.dim, dtype=self.lb.dtype, device=self.lb.device)  # simply random sampling
        pop = pop * (self.ub - self.lb)[None, :] + self.lb[None, :]
        control_number = torch.rand()
        self.pop = torch.cond(control_number < 0.5, self.strategy_1, self.strategy_2, (pop,))
        self.fit = self.evaluate(self.pop)

Um den Python-Kontrollfluss zu handhaben, verwenden wir torch.cond. Als Nächstes können wir den StdWorkflow verwenden, um das problem, den algorithm und den monitor zu verpacken. Dann verwenden wir den HPOProblemWrapper, um den StdWorkflow in ein HPO-Problem zu transformieren.

torch.set_default_device("cuda" if torch.cuda.is_available() else "cpu")
inner_algo = ExampleAlgorithm(10, -10 * torch.ones(8), 10 * torch.ones(8))
inner_prob = Sphere()
inner_monitor = HPOFitnessMonitor()
inner_monitor.setup()
inner_workflow = StdWorkflow()
inner_workflow.setup(inner_algo, inner_prob, monitor=inner_monitor)
# Transform the inner workflow to an HPO problem
hpo_prob = HPOProblemWrapper(iterations=9, num_instances=7, workflow=inner_workflow, copy_init_state=True)

Wir können testen, ob der HPOProblemWrapper die von uns definierten Hyperparameter korrekt erkennt. Da wir keine Modifikationen an den Hyperparametern für die 7 Instanzen vorgenommen haben, sollten sie über alle Instanzen identisch sein.

params = hpo_prob.get_init_params()
print("init params:\n", params)

Wir können auch unsere eigenen Hyperparameterwerte angeben. Beachten Sie, dass die Anzahl der Hyperparameter-Sets mit der Anzahl der Instanzen im HPOProblemWrapper übereinstimmen muss. Die benutzerdefinierten Hyperparameter sollten als Dictionary bereitgestellt werden, dessen Werte in Parameter eingewickelt sind.

params = hpo_prob.get_init_params()
# since we have 7 instances, we need to pass 7 sets of hyperparameters
params["self.algorithm.hp"] = torch.nn.Parameter(torch.rand(7, 4), requires_grad=False)
result = hpo_prob.evaluate(params)
print("params:\n", params, "\n")
print("result:\n", result)

Nun verwenden wir den PSO-Algorithmus, um die Hyperparameter von ExampleAlgorithm zu optimieren. Beachten Sie, dass die Populationsgröße des PSO mit der Anzahl der Instanzen übereinstimmen muss; andernfalls können unerwartete Fehler auftreten. In diesem Fall müssen wir die Lösung im äußeren Workflow transformieren, da der HPOProblemWrapper ein Dictionary als Eingabe erfordert.

class solution_transform(torch.nn.Module):
    def forward(self, x: torch.Tensor):
        return {"self.algorithm.hp": x}


outer_algo = PSO(7, -3 * torch.ones(4), 3 * torch.ones(4))
monitor = EvalMonitor(full_sol_history=False)
outer_workflow = StdWorkflow()
outer_workflow.setup(outer_algo, hpo_prob, monitor=monitor, solution_transform=solution_transform())
outer_workflow.init_step()
for _ in range(20):
    outer_workflow.step()
monitor = outer_workflow.get_submodule("monitor")
print("params:\n", monitor.topk_solutions, "\n")
print("result:\n", monitor.topk_fitness)
← Zurück zur Dokumentation