`pygx.algo.evolution`¶

Compositional evolutionary algorithms.

evolution ¶

Compositional evolutionary algorithms.

pygx.algo.evolution builds search algorithms from small, composable operations: selectors choose individuals, mutators perturb them, recombinators combine them, and pipelines connect them with the >> operator.

The top-level Evolution class wires reproduction, population initialization, and population updates into a single DNAGenerator that plugs into pg.sample, pg.iter, or any other consumer.

Common algorithm constructors are exposed at this level:

hill_climb
neat
nsga2
regularized_evolution

Each takes a Mutator (defaulting to mutators.Uniform()) and returns a ready-to-use Evolution object.

Choice ¶

Choice(
    ops: list[tuple[Operation, float | Callable[[int], float]]],
    limit: int | Callable[[int], int] | None = None,
    **kwargs
)

Bases: Operation

Applying a list of operations with probabilities in a pipeline.

Example:

pg.algo.evolution.Choice([
  (pg.algo.evolution.mutators.Uniform() ** 3, 0.5),
  (pg.algo.evolution.mutators.Swap(), 0.4)
], limit=1)

The code above applies 3 pipelined uniform mutations with probability 0.5, and a swap of two sibling DNA branches with probability 0.4. Due to limit=1, at most 1 of the two operations is applied.

Source code in pygx/algo/evolution/_base.py

def __init__(
    self,
    ops: list[tuple['Operation', float | Callable[[int], float]]],
    limit: int | Callable[[int], int] | None = None,
    **kwargs,
):
    super().__init__(**dict(ops=ops, limit=limit), **kwargs)

Concatenation ¶

Concatenation(ops: list[Operation], **kwargs)

Bases: Operation

Concatenating the operations' outputs based on the same input.

A concatenate operation passes the input to all its child operations, and concatenate their outputs. Different from Union, the duplicated items will be kept.

Example:

pg.algo.evolution.Concatenation([
  pg.algo.evolution.selectors.Top(10),
  pg.algo.evolution.selectors.First(10)
])

which is equivalent to:

pg.algo.evolution.selectors.Top(10) + pg.algo.evolution.selectors.First(10)

The code above returns top 10 DNA concatenated with the first 10 DNAs. The result may contain duplicated items.

Source code in pygx/algo/evolution/_base.py

def __init__(self, ops: 'list[Operation]', **kwargs):
    super().__init__(**dict(ops=ops), **kwargs)

Conditional ¶

Conditional(
    predicate: Callable[..., bool],
    true_op: Operation | None = None,
    false_op: Operation | None = None,
    **kwargs
)

Bases: Operation

Conditional operation.

A conditional operation allows sub-operations to be applied only when a condition is met.

Example:

pg.algo.evolution.Conditional(
  lambda x: len(x) > 10,
  true_op=Top(10))

which is equivalent to:

Top(10).if_true(lambda x: len(x) > 10)

Source code in pygx/algo/evolution/_base.py

def __init__(
    self,
    predicate: Callable[..., bool],
    true_op: 'Operation | None' = None,
    false_op: 'Operation | None' = None,
    **kwargs,
):
    super().__init__(
        **dict(predicate=predicate, true_op=true_op, false_op=false_op),
        **kwargs,
    )

Difference ¶

Difference(ops: list[Operation], **kwargs)

Bases: Operation

Compute the difference of operation outputs based on the same input.

A difference operation passes the input to all its child operations, and subtracts the other operations' outputs from the first operation's output: it keeps the items of the first operation's output — in that output's order — that appear (by object identity) in no other operation's output.

Example:

pg.algo.evolution.Difference([
  pg.algo.evolution.selectors.Top(10),
  pg.algo.evolution.selectors.First(10)
])

which is equivalent to:

pg.algo.evolution.selectors.Top(10) - pg.algo.evolution.selectors.First(10)

The code above returns top 10 DNA only when they are not among the first 10 elements of the input.

Source code in pygx/algo/evolution/_base.py

def __init__(self, ops: 'list[Operation]', **kwargs):
    super().__init__(**dict(ops=ops), **kwargs)

DNAOperation ¶

Bases: Operation

Operation that takes a list of DNA as both input and output.

input_element_type `property` ¶

input_element_type: type[DNA]

Returns the input element type.

output_element_type `property` ¶

output_element_type: type[DNA]

Returns the output element type.

ElementWise ¶

ElementWise(op: Operation, **kwargs)

Bases: Operation

Apply the operation on each input item and aggregate them.

Source code in pygx/algo/evolution/_base.py

def __init__(self, op: 'Operation', **kwargs):
    super().__init__(**dict(op=op), **kwargs)

Evolution ¶

Evolution(reproduction: Operation, **kwargs)

Bases: DNAGenerator

An evolutionary algorithm based on compositional operations.

Common evolutionary algorithms can be abstracted with 3 operations:

Select parent(s) from the population.
Recombine the DNA from the parents into a list of child DNA.
Mutate each child DNA.

Plus 2 population related operations:

Initialize the population.
Update the population when the fitness is determined for the child DNA.

All the operations above except population initialization can be described by the Operation interface, which transforms a list of DNA into another list of DNA. Moreover, Operation allows complex algorithms to be represented as a composition of elementary operations. For example, Regularized Evolution can be described as:

algo = pg.algo.evolution.Evolution(
    (pg.algo.evolution.selectors.Random(10)
     >> pg.algo.evolution.selectors.Top(1)
     >> pg.algo.evolution.mutators.Uniform()),
    population_init=(pg.geno.Random(), 50),
    population_update=pg.algo.evolution.selectors.Last(50))

(Also available as the regularized_evolution helper.)

The algorithm can then drive a sampling loop over a search space:

search_space = pg.Dict(x=pg.oneof(range(100)), y=pg.oneof(range(100)))

best_value, best_reward = None, None
for value, feedback in pg.iter(search_space, 50, algo):
  reward = value.x + value.y
  if best_reward is None or reward > best_reward:
    best_value, best_reward = value, reward
  feedback(reward)

Source code in pygx/algo/evolution/_base.py

def __init__(self, reproduction: 'Operation', **kwargs):
    super().__init__(**dict(reproduction=reproduction), **kwargs)

population `property` ¶

population: list[DNA]

Returns current population.

