EVO X

Déployer HPO avec des algorithmes personnalisés

Déployer HPO avec des algorithmes personnalisés

Dans ce chapitre, nous nous concentrerons sur le déploiement HPO avec des algorithmes personnalisés, en mettant l’accent sur les détails plutôt que sur le workflow global. Une brève introduction au déploiement HPO est fournie dans le tutoriel, et une lecture préalable est fortement recommandée.

Rendre les algorithmes parallélisables

Puisque nous devons transformer l’algorithme interne en problème, il est crucial que l’algorithme interne soit parallélisable. Par conséquent, certaines modifications de l’algorithme peuvent être nécessaires.

  1. L’algorithme ne doit pas avoir de méthodes avec des opérations en place sur les attributs de l’algorithme lui-même.
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. La logique du code ne repose pas sur le flux de contrôle Python.
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)

Utilisation du HPOMonitor

Dans la tâche HPO, nous devons utiliser le HPOMonitor pour suivre les métriques de chaque algorithme interne. Le HPOMonitor n’ajoute qu’une seule méthode, tell_fitness, par rapport au monitor standard. Cet ajout est conçu pour offrir une plus grande flexibilité dans l’évaluation des métriques, car les tâches HPO impliquent souvent des métriques multidimensionnelles et complexes.

Les utilisateurs n’ont qu’à créer une sous-classe de HPOMonitor et surcharger la méthode tell_fitness pour définir des métriques d’évaluation personnalisées.

Nous fournissons également un simple HPOFitnessMonitor, qui supporte le calcul des métriques ‘IGD’ et ‘HV’ pour les problèmes multi-objectif, et la valeur minimale pour les problèmes mono-objectif.

Un exemple simple

Ici, nous démontrerons un exemple simple d’utilisation de HPO avec EvoX. Nous utiliserons l’algorithme PSO pour rechercher les hyperparamètres optimaux d’un algorithme de base pour résoudre le problème sphere.

D’abord, importons les modules nécessaires.

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

Ensuite, nous définissons un problème sphere simple. Notez que cela ne diffère pas des problems courants.

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

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

Ensuite, nous définissons l’algorithme, nous utilisons la fonction torch.cond et nous nous assurons qu’il est parallélisable. Plus précisément, nous modifions les opérations en place et ajustons le flux de contrôle Python.

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)

Pour gérer le flux de contrôle Python, nous utilisons torch.cond, ensuite, nous pouvons utiliser le StdWorkflow pour encapsuler le problem, l’algorithm et le monitor. Puis nous utilisons le HPOProblemWrapper pour transformer le StdWorkflow en problème HPO.

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)

Nous pouvons tester si le HPOProblemWrapper reconnaît correctement les hyperparamètres que nous avons définis. Puisque nous n’avons apporté aucune modification aux hyperparamètres pour les 7 instances, ils devraient être identiques pour toutes les instances.

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

Nous pouvons également spécifier notre propre ensemble de valeurs d’hyperparamètres. Notez que le nombre d’ensembles d’hyperparamètres doit correspondre au nombre d’instances dans le HPOProblemWrapper. Les hyperparamètres personnalisés doivent être fournis sous forme de dictionnaire dont les valeurs sont encapsulées dans le Parameter.

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)

Maintenant, nous utilisons l’algorithme PSO pour optimiser les hyperparamètres de ExampleAlgorithm. Notez que la taille de la population du PSO doit correspondre au nombre d’instances ; sinon, des erreurs inattendues peuvent survenir. Dans ce cas, nous devons transformer la solution dans le workflow externe, car le HPOProblemWrapper nécessite un dictionnaire en entrée.

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)
← Retour à la documentation