Evolutionary Algorithm

This notebook introduces PyGlove’s abstraction of an evolutionary algorithm. It explains how a new evolutionary algorithm can be created through pg.evolution.Evolution class with simplicity and flexibility.

!pip install pyglove

import pyglove as pg

Introduction

An evolutionary algorithm is an algorithm that uses mechanisms inspired by nature and solves problems through processes that emulate the behaviors of living organisms. The solutions play the role of individual organisms in a population. A mix of potential solutions is initially populated (usually randomly) first. Then each individual in the population is evaluated to provide one or more measures of its quality or fitness. Next, individuals are selected for reproduction, according to their fitnesses (e.g. pick the ones with highest fitness or make each of them reproduce in proportion to their fitness). It is possible for selection to go beyond fitness and include population-wide assessments such as how different an individual is from the rest. The selected individuals are modified via recombination and/or mutation, resulting in a set of new individuals called children. Children are similar to their parents but typically differ from them in small random ways. This reproduction process eventually results in a change in the composition of the population called the population update. The population update forms a new population from the previous one by adding newly created children and either keeping or discarding the parents, sometimes stochastically. This update might happen synchronously (a new generation of children is created from the previous generation of parents resulting in one large population update) or asynchronously (a single child is created in each cycle and the population is updated before the next child is selected).

The stages involved in the evolution process can be described as follows:


Programming Interface

With the program genome (pg.DNA) representing an individual in the population, the evolution process can be abstracted into three policies based on a list of DNAs:

• Population initialization: an empty list => initial population (List[pg.DNA])

• Population update: old population (List[pg.DNA]) => new population (List[pg.DNA])

• Reproduction: current population (List[pg.DNA]) => new children (List[pg.DNA])

Adding a flag to indicate whether the algorithm is for single-objective or multi-objective optimization. The evolution programming interface is defined as:

pg.evolution.Evolution(
    reproduction=<operation for mapping a DNA list to another DNA list>,
    population_init=<operation that generates a DNA list>,
    population_update=<operation for mapping a DNA list to another DNA list>,
    multi_objective=<True|False>)

For population_init, we can pass a pg.DNAGenerator to it or a tuple of (pg.DNAGenerator, <size of initial population>). For reproduction and population_update, they share the same interface which transforms a list of DNA to another list of DNA. Such transformation could be a single operation or a pipeline, which is discussed in the next section.

Operation: The Basics

In previous section, we have concluded that reproduction and population_update depends the operation that transforms a DNA list into another DNA list. In this section, we will dive into list processing and see how operations are expressed individually or as a pipeline using common Python operators.

The calling contract

Any callable object that follows the calling contract can be used as evolutionary operations passed to reproduction and population_update.

The simplest form of calling contract is (List[pg.DNA]) -> List[pg.DNA]. For example:

from typing import List

def first_ten(inputs: List[pg.DNA]) -> List[pg.DNA]:
  return inputs[:10]

Supporting stateful operations

In this example, the first_ten is a stateless operation which returns up to the first 10 elements from the input. For stateful operations, we can use objects, which could keep states as their members, for example:

class NextTen:

  def __init__(self):
    self._index = 0
  
  def __call__(self, inputs: List[pg.DNA]) -> List[pg.DNA]:
    start = min(self._index, len(inputs))
    end = min(self._index + 10, len(inputs))
    outputs = inputs[start:end]
    self._index += 10
    return outputs

What if a state is shared among multiple operations? One example is that in NEAT an operation in population_update will produce states like species, and the operations in reproduction may consume the species.

This can be done via receiving a global_state argument, which is a key value store that an operation can read it and write it. For example:

def top_species(inputs: List[pg.DNA],
                global_state: pg.geno.AttributeDict) -> List[pg.DNA]:
  return [dna in global_state.species[0] for dna in inputs]

Supporting step-based behaviors

In some cases, the behavior of an operation will change according to the search progress. For example, we may want to exploit more at the end of the search. This can be done by introducing step-based behaviors with adding the step argument:

def top(inputs: List[pg.DNA], step: int) -> List[pg.DNA]:
  n = 2 if step > 1000 else 5
  return inputs[:n]

Combining operations

The symmetry between the inputs and outputs allows operation chaining. However, the inputs is not available when we pass the operations to create a pg.evolution.Evolution object, which means inputs needs to be late bound.

To make operations easy to combine, we introduce class pg.evolution.Operation to facilitate creation of new operations.

More concretely, pg.evolution.Operation:

• defines an abstract call method that follows the calling contract of operations;