global_state `property` ¶

global_state: AttributeDict

Returns the global state.

recover ¶

recover(
    history: Iterable[tuple[DNA, None | float | tuple[float, ...]]],
) -> None

Recover states by replaying the proposal history.

Source code in pygx/algo/evolution/_base.py

def recover(
    self,
    history: Iterable[tuple[pg.DNA, None | float | tuple[float, ...]]],
) -> None:
    """Recover states by replaying the proposal history."""
    # Recover the state of the population.
    init_population = []

    dna_spec = self.dna_spec
    assert dna_spec is not None, self
    for dna, reward in history:
        self._num_proposals += 1
        dna.use_spec(dna_spec)
        if reward is not None:
            # NOTE(daiyip): There is a possibility that the client has provided the
            # reward to the controller, but the controller process restarted before
            # calling the `feedback` method. Such cases can be identified by
            # checking the `feedback_sequence_number` metadata.
            if get_feedback_sequence_number(dna) is None:
                self.feedback(dna, reward)
            else:
                assert get_fitness(dna) == reward, (dna, reward)
                self._population.append(dna)
                if self._population_update:
                    self._population = self._population_update(
                        self._population,
                        global_state=self._global_state,
                        step=self._num_feedbacks,
                    )
                self._num_feedbacks += 1
            if is_initial_population(dna):
                init_population.append((dna, reward))

        # Recover `self.num_generations`.
        generation_id = get_generation_id(dna)
        if generation_id > self.num_generations:
            self._global_state.num_generations = generation_id

    # Recover the state of the population initializer.
    if (
        self._init_population_size is not None
        and len(init_population) >= self._init_population_size
    ):
        self._population_initialized = True
    self._init_population_generator.recover(init_population)

Flatten ¶

Flatten(max_level: int | None = None, **kwargs)

Bases: Operation

Flatten elements from the input.

Source code in pygx/algo/evolution/_base.py

def __init__(self, max_level: int | None = None, **kwargs):
    super().__init__(**dict(max_level=max_level), **kwargs)

GlobalStateGetter ¶

GlobalStateGetter(key: str, default: list | None = None, **kwargs)

Bases: Operation

Returns a key in the global_state as output.

Source code in pygx/algo/evolution/_base.py

def __init__(self, key: str, default: list | None = None, **kwargs):
    super().__init__(**dict(key=key, default=default), **kwargs)

GlobalStateSetter ¶

GlobalStateSetter(key: str, **kwargs)

Bases: Operation

Set the value for a key in global_state and return an empty list.

Source code in pygx/algo/evolution/_base.py

def __init__(self, key: str, **kwargs):
    super().__init__(**dict(key=key), **kwargs)

Identity ¶

Bases: Operation

Returns the input itself.

Intersection ¶

Intersection(ops: list[Operation], **kwargs)

Bases: Operation

Intersecting operations's outputs based on the same input.

An intersect operation passes the input to all its child operations, and intersects their outputs: it keeps the items of the first operation's output — in that output's order — that also appear (by object identity) in every other operation's output.

Example:

pg.algo.evolution.Intersection([
  pg.algo.evolution.selectors.Last(10),
  pg.algo.evolution.selectors.Top(1),
])

which is equivalent to:

pg.algo.evolution.selectors.Last(10) & pg.algo.evolution.selectors.Top(1)

The code above returns top 1 DNA only when it's among the last 10 items of the input.

Source code in pygx/algo/evolution/_base.py

def __init__(self, ops: 'list[Operation]', **kwargs):
    super().__init__(**dict(ops=ops), **kwargs)

Inversion ¶

Inversion(op: Operation, **kwargs)

Bases: Operation

Computing the inversion of operation's output against the input.

Source code in pygx/algo/evolution/_base.py

def __init__(self, op: 'Operation', **kwargs):
    super().__init__(**dict(op=op), **kwargs)

Lambda ¶

Lambda(fn: Callable[..., Any], **kwargs)

Bases: Operation

A lambda operation.

Source code in pygx/algo/evolution/_base.py

def __init__(self, fn: Callable[..., Any], **kwargs):
    super().__init__(**dict(fn=fn), **kwargs)

Mutator ¶

Bases: DNAOperation

Base class for mutators.

A mutator performs a mutation, i.e. a random transformation that converts a parent DNA to a child DNA. Mutations should reach the full search space through composition and should prefer local transformations individually.

mutate ¶

mutate(dna: DNA, global_state: AttributeDict, step: int = 0) -> DNA | list[DNA]

Mutates the DNA at a given step.

User should override this method or mutate_list method with optional keyword arguments 'global_state' and 'step'.

Parameters:

Name	Type	Description	Default
`dna`	`DNA`	DNA to mutate.	required
`global_state`	`AttributeDict`	An `AttributeDict` object as the container of global states.	required
`step`	`int`	Number of examples historically proposed, which can be used for determining a mutation schedule.	`0`

Returns:

Type	Description
`DNA \| list[DNA]`	A new DNA or a DNA list as the result of the mutation.

Source code in pygx/algo/evolution/_base.py

def mutate(
    self, dna: pg.DNA, global_state: pg.geno.AttributeDict, step: int = 0
) -> pg.DNA | list[pg.DNA]:
    """Mutates the DNA at a given step.

    User should override this method or `mutate_list` method with optional
    keyword arguments 'global_state' and 'step'.

    Args:
      dna: DNA to mutate.
      global_state: An `AttributeDict` object as the container of global states.
      step: Number of examples historically proposed, which can be used for
        determining a mutation schedule.

    Returns:
      A new DNA or a DNA list as the result of the mutation.
    """
    raise NotImplementedError()

mutate_list ¶

mutate_list(
    dna_list: list[DNA], global_state: AttributeDict, step: int = 0
) -> list[DNA]

Mutate the DNA in the input one by one and concatenate their outputs.

User should override this method instead of mutate if mutation depends on the list-wise information. Keyword arguments global_state and step are optional when override.

Parameters:

Name	Type	Description	Default
`dna_list`	`list[DNA]`	a list of DNA to mutate.	required
`global_state`	`AttributeDict`	An `AttributeDict` object as the container of global states.	required
`step`	`int`	Number of examples historically proposed, which can be used for determining a mutation schedule.	`0`

Returns:

Type	Description
`list[DNA]`	a list of DNA as the result of the mutation.

Source code in pygx/algo/evolution/_base.py

def mutate_list(
    self,
    dna_list: list[pg.DNA],
    global_state: pg.geno.AttributeDict,
    step: int = 0,
) -> list[pg.DNA]:
    """Mutate the DNA in the input one by one and concatenate their outputs.

    User should override this method instead of `mutate` if mutation depends on
    the list-wise information. Keyword arguments `global_state` and `step` are
    optional when override.

    Args:
      dna_list: a list of DNA to mutate.
      global_state: An `AttributeDict` object as the container of global states.
      step: Number of examples historically proposed, which can be used for
        determining a mutation schedule.

    Returns:
      a list of DNA as the result of the mutation.
    """
    results = []
    for dna in dna_list:
        output = self._mutate(dna, global_state=global_state, step=step)
        if isinstance(output, list):
            results.extend(output)
        else:
            results.append(output)
    return results

Operation ¶

Bases: Object

Base class for evolutionary operations.

An evolutionary operation transforms a DNA list into another DNA list. This abstraction provides the flexibility to describe common evolutionary operations including but not limited to:

Selection: Select M parents from a population of size N.
Recombination: Generate M child candidates from N parents.
Mutation: Mutate the N child candidates one by one, each can produce one or K DNA, resulting in N * K child DNA.

Operations are compositional via the following operators:

x >> y:    Pipeline x and y by passing the output of x to y as input.
x + y:     Concatenate the outputs of x and y, based on the same input.
           Resulted list could contain duplicates.
x[m:n:p]   Slice x's output from m (inclusive) to n (exclusive) with step p.
x | y:     Union the outputs of x and y based on the same input.
           Resulted list will not contain duplicated items.
x & y:     Intersect the outputs of x and y, based on the same input.
           Resulted list will not contain duplicated items.
x - y:     Remove elements from x's output which also appears in y's output,
           based on the same input.
x ^ y:     Keep items that are either in x's output or in y's output, but
           not both. Duplicated items in x's output or in y's output will
           be preserved.
x * K:     Repeat operation x for K times based on the same input,
           concatenate their outputs.
x ** K:    Pipeline operation x for K times, equivalent to `x >> .. >> x`
           (`x` appears K times in the formula).
-x or ~x:  Compute the inversion of x's output based on the input.
           Equivalent to `Identity() - x`

x.if_true(f):   Conditionally apply x if predicate f returns True.
x.if_false(f):  Conditionally apply x if predicate f returns False.

Besides, the following operations are commonly used during composition:

Identity()           :   Returns the input DNA list.

Lambda(              :   Creates an Operation using lambda.
  lambda x: x[1:])

Choice([             :   Perform x, y based on probabilities.
  (x, prob1),
  (y, prob2)])

Conditional(         :   Perform x if input length > 5, otherwise y.
  lambda x: len(x) > 5,
  x, y)

UntilChange(x)       :   Repeat x until its output changes from the input.

ElementWise(         :   Perform element-wise operation and concatenate.
  lambda x: sorted(x))

Flatten()            :   Flatten nested lists into a flat list.

GlobalStateGetter(   :   Fetch global state by key 'species'.
  'species')

GlobalStateSetter(   :   Set global state by key 'species' from input.
  'species')

The operand in compositional operations also supports plain callable objects that take the input list and return the output list, with optional keyword arguments global_state and step. For example:

pg.algo.evolution.selectors.Last(50) >> (lambda x: x[:2])

input_element_type `property` ¶

input_element_type: None | type[Any] | tuple[type[Any]]

Returns the input element type. Subclasses can override.

output_element_type `property` ¶

output_element_type: None | type[Any] | tuple[type[Any]]

Returns the output element type. Subclasses can override.

operation_method `property` ¶

operation_method: Callable[..., list[Any]]

Returns a member method as operation.

_on_input ¶

_on_input(inputs: list[Any]) -> None

Event that is triggered when an input is received.

Source code in pygx/algo/evolution/_base.py

def _on_input(self, inputs: list[Any]) -> None:
    """Event that is triggered when an input is received."""

call ¶

call(
    inputs: list[Any], global_state: AttributeDict, step: int = 0
) -> list[Any]

Subclasses should override this method.

The global_state and step are optional for the subclasses' call signature.

Parameters:

Name	Type	Description	Default
`inputs`	`list[Any]`	A list of values as inputs.	required
`global_state`	`AttributeDict`	An `AttributeDict` object as the global state container, which is readable/writable during the operation.	required
`step`	`int`	Number of examples historically proposed, which can be used for determining a cross over schedule.	`0`

Returns:

Type	Description
`list[Any]`	A list of values as output of current operation.

Source code in pygx/algo/evolution/_base.py

def call(
    self,
    inputs: list[Any],
    global_state: pg.geno.AttributeDict,
    step: int = 0,
) -> list[Any]:
    """Subclasses should override this method.

    The `global_state` and `step` are optional for the subclasses' call
    signature.

    Args:
      inputs: A list of values as inputs.
      global_state: An `AttributeDict` object as the global state container,
        which is readable/writable during the operation.
      step: Number of examples historically proposed, which can be used for
        determining a cross over schedule.

    Returns:
      A list of values as output of current operation.
    """
    raise NotImplementedError()

with_prob ¶

with_prob(probability: float | Callable[[int], float], seed: int | None = None)

With probability.

Source code in pygx/algo/evolution/_base.py

def with_prob(
    self,
    probability: float | Callable[[int], float],
    seed: int | None = None,
):
    """With probability."""
    return Choice(ops=[(self, probability)], seed=seed)

if_true ¶

if_true(predicate: Callable[..., bool]) -> Conditional

Conditionally applies current operation when predicate returns True.

Parameters:

Name	Type	Description	Default
`predicate`	`Callable[..., bool]`	The predicate that takes the outputs from the previous operation as input, with optional keyword arguments `global_state` and `step`. Returns True if current operation needs to be enabled. Otherwise no operation will be performed.	required

Returns:

Type	Description
`Conditional`	A conditional operation.

Source code in pygx/algo/evolution/_base.py

