pygx.algo.mutfun

General mutable functions for evolution and symbolic regression.

mutfun

General mutable functions for evolution and symbolic regression.

While users can use hyper primitives such as pg.oneof to fine-tune specific parts of pre-existing programs, there are other use cases where it's necessary to create a program from scratch or transform an existing program into something drastically different. For example, one may need to find an activation formula for neural networks or symbolically regress a function with only a few observations. To address these needs, pg.algo.mutfun has been introduced.

pg.algo.mutfun provides functions and instructions represented by symbolic objects that allow for maximum flexibility in manipulating a program. This includes but is not limited to inserting new lines, deleting existing ones, replacing operations, and creating new functions. With pg.algo.mutfun, users have access to APIs that make these tasks more manageable. They can easily identify all downstream instructions that depend on the current instruction, all upstream instructions that the current instruction depends on, or all defined variables up to current instruction.

A mutfun program is a Function object illustrated as below:

f = pg.algo.mutfun.Function('f',
    [
        pg.algo.mutfun.Assign('y', pg.algo.mutfun.Var('x') + 1),
        pg.algo.mutfun.Assign('z', pg.algo.mutfun.Var('x') ** 2),
        pg.algo.mutfun.Var('y') * pg.algo.mutfun.Var('z'),
    ], args=['x'])

assert f(2) == (2 + 1) * 2 ** 2
print(f)

which prints:

def f(x):
  y = x + 1
  z = x ** 2
  return y * z

The Function Regression notebook provides an example for evolving and doing symbolic regression on mutable functions.

Class hierarchy:

classDiagram
  Code <|-- SymbolDefinition
  Code <|-- Instruction
  SymbolDefinition <|-- Assign
  SymbolDefinition <|-- Function
  Instruction <|-- SymbolReference
  Instruction <|-- Operator
  Instruction <|-- UserDefinedInstructions
  SymbolReference <|-- Var
  SymbolReference <|-- FunctionCall
  Operator <|-- UnaryOperator
  Operator <|-- BinaryOperator
  Function *-- Code : body

Assign

Assign(name: str, value: Any, **kwargs)

Bases: SymbolDefinition

Assignment instruction.

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

def __init__(self, name: str, value: Any, **kwargs):
    super().__init__(name=name, value=value, **kwargs)

Code

Bases: Object

Interface for code entity.

eq=False lets Code instances be stored in dict/set by their ID. pg.eq/pg.ne remain available for symbolic comparison.

evaluate abstractmethod

evaluate(context: dict[str, Any]) -> Any

Evaluates current instruction with variable dictionary.

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

@abc.abstractmethod
def evaluate(self, context: dict[str, Any]) -> Any:
    """Evaluates current instruction with variable dictionary."""

python_repr abstractmethod

python_repr(block_indent: int = 0) -> str

Returns a Python code representation of current instruction.

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

@abc.abstractmethod
def python_repr(self, block_indent: int = 0) -> str:
    """Returns a Python code representation of current instruction."""

format

format(
    compact: bool = True, verbose: bool = False, root_indent: int = 0, **kwargs
) -> str

Overrides pg.Symbolic.format to support Python program representation.

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

