EVO X

Desplegar HPO con Algoritmos Personalizados

Desplegar HPO con Algoritmos Personalizados

En este capitulo, nos enfocaremos en desplegar HPO con algoritmos personalizados, enfatizando los detalles en lugar del flujo de trabajo general. Una breve introduccion al despliegue de HPO se proporciona en el tutorial, y se recomienda encarecidamente leerlo previamente.

Hacer los Algoritmos Paralelizables

Dado que necesitamos transformar el algoritmo interno en el problema, es crucial que el algoritmo interno sea paralelizable. Por lo tanto, algunas modificaciones al algoritmo pueden ser necesarias.

  1. El algoritmo no debe tener metodos con operaciones in-place sobre los atributos del propio algoritmo.
class ExampleAlgorithm(Algorithm):
    def __init__(self,...):
        self.pop = torch.rand(10,10) #atributo del propio algoritmo

    def step_in_place(self): # metodo con operaciones in-place
        self.pop.copy_(pop)

    def step_out_of_place(self): # metodo sin operaciones in-place
        self.pop = pop
  1. La logica del codigo no depende del flujo de control de Python.
class ExampleAlgorithm(Algorithm):
    def __init__(self,...):
        self.pop = rand(10,10) #atributo del propio algoritmo
        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): # funcion con flujo de control de Python
        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): # funcion sin flujo de control de Python
        x = rand()
        cond = x > 0.5
        self.pop = torch.cond(cond, self.plus, self.minus, y)

Uso del HPOMonitor

En la tarea de HPO, debemos usar el HPOMonitor para rastrear las metricas de cada algoritmo interno. El HPOMonitor agrega solo un metodo, tell_fitness, comparado con el monitor estandar. Esta adicion esta disenada para ofrecer mayor flexibilidad en la evaluacion de metricas, ya que las tareas de HPO frecuentemente involucran metricas multidimensionales y complejas.

Los usuarios solo necesitan crear una subclase de HPOMonitor y sobreescribir el metodo tell_fitness para definir metricas de evaluacion personalizadas.

Tambien proporcionamos un simple HPOFitnessMonitor, que soporta el calculo de las metricas ‘IGD’ y ‘HV’ para problemas multiobjetivo, y el valor minimo para problemas de objetivo unico.

Un ejemplo simple

Aqui, demostraremos un ejemplo simple de como usar HPO con EvoX. Usaremos el algoritmo PSO para buscar los hiperparametros optimos de un algoritmo basico para resolver el problema sphere.

Primero, importemos los modulos necesarios.

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

A continuacion, definimos un problema sphere simple. Ten en cuenta que esto no tiene diferencia con los problems comunes.

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

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

A continuacion, definimos el algoritmo, usamos la funcion torch.cond y nos aseguramos de que sea paralelizable. Especificamente, modificamos las operaciones in-place y ajustamos el flujo de control de 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)

Para manejar el flujo de control de Python, usamos torch.cond, a continuacion, podemos usar el StdWorkflow para envolver el problem, algorithm y monitor. Luego usamos el HPOProblemWrapper para transformar el StdWorkflow en un problema 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)

Podemos probar si el HPOProblemWrapper reconoce correctamente los hiperparametros que definimos. Dado que no hemos hecho modificaciones a los hiperparametros para las 7 instancias, deberian ser identicos en todas las instancias.

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

Tambien podemos especificar nuestro propio conjunto de valores de hiperparametros. Ten en cuenta que el numero de conjuntos de hiperparametros debe coincidir con el numero de instancias en el HPOProblemWrapper. Los hiperparametros personalizados deben proporcionarse como un diccionario cuyos valores estan envueltos en 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)

Ahora, usamos el algoritmo PSO para optimizar los hiperparametros de ExampleAlgorithm. Ten en cuenta que el tamano de la poblacion del PSO debe coincidir con el numero de instancias; de lo contrario, pueden ocurrir errores inesperados. En este caso, necesitamos transformar la solucion en el flujo de trabajo externo, ya que el HPOProblemWrapper requiere un diccionario como entrada.

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)
← Volver a la documentación