def if_true(self, predicate: Callable[..., bool]) -> 'Conditional':
    """Conditionally applies current operation when predicate returns True.

    Args:
      predicate: The predicate that takes the outputs from the previous
        operation as input, with optional keyword arguments `global_state` and
        `step`. Returns True if current operation needs to be enabled.
        Otherwise no operation will be performed.

    Returns:
       A conditional operation.
    """
    return Conditional(predicate=predicate, true_op=self, false_op=None)

if_false ¶

if_false(predicate: Callable[..., bool]) -> Conditional

Conditionally applies current operation when predicate returns False.

Parameters:

Name	Type	Description	Default
`predicate`	`Callable[..., bool]`	The predicate that takes the outputs from the previous operation as input, with optional keyword arguments `global_state` and `step`. Returns False if current operation needs to be enabled. Otherwise no operation will be performed.	required

Returns:

Type	Description
`Conditional`	A conditional operation.

Source code in pygx/algo/evolution/_base.py

def if_false(self, predicate: Callable[..., bool]) -> 'Conditional':
    """Conditionally applies current operation when predicate returns False.

    Args:
      predicate: The predicate that takes the outputs from the previous
        operation as input, with optional keyword arguments `global_state` and
        `step`. Returns False if current operation needs to be enabled.
        Otherwise no operation will be performed.

    Returns:
       A conditional operation.
    """
    return Conditional(predicate=predicate, true_op=None, false_op=self)

for_each ¶

for_each(op)

For each element in the output perform the operation.

Source code in pygx/algo/evolution/_base.py

def for_each(self, op):
    """For each element in the output perform the operation."""
    return self >> ElementWise(op=op)

until_change ¶

until_change(max_attempts: int | None = None)

Ensure current operation will change the inputs.

Source code in pygx/algo/evolution/_base.py

def until_change(self, max_attempts: int | None = None):
    """Ensure current operation will change the inputs."""
    return UntilChange(op=self, max_attempts=max_attempts)

flatten ¶

flatten(max_level: int | None = None)

Flatten output with a max level.

Source code in pygx/algo/evolution/_base.py

def flatten(self, max_level: int | None = None):
    """Flatten output with a max level."""
    return self >> Flatten(max_level=max_level)

global_state ¶

global_state(key: str)

Returns a global state getter based on the key.

Source code in pygx/algo/evolution/_base.py

def global_state(self, key: str):
    """Returns a global state getter based on the key."""
    return GlobalStateGetter(key=key)

as_global_state ¶

as_global_state(key: str)

Returns a global state setter based on the key.

Source code in pygx/algo/evolution/_base.py

def as_global_state(self, key: str):
    """Returns a global state setter based on the key."""
    return Pipeline(ops=[self, GlobalStateSetter(key=key)])

set_global_state ¶

set_global_state(key: str, value: Any)

Set global state and return current output.

Source code in pygx/algo/evolution/_base.py

def set_global_state(self, key: str, value: Any):
    """Set global state and return current output."""
    return self + GlobalStateSetter(key=key, value=value)

Pipeline ¶

Pipeline(ops: list[Operation] | Any, **kwargs)

Bases: Operation

Chaining operations into a pipeline.

A pipeline operation chains multiple operations by passing the output of previous operation to the next operations.

Example:

pg.algo.evolution.Pipeline([
  pg.algo.evolution.selectors.Random(10),
  pg.algo.evolution.selectors.Top(1),
  pg.algo.evolution.mutators.Uniform()
])

which is equivalent to:

(pg.algo.evolution.selectors.Random(10)
  >> pg.algo.evolution.selectors.Top(1)
  >> pg.algo.evolution.mutators.Uniform())

The code above returns a uniformly mutated DNA from the winner of 10 randomly selected DNA from the input.

Source code in pygx/algo/evolution/_base.py

def __init__(self, ops: 'list[Operation] | Any', **kwargs):
    super().__init__(**dict(ops=ops), **kwargs)

Power ¶

Power(op: Operation, k: int | Callable[[int], int], **kwargs)

Bases: Operation

Pipeline a repeated operation multiple times.

Power(x, k) (or x ** k) is equivalent to x >> x >> ... >> x (k items).

Example:

pg.algo.evolution.Power(pg.algo.evolution.mutators.Uniform(), 3)

which is equivalent to:

pg.algo.evolution.mutators.Uniform() ** 3

The code above creates a DNA with 3 mutated positions for each DNA in the input.

Source code in pygx/algo/evolution/_base.py

def __init__(
    self,
    op: 'Operation',
    k: int | Callable[[int], int],
    **kwargs,
):
    super().__init__(**dict(op=op, k=k), **kwargs)

Recombinator ¶

Bases: DNAOperation

Base class for recombinators.

recombine `abstractmethod` ¶

recombine(
    parents: list[DNA], global_state: AttributeDict, step: int
) -> list[DNA]

Generate a list of child DNA based on the list of parents given.

User should override this method with optional keyword arguments 'global_state' and 'step'.

The parents DNA contains a metadata field 'generation', which is the generation of the parent DNA. If the Recombinator does not assign this field for the new child DNA, the child DNA will have the maximum generation from the parents plus 1.

Parameters:

Name	Type	Description	Default
`parents`	`list[DNA]`	Parent trials.	required
`global_state`	`AttributeDict`	An `AttributeDict` object as the global state container, which is readable/writable during the operation.	required
`step`	`int`	Number of examples historically proposed, which can be used for determining a cross over schedule.	required

Returns:

Type	Description
`list[DNA]`	A list of generated child DNA.

Source code in pygx/algo/evolution/_base.py

@abc.abstractmethod
def recombine(
    self,
    parents: list[pg.DNA],
    global_state: pg.geno.AttributeDict,
    step: int,
) -> list[pg.DNA]:
    """Generate a list of child DNA based on the list of parents given.

    User should override this method with optional keyword arguments
    'global_state' and 'step'.

    The parents DNA contains a metadata field 'generation', which is the
    generation of the parent DNA. If the Recombinator does not assign this
    field for the new child DNA, the child DNA will have the maximum generation
    from the parents plus 1.

    Args:
      parents: Parent trials.
      global_state: An `AttributeDict` object as the global state container,
        which is readable/writable during the operation.
      step: Number of examples historically proposed, which can be used for
        determining a cross over schedule.

    Returns:
      A list of generated child DNA.
    """