def format(
    self,
    compact: bool = True,
    verbose: bool = False,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Overrides pg.Symbolic.format to support Python program representation."""
    if verbose:
        return super().format(
            compact, verbose, root_indent=root_indent, **kwargs
        )
    return self.python_repr(root_indent)

parent_code

parent_code() -> Optional[Code]

Returns the parent code entity of current code entity.

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

def parent_code(self) -> Optional['Code']:
    """Returns the parent code entity of current code entity."""
    parent = self.topo_parent
    while parent is not None and not isinstance(parent, Code):
        parent = parent.topo_parent
    return parent

parent_func

parent_func() -> Optional[Function]

Returns the parent function of current code entity.

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

def parent_func(self) -> Optional['Function']:
    """Returns the parent function of current code entity."""
    parent = self.topo_parent
    while parent is not None and not isinstance(parent, Function):
        parent = parent.topo_parent
    return parent

line

line() -> Code

Returns the top-level code entity of current line.

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

def line(self) -> 'Code':
    """Returns the top-level code entity of current line."""
    parent = self.topo_parent
    line = self
    while parent is not None and not isinstance(parent, Function):
        if isinstance(parent, Code):
            line = parent
        parent = parent.topo_parent
    return line

line_number

line_number() -> int

Returns the 0-based line number of current line within its function.

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

def line_number(self) -> int:
    """Returns the 0-based line number of current line within its function."""
    index = self.line().topo_path.key
    assert isinstance(index, int), index
    return index

preceding_lines

preceding_lines() -> Iterable[Code]

Iterates the preceding lines (earliest first).

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

def preceding_lines(self) -> Iterable['Code']:
    """Iterates the preceding lines (earliest first)."""
    parent_func = self.parent_func()
    if parent_func is not None:
        for i in range(self.line_number()):
            yield parent_func.body[i]

preceding_lines_reversed

preceding_lines_reversed() -> Iterable[Code]

Iterates the preceding lines in reversed order (recent first).

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

def preceding_lines_reversed(self) -> Iterable['Code']:
    """Iterates the preceding lines in reversed order (recent first)."""
    parent_func = self.parent_func()
    if parent_func is not None:
        for i in reversed(range(self.line_number())):
            yield parent_func.body[i]

succeeding_lines

succeeding_lines() -> Iterable[Code]

Iterates the top-level instructions of succeeding lines.

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

def succeeding_lines(self) -> Iterable['Code']:
    """Iterates the top-level instructions of succeeding lines."""
    parent_func = self.parent_func()
    if parent_func is not None:
        for i in range(self.line_number() + 1, len(parent_func.body)):
            yield parent_func.body[i]

input_vars

input_vars(transitive: bool = False) -> set[str]

Returns the names of variables that this code entity depends on.

Parameters:

Name	Type	Description	Default
transitive	bool	If True, variables that the direct inputs transitively depend on will be included.	False

Returns:

Type	Description
set[str]	A set of input variable names.

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

def input_vars(self, transitive: bool = False) -> set[str]:
    """Returns the names of variables that this code entity depends on.

    Args:
      transitive: If True, variables that the direct inputs transitively
        depend on will be included.

    Returns:
      A set of input variable names.
    """
    input_vars = set()

    def list_var_refs(k, v, p):
        del k, p
        # NOTE(daiyip): we do not step into function definitions as for now
        # closure is not supported.
        if isinstance(v, Function):
            return pg.TraverseAction.CONTINUE
        if isinstance(v, SymbolReference):
            input_vars.add(v.name)
        return pg.TraverseAction.ENTER

    pg.traverse(self, list_var_refs)

    if transitive:
        parent_func = self.parent_func()
        if parent_func is not None:
            unresolved_vars = input_vars.copy()
            for i in reversed(range(self.line_number())):
                line = parent_func.body[i]
                line_output_vars = line.output_vars()
                if line_output_vars & unresolved_vars:
                    line_input_vars = line.input_vars()
                    input_vars.update(line_input_vars)
                    unresolved_vars -= line_output_vars
                    unresolved_vars.update(line_input_vars)
            assert unresolved_vars.issubset(set(parent_func.args)), (
                unresolved_vars
            )
    return input_vars

output_vars

output_vars(transitive: bool = False) -> set[str]

Returns the names of variables that this code entity outputs.

Parameters:

Name	Type	Description	Default
transitive	bool	If True, variables computed from this entity's outputs by succeeding lines will be included.	False

Returns:

Type	Description
set[str]	A set of output variable names.

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

def output_vars(self, transitive: bool = False) -> set[str]:
    """Returns the names of variables that this code entity outputs.

    Args:
      transitive: If True, variables computed from this entity's outputs
        by succeeding lines will be included.

    Returns:
      A set of output variable names.
    """
    output_vars = set()

    def list_var_defs(k, v, p):
        del k, p
        if isinstance(v, SymbolDefinition):
            output_vars.add(v.name)
        # NOTE(daiyip): we do not step into function definitions as for now
        # closure is not supported.
        if isinstance(v, Function):
            return pg.TraverseAction.CONTINUE
        return pg.TraverseAction.ENTER

    pg.traverse(self.line(), list_var_defs)

    if transitive:
        parent_func = self.parent_func()
        if parent_func is not None:
            for i in range(self.line_number(), len(parent_func.body)):
                line = parent_func.body[i]
                line_input_vars = line.input_vars()
                if output_vars & line_input_vars:
                    output_vars.update(line.output_vars())
    return output_vars

seen_vars

seen_vars() -> set[str]

Returns the names of variables visible prior to this code entity.

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

def seen_vars(self) -> set[str]:
    """Returns the names of variables visible prior to this code entity."""
    seen = set()
    parent_func = self.parent_func()
    if parent_func is not None:
        seen.add(parent_func.name)
        seen.update(parent_func.args)
        for line in self.preceding_lines():
            seen.update(line.output_vars())
    return seen

input_defs

input_defs(transitive: bool = True) -> list[SymbolDefinition]

Returns the symbol definitions for the inputs of this code entity.

Parameters:

Name	Type	Description	Default
transitive	bool	If True, transitive inputs will be included. Otherwise, only the direct dependencies will be included.	True

Returns:

Type	Description
list[SymbolDefinition]	A list of SymbolDefinition in their declaration order that produce
list[SymbolDefinition]	the inputs required for current code entity.

Raises:

Type	Description
ValueError	If this entity is not inside a Function. Input dependencies are resolved against a function's argument names and preceding statements; the concept only applies inside a function.

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

def input_defs(self, transitive: bool = True) -> list['SymbolDefinition']:
    """Returns the symbol definitions for the inputs of this code entity.

    Args:
      transitive: If True, transitive inputs will be included.
        Otherwise, only the direct dependencies will be included.

    Returns:
      A list of `SymbolDefinition` in their declaration order that produce
      the inputs required for current code entity.

    Raises:
      ValueError: If this entity is not inside a `Function`. Input
        dependencies are resolved against a function's argument names and
        preceding statements; the concept only applies inside a function.
    """
    parent_func = self.parent_func()
    if parent_func is None:
        raise ValueError(
            'input_defs() requires this entity to be inside a Function; '
            f'no parent Function found for {self!r}.'
        )
    var_producers: dict[str, set[SymbolDefinition]] = {
        arg: set() for arg in parent_func.args
    }
    var_producers[parent_func.name] = set()

    def analyze_var_producers(k: pg.KeyPath, v: Any, p: pg.Symbolic):
        del k, p
        if v is self:
            return pg.TraverseAction.STOP
        if isinstance(v, SymbolDefinition):
            var_entry = {v}  # pyright: ignore[reportUnhashable]
            if transitive:
                for var_name in v.input_vars():
                    var_entry.update(var_producers[var_name])
            var_producers[v.name] = var_entry
        # NOTE(daiyip): do not enter child function definitions.
        if v is not parent_func and isinstance(v, Function):
            return pg.TraverseAction.CONTINUE
        return pg.TraverseAction.ENTER

    pg.traverse(parent_func, analyze_var_producers)

    dependencies: set[SymbolDefinition] = set()
    for var_name in self.input_vars():
        if var_name not in var_producers:
            raise ValueError(
                f'Undefined variable {repr(var_name)} found in function '
                f"'{parent_func.name}' line#{self.line_number()}"
            )
        dependencies.update(var_producers[var_name])
    return sorted(dependencies, key=lambda x: x.line_number())

output_refs

output_refs(transitive: bool = True) -> list[SymbolReference]

Returns the references to the symbols that this code outputs.

Parameters:

Name	Type	Description	Default
transitive	bool	If True, transitive symbol references will be included. Otherwise, only the direct dependencies will be included.	True

Returns:

Type	Description
list[SymbolReference]	A list of Var or FunctionCall objects in their definition
list[SymbolReference]	order that consume the outputs of current code entity. Users can
list[SymbolReference]	use parent_instruction or line to get their context.

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

def output_refs(self, transitive: bool = True) -> list['SymbolReference']:
    """Returns the references to the symbols that this code outputs.

    Args:
      transitive: If True, transitive symbol references will be included.
        Otherwise, only the direct dependencies will be included.

    Returns:
      A list of ``Var`` or ``FunctionCall`` objects in their definition
      order that consume the outputs of current code entity. Users can
      use `parent_instruction` or `line` to get their context.
    """
    parent_func = self.parent_func()
    references: list[SymbolReference] = []

    if parent_func is not None:
        output_vars = self.output_vars()

        def find_references(code: Code):
            refs = []

            def identify_reference(k, v, p):
                del k, p
                if isinstance(v, SymbolReference):
                    if v.name in output_vars:
                        refs.append(v)

            pg.traverse(code, identify_reference)
            return refs

        for line in self.succeeding_lines():
            ins_refs = find_references(line)
            references.extend(ins_refs)
            # Deal with reassignment.
            # For example:
            # ```
            # LN#1: x = 1
            # LN#2: x = 2
            # ```
            # After LN#2, the output vars from LN#1 should be cleared.
            new_assigned = line.output_vars()
            if ins_refs and transitive:
                output_vars.update(new_assigned)
            else:
                output_vars -= new_assigned
    return references

compile

compile() -> None

Validates this code entity; raises ValueError on undefined vars.

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

def compile(self) -> None:
    """Validates this code entity; raises ValueError on undefined vars."""
    if not self.seen_vars().issuperset(self.input_vars()):
        raise ValueError(
            f'Undefined variables {self.input_vars() - self.seen_vars()} '
            f"found at '{self.topo_path}'."
        )

Function

Function(name: str, body: list[Code], args: list[str] | None = None, **kwargs)

Bases: SymbolDefinition

A function that contains a list of instructions.

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

def __init__(
    self,
    name: str,
    body: list[Code],
    args: list[str] | None = None,
    **kwargs,
):
    if args is not None:
        kwargs['args'] = args
    super().__init__(name=name, body=body, **kwargs)

compile

compile()

Validates the function body; raises ValueError on undefined vars.

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

def compile(self):
    """Validates the function body; raises ValueError on undefined vars."""
    seen_vars = set(self.args)
    for line in self.body:
        if not seen_vars.issuperset(line.input_vars()):
            diff = line.input_vars() - seen_vars
            raise ValueError(
                f"Undefined variables {diff} found at '{line.topo_path}'."
            )
        seen_vars |= line.output_vars()

prune

prune()

Removes lines that don't contribute to the return value.

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

def prune(self):
    """Removes lines that don't contribute to the return value."""
    if self.body:
        effective_lines = {
            x.line_number() for x in self.body[-1].input_defs()
        }
        ineffective_lines = set(range(len(self.body) - 1)) - effective_lines
        for i in sorted(ineffective_lines, reverse=True):
            with pg.allow_writable_accessors(True):
                del self.body[i]

FunctionCall

FunctionCall(name: str, args: list[Any] | None = None, **kwargs)

Bases: SymbolReference

Function call.

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

def __init__(self, name: str, args: list[Any] | None = None, **kwargs):
    if args is not None:
        kwargs['args'] = args
    super().__init__(name=name, **kwargs)

Instruction

Bases: Code

Base class for all instructions.

parent_instruction

parent_instruction() -> Optional[Instruction]

Returns the parent instruction of current instruction.

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

def parent_instruction(self) -> Optional['Instruction']:
    """Returns the parent instruction of current instruction."""
    parent = self.topo_parent
    while parent is not None and not isinstance(parent, Instruction):
        parent = parent.topo_parent
    return parent

select_types classmethod

select_types(
    where: Callable[[type[Instruction]], bool] = lambda x: True,
) -> Iterable[type[Instruction]]

Selects all instruction types that match the condition.

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

@classmethod
def select_types(
    cls, where: Callable[[type['Instruction']], bool] = lambda x: True
) -> Iterable[type['Instruction']]:
    """Selects all instruction types that match the condition."""
    for _, t in pg.registered_types():
        if issubclass(t, Instruction) and t is not Instruction and where(t):
            yield t

SymbolDefinition

SymbolDefinition(name: str, **kwargs)

Bases: Code

Base class for symbol definition.

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

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

SymbolReference

SymbolReference(name: str, **kwargs)

Bases: Instruction

Base class for symbol references.

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

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

Var

Var(name: str, **kwargs)

Bases: SymbolReference

Reference to a variable by name.

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

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

Add

Add(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Add operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

BinaryOperator

BinaryOperator(x: Any, y: Any, **kwargs)

Bases: Operator

Base class for binary math operations.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Divide

Divide(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Divide operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Equals

Equals(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Equals operation.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

FloorDivide

FloorDivide(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Floor divide operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

GreaterThan

GreaterThan(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Greater than operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

LessThan

LessThan(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Less than operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Mod

Mod(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Mod operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Multiply

Multiply(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Multiply operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Negate

Negate(x: Any, **kwargs)

Bases: UnaryOperator

Negates the operand.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

NotEquals

NotEquals(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Not Equals operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Operator

Bases: Instruction

Base class for operators.

maybe_parenthesize

maybe_parenthesize(child: Any) -> str

Maybe add parenthesis to a child expression.

Source code in pygx/algo/mutfun/_basic_ops.py

def maybe_parenthesize(self, child: Any) -> str:
    """Maybe add parenthesis to a child expression."""
    if isinstance(child, Operator):
        assert self.ORDER is not None, self.__class__
        assert child.ORDER is not None, child.__class__
        if self.ORDER >= child.ORDER:
            return '(' + base.python_repr(child) + ')'
    return base.python_repr(child)

Power

Power(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Power operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Subtract

Subtract(x: Any, y: Any, **kwargs)

Bases: BinaryOperator

Subtract operator.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, y: Any, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

UnaryOperator

UnaryOperator(x: Any, **kwargs)

Bases: Operator

Base class for unary math operations.

Source code in pygx/algo/mutfun/_basic_ops.py

def __init__(self, x: Any, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

evaluate

evaluate(value: Any, context: dict[str, Any]) -> Any

Evaluates an instruction or a normal value.

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

def evaluate(value: Any, context: dict[str, Any]) -> Any:
    """Evaluates an instruction or a normal value."""
    if isinstance(value, Code):
        return value.evaluate(context)
    return value

python_repr

python_repr(value: Any, block_indent: int = 0) -> str

Returns Python code representation of a value.

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

def python_repr(value: Any, block_indent: int = 0) -> str:
    """Returns Python code representation of a value."""
    return indent(str(value), block_indent)

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