• overrides Python operators that glue operations together with rich semantics;

• is symbolic and can be manipulated at runtime;

The call method is defined as:

@abc.abstractmethod
def call(self, inputs: List[Any], global_state: pg.geno.AttributeDict, step: int = 0) -> List[Any]:

You may have noticed that its signature conforms to the calling contract except for the type of inputs and the return value, which must be a pg.DNA list for reproduction and population_update. But why do we make it List[Any]?

Processing heterogenous entities

Though the inputs and outputs of the operation as a whole should be a DNA list, intermediate inputs and outputs do not necessarily be. This allows intermediate operations to process heterogenous entities.

For example, in NSGA2,

• frontiers are derived via non-dorminated sorting on individuals in the population;

• during reproduction, top frontiers are selected with their top performing individuals, resulting a list of DNA list;

• finally, the list of DNA list will be flattened into a list for mutation, resulting a DNA list as new born individuals.


Chaining operations together

Operations can be composed together using a chain rule: every operator or called upon the operation should return the resulting operation to allow the chain to continue. For example:

op = (pg.evolution.selectors.Random(5)
      >> pg.evolution.selectors.Top(1)
      >> pg.evolution.mutators.Uniform())

Examples

Regularized Evolution

Regularized evolution keeps its last N (population_size) individuals as its population. For new suggestions, it selects the top performer among M (tournament_size) randomly chosen individuals from the population, and mutates it into a child:

pg.evolution.Evolution(
    reproduction=(
        pg.evolution.selectors.Random(tournament_size)
        >> pg.evolution.selectors.Top(1))
        >> mutator),
    population_init=(pg.geno.Random(), population_size),
    population_update=pg.evolution.selectors.Last(population_size))

See also: Reguarized evolution source code.

NEAT

NEAT proposes an evolutionary algorithm based on the concept of species, which is a cluster of individuals based on the distances of their genome representations. When a new individual is evaluated, it is added to the population and triggers speciation - a process to assign new individuals to a species, and remove retired individuals from its species. Reproduction takes two levels of selection, first it selects a list of species with probabilities proportional to their average fitnesses, then 1 member among the top performers in each species will be randomly selected. The selected individuals will be then recombined and mutated, leading to the next generation of population:

pg.evolution.Evolution(
    reproduction=(
        pg.evolution.GlobalStateGetter(‘living_species’)
        >> pg.evolution.selectors.Proportional(
            population_size, scaled_average_fitness()).for_each(
                ((lambda x: x.members)
                >> pg.evolution.selectors.Top(remaining_ratio)
                >> pg.evolution.selectors.Random(1))
            ).flatten()
        >> recombinator >> mutator),
    population_init=(pg.geno.Random(), population_size),
    population_update=(
        pg.evolution.selectors.Top(
        1, cluster=True, key=pg.evolution.get_generation_id)
        >> pg.evolution.neat.speciate(
        distance=compatibility_distance(
            disjoint_coefficient=disjoint_coefficient,
            matching_coefficient=matching_coefficient),
                distance_threshold=compatibility_threshold)))

See also: NEAT source code.

NSGA2

NSGA2 is a multi-objective evolutionary algorithm based on the concept of frontiers. A frontier is a group of individuals whose fitnesses (multiple metrics) are non-dominated to each other. Frontiers are sorted, frontiers in the front will have dominance over frontiers in the end. Therefore, we always choose individuals from earlier frontiers for reproduction than the latter ones. Within each frontier, individuals are sorted by pairwise crowding distance, which measures how different an individual performs compared to the other. The sorted individuals within sorted frontiers eventually flattens to form a list of elites, which will be chosen one by one for reproduction. Elites are computed in batches (=population_size), therefore we only trigger the frontier discovery and in-frontier sorting when the last generation are all evaluated:

pg.evolution.Evolution(
    reproduction=pg.evolution.nsga2.next_elite() >> mutator,
    population_init=(pg.geno.Random(), population_size * 2),
    population_update=(
        (pg.evolution.GlobalStateGetter('elites', [])
         + pg.evolution.Identity())
        >> pg.evolution.Lambda(
            pg.evolution.nsga2.nondominated_sort()).for_each(
                pg.evolution.nsga2.crowding_distance_sort()
            ).flatten()
        >> pg.evolution.selectors.First(
        population_size).as_global_state('elites')
        .set_global_state('elite_cursor', 0))
        ).if_true(lambda x: len(x) >= population_size),
    multi_objective=True)

See also: NSGA2 source code.