_on_input ¶

_on_input(inputs: list[Any]) -> None

Override to check number of parents.

Source code in pygx/algo/evolution/_base.py

def _on_input(self, inputs: list[Any]) -> None:
    """Override to check number of parents."""
    if self.NUM_PARENTS is not None and len(inputs) != self.NUM_PARENTS:
        raise ValueError(
            f'{self.__class__.__name__} supports recombination on exact '
            f'{self.NUM_PARENTS} parents. Encountered: {inputs!r}.'
        )

Repeat ¶

Repeat(op: Operation, k: int | Callable[[int], int], **kwargs)

Bases: Operation

Repeat an operation multiple times based on the same input.

A repeat operation passes the input to the child operation for K times, and concatenate their outputs.

Example:

Repeat(pg.algo.evolution.mutators.Uniform(), 3)

which is equivalent to:

pg.algo.evolution.mutators.Uniform() * 3

The code above creates 3 mutated DNA for each DNA from the input.

It's important to differentiate * and **. For instance, Uniform() ** 3 creates one DNA (mutated on 3 positions) for each DNA from the input.

Source code in pygx/algo/evolution/_base.py

def __init__(
    self,
    op: 'Operation',
    k: int | Callable[[int], int],
    **kwargs,
):
    super().__init__(**dict(op=op, k=k), **kwargs)

Selector ¶

Bases: Operation

Base class for selectors.

select `abstractmethod` ¶

select(inputs: list[Any], global_state: AttributeDict, step: int) -> list[Any]

Select a list of outputs from the inputs.

A selector has two use cases:

Used as parents selector, which selects individuals from the population as parents for recombination. It will be called before the recombination step within Evolution's propose method (inherited from pg.geno.DNAGenerator).
Used as a population updater, which selects individuals from previous population as a new population. It will be called everytime the population is updated, triggered by Evolution's feedback method.

Parameters:

Name	Type	Description	Default
`inputs`	`list[Any]`	a list of objects as input.	required
`global_state`	`AttributeDict`	An `AttributeDict` object as the global state container, which is readable/writable during the operation.	required
`step`	`int`	Number of examples historically proposed, which can be used for determining a cross over schedule.	required

Source code in pygx/algo/evolution/_base.py

@abc.abstractmethod
def select(
    self, inputs: list[Any], global_state: pg.geno.AttributeDict, step: int
) -> list[Any]:
    """Select a list of outputs from the inputs.

    A selector has two use cases:

    * Used as parents selector, which selects individuals from the population
      as parents for recombination. It will be called before the recombination
      step within `Evolution`'s
      [`propose`][pygx.geno.DNAGenerator.propose] method (inherited from
      `pg.geno.DNAGenerator`).

    * Used as a population updater, which selects individuals from previous
      population as a new population. It will be called everytime the
      population is updated, triggered by `Evolution`'s
      [`feedback`][pygx.geno.DNAGenerator.feedback] method.

    Args:
      inputs: a list of objects as input.
      global_state: An `AttributeDict` object as the global state container,
        which is readable/writable during the operation.
      step: Number of examples historically proposed, which can be used for
        determining a cross over schedule.
    """

Slice ¶

Slice(
    op: Operation, index: int | slice | Callable[[int], int | slice], **kwargs
)

Bases: Operation

Slice the operation's output.

A slice operation passes the input to the child operation and slices its output based on the index.

Example:

Slice(pg.algo.evolution.Identity(), slice(0, 4, 2))

which is equivalent to:

pg.algo.evolution.Identity()[:4:2]

Source code in pygx/algo/evolution/_base.py

def __init__(
    self,
    op: 'Operation',
    index: int | slice | Callable[[int], int | slice],
    **kwargs,
):
    super().__init__(**dict(op=op, index=index), **kwargs)

SymmetricDifference ¶

SymmetricDifference(ops: list[Operation], **kwargs)

Bases: Operation

Compute the symmetric difference of operation outputs based on the input.

A symmetric difference operation passes the input to all its child operations, and keeps the items that appear (by object identity) in exactly one operation's output, in the order the operations emit them.

Example:

pg.algo.evolution.SymmetricDifference([
  pg.algo.evolution.selectors.Top(10),
  pg.algo.evolution.selectors.First(10)
])

which is equivalent to:

pg.algo.evolution.selectors.Top(10) ^ pg.algo.evolution.selectors.First(10)

The code above returns items that are either among the top 10 or among the first 10 but not both.

Source code in pygx/algo/evolution/_base.py

def __init__(self, ops: 'list[Operation]', **kwargs):
    super().__init__(**dict(ops=ops), **kwargs)

Union ¶

Union(ops: list[Operation], **kwargs)

Bases: Operation

Unioning operations' outputs based on the same input.

A union operation passes the input to all its child operations, and unions their outputs. Items appear in the order the child operations emit them (child operations run in their declaration order), with duplicates — determined by object identity — removed.

Example:

pg.algo.evolution.Union([
  pg.algo.evolution.selectors.Random(10),
  pg.algo.evolution.selectors.Top(1),
])

which is equivalent to:

pg.algo.evolution.selectors.Random(10) | pg.algo.evolution.selectors.Top(1)

The code above returns 10 random DNAs and the top 1 DNA. If the top DNA is among the random ones, only one occurrence is returned.

Source code in pygx/algo/evolution/_base.py

def __init__(self, ops: 'list[Operation]', **kwargs):
    super().__init__(**dict(ops=ops), **kwargs)

UntilChange ¶

UntilChange(
    op: Operation,
    max_attempts: int | Callable[[int], int] | None = None,
    **kwargs
)

Bases: Operation

Repeat an operation until its output is different from the input.

Source code in pygx/algo/evolution/_base.py

def __init__(
    self,
    op: 'Operation',
    max_attempts: int | Callable[[int], int] | None = None,
    **kwargs,
):
    super().__init__(**dict(op=op, max_attempts=max_attempts), **kwargs)

Scalar ¶

Bases: Object

Interface for step-based scalars.

A scalar maps an integer step to a value. Scalars compose with the standard arithmetic operators, so schedules can be written as expressions of STEP:

Example:

s = pg.algo.scalars.STEP * 0.1 + 1.0
assert s(0) == 1.0
assert s(10) == 2.0

call `abstractmethod` ¶

call(step: int) -> Any

Implementation. Subclass should override this method.

Source code in pygx/algo/scalars/_base.py

@abc.abstractmethod
def call(self, step: int) -> Any:
    """Implementation. Subclass should override this method."""

floor ¶

floor()

Returns the floor of current scalar.

Source code in pygx/algo/scalars/_base.py

def floor(self):
    """Returns the floor of current scalar."""
    return Floor(x=self)

ceil ¶

ceil()

Returns the ceiling of current scalar.

Source code in pygx/algo/scalars/_base.py

def ceil(self):
    """Returns the ceiling of current scalar."""
    return Ceiling(x=self)

get_feedback_sequence_number ¶

get_feedback_sequence_number(dna: DNA) -> int | None

Returns the 1-based feedback sequence number of the DNA.

Source code in pygx/algo/evolution/_base.py

def get_feedback_sequence_number(dna: pg.DNA) -> int | None:
    """Returns the 1-based feedback sequence number of the DNA."""
    return dna.metadata.get(DNA_METADATA_FEEDBACK_SEQUENCE_NUMBER)

get_fitness ¶

get_fitness(dna: DNA) -> float | tuple[float, ...]

Returns the fitness associated with a DNA.

Source code in pygx/algo/evolution/_base.py

def get_fitness(dna: pg.DNA) -> float | tuple[float, ...]:
    """Returns the fitness associated with a DNA."""
    return dna.metadata[DNA_METADATA_FITNESS]

get_generation_id ¶

get_generation_id(dna: DNA) -> int

Returns the 1-based generation ID.

Source code in pygx/algo/evolution/_base.py

def get_generation_id(dna: pg.DNA) -> int:
    """Returns the 1-based generation ID."""
    return dna.metadata[DNA_METADATA_GENERATION_ID]

is_initial_population ¶

is_initial_population(dna: DNA) -> bool

Returns True if the DNA belongs to the initial population.

Source code in pygx/algo/evolution/_base.py

def is_initial_population(dna: pg.DNA) -> bool:
    """Returns True if the DNA belongs to the initial population."""
    return dna.metadata[DNA_METADATA_INITIAL_POPULATION]

set_fitness ¶

set_fitness(dna: DNA, fitness: float | tuple[float, ...]) -> DNA

Set the evaluated fitness to a DNA and return itself.

Source code in pygx/algo/evolution/_base.py

def set_fitness(dna: pg.DNA, fitness: float | tuple[float, ...]) -> pg.DNA:
    """Set the evaluated fitness to a DNA and return itself."""
    return dna.set_metadata(DNA_METADATA_FITNESS, fitness)

hill_climb ¶

hill_climb(
    mutator: Mutator = Uniform(),
    batch_size: int = 1,
    init_population_size: int = 1,
    seed: int | None = None,
) -> Evolution

Hill-Climbing algorithm, with an extra batched setting.

Batched setting was shown to be effective in https://arxiv.org/pdf/1911.06317.pdf and https://arxiv.org/pdf/2003.01239.pdf, especially in noisy objective settings.

Parameters:

Name	Type	Description	Default
`mutator`	`Mutator`	Mutator to use.	`Uniform()`
`batch_size`	`int`	Number of mutations of the current best.	`1`
`init_population_size`	`int`	Initial population size (randomly generated).	`1`
`seed`	`int \| None`	Random seed.	`None`

Returns:

Type	Description
`Evolution`	An `Evolution` object.

Source code in pygx/algo/evolution/_hill_climb.py

def hill_climb(
    mutator: base.Mutator = mutators.Uniform(),
    batch_size: int = 1,
    init_population_size: int = 1,
    seed: int | None = None,
) -> base.Evolution:
    """Hill-Climbing algorithm, with an extra batched setting.

    Batched setting was shown to be effective in
    https://arxiv.org/pdf/1911.06317.pdf and https://arxiv.org/pdf/2003.01239.pdf,
    especially in noisy objective settings.

    Args:
      mutator: Mutator to use.
      batch_size: Number of mutations of the current best.
      init_population_size: Initial population size (randomly generated).
      seed: Random seed.

    Returns:
      An `Evolution` object.
    """
    return base.Evolution(
        reproduction=(
            selectors.Top(n=1)
            >> (mutator * batch_size)  # pytype: disable=unsupported-operands
        ),
        population_init=(pg.geno.Random(seed=seed), init_population_size),
        population_update=selectors.Top(n=1),
    )

neat ¶

neat(
    mutator: Mutator = Uniform(),
    population_size: int = 100,
    disjoint_coefficient: float = 1.0,
    matching_coefficient: float = 3.0,
    compatibility_threshold: float = 0.4,
    remaining_ratio: float = 0.6,
    seed: int | None = None,
) -> Evolution

NEAT Algorithm.

See http://nn.cs.utexas.edu/downloads/papers/stanley.cec02.pdf for the original paper.

NOTE(daiyip): PyGX supports search spaces that do not change during exploration. Therefore we do not grow the program (e.g. Neural Architecture in the paper) from a minimal program. Also, crossover can take place on any two individuals due to a fixed program structure, which will be implemented later. This algorithm illustrates how speciation is expressed in the compositional evolution framework.

Parameters:

Name	Type	Description	Default
`mutator`	`Mutator`	Mutator to use.	`Uniform()`
`population_size`	`int`	Population size for each generation.	`100`
`disjoint_coefficient`	`float`	Coefficient for DNA disjointness. Used for distance computations.	`1.0`
`matching_coefficient`	`float`	Coefficient for DNA matching. Used for distance computations.	`3.0`
`compatibility_threshold`	`float`	Threshold for max distances between two DNA in the same species.	`0.4`
`remaining_ratio`	`float`	Ratio of best individuals in a species to remain.	`0.6`
`seed`	`int \| None`	Random seed. If None, use the current system time.	`None`

Returns:

Type	Description
`Evolution`	An `Evolution` object that represents the NEAT algorithm.

Source code in pygx/algo/evolution/_neat.py

def neat(
    mutator: base.Mutator = mutators.Uniform(),
    population_size: int = 100,
    disjoint_coefficient: float = 1.0,
    matching_coefficient: float = 3.0,
    compatibility_threshold: float = 0.4,
    remaining_ratio: float = 0.6,
    seed: int | None = None,
) -> base.Evolution:
    """NEAT Algorithm.

    See http://nn.cs.utexas.edu/downloads/papers/stanley.cec02.pdf for the
    original paper.

    NOTE(daiyip): PyGX supports search spaces that do not change during
    exploration. Therefore we do not grow the program (e.g. Neural Architecture
    in the paper) from a minimal program. Also, crossover can take place on any
    two individuals due to a fixed program structure, which will be implemented
    later. This algorithm illustrates how speciation is expressed in the
    compositional evolution framework.

    Args:
      mutator: Mutator to use.
      population_size: Population size for each generation.
      disjoint_coefficient: Coefficient for DNA disjointness. Used for distance
        computations.
      matching_coefficient: Coefficient for DNA matching. Used for distance
        computations.
      compatibility_threshold: Threshold for max distances between two DNA in
        the same species.
      remaining_ratio: Ratio of best individuals in a species to remain.
      seed: Random seed. If None, use the current system time.

    Returns:
      An `Evolution` object that represents the NEAT algorithm.
    """
    # pylint: disable=no-value-for-parameter
    return base.Evolution(
        reproduction=(
            # Get the living species from the population.
            base.GlobalStateGetter(key='living_species')
            # Select species proportional to their scaled average fitness.
            >> selectors.Proportional(
                n=population_size, weights=scaled_average_fitness()
            )
            .for_each(
                # Randomly select 1 member from each species's top performers.
                (lambda x: x.members)
                >> selectors.Top(n=remaining_ratio)
                >> selectors.Random(n=1, seed=seed)
            )
            .flatten()
            >> mutator
        ),
        population_init=(pg.geno.Random(seed=seed), population_size),
        population_update=(
            # Keep only the individuals from the latest generation in the
            # population.
            selectors.Top(n=1, cluster=True, key=base.get_generation_id)
            # Speciate new individuals that are recently added to the population.
            >> speciate(
                distance=compatibility_distance(
                    disjoint_coefficient=disjoint_coefficient,
                    matching_coefficient=matching_coefficient,
                ),
                distance_threshold=compatibility_threshold,
            )
        ),
    )

nsga2 ¶

nsga2(
    mutator: Mutator = Uniform(),
    population_size: int = 100,
    seed: int | None = None,
) -> Evolution

NSGA-II: A multi-objective evolutionary search algorithm.

For reference and for citations, please use: https://ieeexplore.ieee.org/document/996017

Parameters:

Name	Type	Description	Default
`mutator`	`Mutator`	Mutator to use.	`Uniform()`
`population_size`	`int`	Population size, which will be used as both the batch size for proposing new individuals and the tournament size for finding the elites among recently added populations.	`100`
`seed`	`int \| None`	Random seed for initializing the population. If None, the system time will be used.	`None`

Returns:

Type	Description
`Evolution`	An `Evolution` object.

Source code in pygx/algo/evolution/_nsga2.py

def nsga2(
    mutator: base.Mutator = mutators.Uniform(),
    population_size: int = 100,
    seed: int | None = None,
) -> base.Evolution:
    """NSGA-II: A multi-objective evolutionary search algorithm.

    For reference and for citations, please use:
    https://ieeexplore.ieee.org/document/996017

    Args:
      mutator: Mutator to use.
      population_size: Population size, which will be used as both the batch
        size for proposing new individuals and the tournament size for finding
        the elites among recently added populations.
      seed: Random seed for initializing the population. If None, the system time
        will be used.

    Returns:
      An `Evolution` object.
    """
    # pylint: disable=no-value-for-parameter
    return base.Evolution(
        # Reproduction is to simply select the next elite and mutate it.
        reproduction=next_elite() >> mutator,
        population_init=(pg.geno.Random(seed=seed), population_size * 2),
        population_update=(
            # Take previous elites and unprocessed population as input to compute
            # the new elites.
            base.GlobalStateGetter(key='elites', default=[]) + base.Identity()
            # Apply non-dominated sorting, which returns a list of frontiers.
            # Each frontier is a list of DNA.
            # NOTE(daiyip): We converts the functor into a Lambda operation here
            # in order to call `for_each`.
            >> base.Lambda(fn=nondominated_sort())
            .for_each(
                # Apply crowding distance sort on each frontier, and flatten
                # their output into a single list.
                crowding_distance_sort()
            )
            .flatten()
            # Choose the first N (N=population_size) individuals as elites.
            # Reset the next cursor.
            >> (
                selectors.First(n=population_size)
                .as_global_state('elites')
                .set_global_state('elite_cursor', 0)
            )
            # We update the elites only when unprocessed population reaches the
            # population limit.
        ).if_true(lambda x: len(x) >= population_size),
        multi_objective=True,
    )

RegularizedEvolution ¶

RegularizedEvolution(
    mutator: Mutator = Uniform(),
    population_size: int = 100,
    tournament_size: int = 10,
    seed: int | None = None,
) -> Evolution

Regularized Evolution algorithm.

https://www.aaai.org/ojs/index.php/AAAI/article/view/4405.

Parameters:

Name	Type	Description	Default
`mutator`	`Mutator`	Mutator to use.	`Uniform()`
`population_size`	`int`	Population size. Must be larger than tournament size.	`100`
`tournament_size`	`int`	Tournament size.	`10`
`seed`	`int \| None`	Random seed. If None, the current system time is used as seed.	`None`

Returns:

Type	Description
`Evolution`	An `Evolution` object.

Source code in pygx/algo/evolution/_regularized_evolution.py

def regularized_evolution(
    mutator: base.Mutator = mutators.Uniform(),
    population_size: int = 100,
    tournament_size: int = 10,
    seed: int | None = None,
) -> base.Evolution:
    """Regularized Evolution algorithm.

    https://www.aaai.org/ojs/index.php/AAAI/article/view/4405.

    Args:
      mutator: Mutator to use.
      population_size: Population size. Must be larger than tournament size.
      tournament_size: Tournament size.
      seed: Random seed. If None, the current system time is used as seed.

    Returns:
      An `Evolution` object.
    """
    if tournament_size < 2:
        raise ValueError(
            f'`tournament_size` must be no less than 2. '
            f'Encountered: {tournament_size}'
        )
    if population_size < tournament_size:
        raise ValueError(
            f'The value of `population_size` ({population_size}) must be no '
            f'less than the value of `tournament_size` ({tournament_size}).'
        )
    return base.Evolution(
        reproduction=(
            selectors.Random(n=tournament_size, seed=seed)
            >> selectors.Top(n=1)
            >> mutator
        ),
        population_init=(pg.geno.Random(seed=seed), population_size),
        population_update=selectors.Last(n=population_size),
    )

regularized_evolution ¶

regularized_evolution(
    mutator: Mutator = Uniform(),
    population_size: int = 100,
    tournament_size: int = 10,
    seed: int | None = None,
) -> Evolution

Regularized Evolution algorithm.

https://www.aaai.org/ojs/index.php/AAAI/article/view/4405.

Parameters:

Name	Type	Description	Default
`mutator`	`Mutator`	Mutator to use.	`Uniform()`
`population_size`	`int`	Population size. Must be larger than tournament size.	`100`
`tournament_size`	`int`	Tournament size.	`10`
`seed`	`int \| None`	Random seed. If None, the current system time is used as seed.	`None`

Returns:

Type	Description
`Evolution`	An `Evolution` object.

Source code in pygx/algo/evolution/_regularized_evolution.py

def regularized_evolution(
    mutator: base.Mutator = mutators.Uniform(),
    population_size: int = 100,
    tournament_size: int = 10,
    seed: int | None = None,
) -> base.Evolution:
    """Regularized Evolution algorithm.

    https://www.aaai.org/ojs/index.php/AAAI/article/view/4405.

    Args:
      mutator: Mutator to use.
      population_size: Population size. Must be larger than tournament size.
      tournament_size: Tournament size.
      seed: Random seed. If None, the current system time is used as seed.

    Returns:
      An `Evolution` object.
    """
    if tournament_size < 2:
        raise ValueError(
            f'`tournament_size` must be no less than 2. '
            f'Encountered: {tournament_size}'
        )
    if population_size < tournament_size:
        raise ValueError(
            f'The value of `population_size` ({population_size}) must be no '
            f'less than the value of `tournament_size` ({tournament_size}).'
        )
    return base.Evolution(
        reproduction=(
            selectors.Random(n=tournament_size, seed=seed)
            >> selectors.Top(n=1)
            >> mutator
        ),
        population_init=(pg.geno.Random(seed=seed), population_size),
        population_update=selectors.Last(n=population_size),
    )

scalar_spec ¶

scalar_spec(value_spec: ValueSpec) -> ValueSpec

Returns the value spec for a field that accepts a scheduled value.

Use this to declare fields that accept either a plain value or a step-based schedule for it (a Scalar or any callable taking a step).

Parameters:

Name	Type	Description	Default
`value_spec`	`ValueSpec`	The value spec for the value produced at each step.	required

Returns:

Type	Description
`ValueSpec`	A value spec that accepts either the value itself or a callable that produces such a value from a step (an integer).

Source code in pygx/algo/scalars/_base.py

def scalar_spec(value_spec: pg.typing.ValueSpec) -> pg.typing.ValueSpec:
    """Returns the value spec for a field that accepts a scheduled value.

    Use this to declare fields that accept either a plain value or a
    step-based schedule for it (a `Scalar` or any callable taking a step).

    Args:
      value_spec: The value spec for the value produced at each step.

    Returns:
      A value spec that accepts either the value itself or a callable that
        produces such a value from a step (an integer).
    """
    return pg.typing.Union(
        [value_spec, pg.typing.Callable([pg.typing.Int()], returns=value_spec)]
    )

scalar_value ¶

scalar_value(value: Any, step: int) -> Any

Evaluates a scheduled value at a step.

Parameters:

Name	Type	Description	Default
`value`	`Any`	A plain value, or a callable (e.g. a `Scalar`) that maps a step to a value.	required
`step`	`int`	The step to evaluate the value at.	required

Returns:

Type	Description
`Any`	`value(step)` if `value` is callable; otherwise `value` unchanged.

Source code in pygx/algo/scalars/_base.py

def scalar_value(value: Any, step: int) -> Any:
    """Evaluates a scheduled value at a step.

    Args:
      value: A plain value, or a callable (e.g. a `Scalar`) that maps a step
        to a value.
      step: The step to evaluate the value at.

    Returns:
      `value(step)` if `value` is callable; otherwise `value` unchanged.
    """
    if callable(value):
        return value(step)
    return value

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2

pygx.algo.evolution¶

evolution ¶

Choice ¶

Concatenation ¶

Conditional ¶

Difference ¶

DNAOperation ¶

input_element_type property ¶

output_element_type property ¶

ElementWise ¶

Evolution ¶

population property ¶

global_state property ¶

recover ¶

Flatten ¶

GlobalStateGetter ¶

GlobalStateSetter ¶

Identity ¶

Intersection ¶

Inversion ¶

Lambda ¶

Mutator ¶

mutate ¶

mutate_list ¶

Operation ¶

input_element_type property ¶

output_element_type property ¶

operation_method property ¶

_on_input ¶

call ¶

with_prob ¶

if_true ¶

if_false ¶

for_each ¶

until_change ¶

flatten ¶

global_state ¶

as_global_state ¶

set_global_state ¶

Pipeline ¶

Power ¶

Recombinator ¶

recombine abstractmethod ¶

_on_input ¶

Repeat ¶

Selector ¶

select abstractmethod ¶

Slice ¶

SymmetricDifference ¶

Union ¶

UntilChange ¶

Scalar ¶

call abstractmethod ¶

floor ¶

ceil ¶

get_feedback_sequence_number ¶

get_fitness ¶

get_generation_id ¶

is_initial_population ¶

set_fitness ¶

hill_climb ¶

neat ¶

nsga2 ¶

RegularizedEvolution ¶

regularized_evolution ¶

scalar_spec ¶

scalar_value ¶

`pygx.algo.evolution`¶

input_element_type `property` ¶

output_element_type `property` ¶

population `property` ¶

global_state `property` ¶

input_element_type `property` ¶

output_element_type `property` ¶

operation_method `property` ¶

recombine `abstractmethod` ¶

select `abstractmethod` ¶

call `abstractmethod` ¶