`pygx.tuning`¶

Distributed tuning with pluggable backends.

tuning ¶

Distributed tuning with pluggable backends.

pygx.iter provides an interface for sampling examples from a search space within a process. To support distributed tuning, PyGX introduces pygx.sample, which is almost identical but with more features:

Allow multiple worker processes (workers) to collaborate on a search with failover handling.
Each worker can process different trials, or can cowork on the same trials via work groups.
Provide APIs for communicating between the co-workers.
Provide API for retrieving the search results.
Provide a pluggable backend system for supporting user infrastructures.

Backend ¶

Backend(
    name: str | None = None,
    group: None | int | str = None,
    dna_spec: DNASpec | None = None,
    algorithm: DNAGenerator | None = None,
    metrics_to_optimize: Sequence[str] | None = None,
    early_stopping_policy: EarlyStoppingPolicy | None = None,
    num_examples: int | None = None,
    **kwargs: Any
)

Interface for the tuning backend.

Source code in pygx/tuning/_backend.py

def __init__(
    self,
    name: str | None = None,
    group: None | int | str = None,
    dna_spec: geno.DNASpec | None = None,
    algorithm: geno.DNAGenerator | None = None,
    metrics_to_optimize: Sequence[str] | None = None,
    early_stopping_policy: EarlyStoppingPolicy | None = None,
    num_examples: int | None = None,
    **kwargs: Any,
) -> None:
    del (
        name,
        group,
        dna_spec,
        algorithm,
        metrics_to_optimize,
        early_stopping_policy,
        num_examples,
        kwargs,
    )

create `classmethod` ¶

create(
    name: str | None,
    group: None | int | str,
    dna_spec: DNASpec,
    algorithm: DNAGenerator,
    metrics_to_optimize: Sequence[str],
    early_stopping_policy: EarlyStoppingPolicy | None = None,
    num_examples: int | None = None,
    **kwargs: Any
) -> Backend

Create an instance of Backend based on pg.sample arguments.

The default implementation is to pass through all the arguments to __init__ for creating an instance of the backend. Users can override.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	A string as a unique identifier for current sampling. Two separate calls to `pg.sample` with the same `name` (also the same algorithm) will share the same sampling queue, whose examples are proposed by the same search algorithm.	required
`group`	`None \| int \| str`	A string or integer as the group ID of current process in distributed sampling, which will be used to group different workers into co-worker groups. Workers with the same group id will work on the same trial. On the contrary, workers in different groups will always be working with different trials. If not specified, each worker in current sampling will be in different groups. `group` is usually used in the outer loops of nested search, in order to allow workers to work on the same higher-order item.	required
`dna_spec`	`DNASpec`	An `pg.DNASpec` object representing the search space.	required
`algorithm`	`DNAGenerator`	The search algorithm that samples the search space.	required
`metrics_to_optimize`	`Sequence[str]`	A sequence of string as the names of the metrics to be optimized by the algorithm, which is ['reward'] by default. When specified, it should have only 1 item for single-objective algorithm and can have multiple items for algorithms that support multi-objective optimization.	required
`early_stopping_policy`	`EarlyStoppingPolicy \| None`	An optional early stopping policy for user to tell if incremental evaluation (which reports multiple measurements) on each example can be early short circuited. After each call to `feedback.add_measurement`, users can use method `feedback.should_stop_early` to check whether current example worth further evaluation or not.	`None`
`num_examples`	`int \| None`	An optional integer as the max number of examples to sample. If None, sample will return an iterator of infinite examples.	`None`
`**kwargs`	`Any`	Additional keyword arguments passed through from `pg.sample` for the backend to interpret.	`{}`

Returns:

Type	Description
`Backend`	A `pg.tuning.Backend` object.

Source code in pygx/tuning/_backend.py

@classmethod
def create(
    cls,
    name: str | None,
    group: None | int | str,
    dna_spec: geno.DNASpec,
    algorithm: geno.DNAGenerator,
    metrics_to_optimize: Sequence[str],
    early_stopping_policy: EarlyStoppingPolicy | None = None,
    num_examples: int | None = None,
    **kwargs: Any,
) -> 'Backend':
    """Create an instance of `Backend` based on `pg.sample` arguments.

    The default implementation is to pass through all the arguments to
    `__init__` for creating an instance of the backend. Users can
    override.

    Args:
      name: A string as a unique identifier for current sampling. Two
        separate calls to `pg.sample` with the same `name` (also the same
        algorithm) will share the same sampling queue, whose examples are
        proposed by the same search algorithm.
      group: A string or integer as the group ID of current process in
        distributed sampling, which will be used to group different
        workers into co-worker groups. Workers with the same group id will
        work on the same trial. On the contrary, workers in different
        groups will always be working with different trials. If not
        specified, each worker in current sampling will be in different
        groups. `group` is usually used in the outer loops of nested
        search, in order to allow workers to work on the same higher-order
        item.
      dna_spec: An `pg.DNASpec` object representing the search space.
      algorithm: The search algorithm that samples the search space.
      metrics_to_optimize: A sequence of string as the names of the metrics
        to be optimized by the algorithm, which is ['reward'] by default.
        When specified, it should have only 1 item for single-objective
        algorithm and can have multiple items for algorithms that support
        multi-objective optimization.
      early_stopping_policy: An optional early stopping policy for user to
        tell if incremental evaluation (which reports multiple
        measurements) on each example can be early short circuited.
        After each call to `feedback.add_measurement`, users can use method
        `feedback.should_stop_early` to check whether current example worth
        further evaluation or not.
      num_examples: An optional integer as the max number of examples to
        sample. If None, sample will return an iterator of infinite
        examples.
      **kwargs: Additional keyword arguments passed through from
        `pg.sample` for the backend to interpret.

    Returns:
      A `pg.tuning.Backend` object.
    """
    return cls(  # pytype: disable=wrong-keyword-args
        name=name,
        group=group,
        dna_spec=dna_spec,
        algorithm=algorithm,
        metrics_to_optimize=metrics_to_optimize,
        early_stopping_policy=early_stopping_policy,
        num_examples=num_examples,
        **kwargs,
    )

poll_result `abstractmethod` `classmethod` ¶

poll_result(name: str, **kwargs) -> Result

Gets tuning result by a unique tuning identifier.

Source code in pygx/tuning/_backend.py

@classmethod
@abc.abstractmethod
def poll_result(cls, name: str, **kwargs) -> Result:
    """Gets tuning result by a unique tuning identifier."""

next `abstractmethod` ¶

next() -> Feedback

Get the feedback object for the next sample.

Source code in pygx/tuning/_backend.py

@abc.abstractmethod
def next(self) -> Feedback:
    """Get the feedback object for the next sample."""

EarlyStoppingPolicy ¶

EarlyStoppingPolicy(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: Object

Interface for early stopping policies.

An early stopping policy is passed to pg.sample via its early_stopping_policy argument. Workers consult it through feedback.should_stop_early() in between incremental measurements to decide whether to abandon (feedback.skip()) the current trial.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

dna_spec `property` ¶

dna_spec: DNASpec | None

The DNASpec this policy was set up with, or None before setup.

setup ¶

setup(dna_spec: DNASpec) -> None

Setup states of an early stopping policy based on dna_spec.

Parameters:

Name	Type	Description	Default
`dna_spec`	`DNASpec`	DNASpec for DNA to propose.	required

Raises:

Type	Description
`RuntimeError`	if dna_spec is not supported.

Source code in pygx/tuning/_early_stopping.py

def setup(self, dna_spec: geno.DNASpec) -> None:
    """Setup states of an early stopping policy based on dna_spec.

    Args:
      dna_spec: DNASpec for DNA to propose.

    Raises:
      RuntimeError: if dna_spec is not supported.
    """
    self._dna_spec = dna_spec

should_stop_early `abstractmethod` ¶

should_stop_early(trial: Trial) -> bool

Should stop the input trial early based on its measurements.

Source code in pygx/tuning/_early_stopping.py

@abc.abstractmethod
def should_stop_early(self, trial: Trial) -> bool:
    """Should stop the input trial early based on its measurements."""

recover ¶

recover(history: Iterable[Trial]) -> None

Recover states by replaying the trial history.

Subclass can override.

NOTE: recover will always be called before the first should_stop_early is called. It could be called multiple times if there are multiple sources of history, e.g: trials from a previous study and existing trials from current study.

The default behavior is to replay should_stop_early on every trial in the history whose status is 'COMPLETED', 'PENDING' or 'STOPPING'.

Parameters:

Name	Type	Description	Default
`history`	`Iterable[Trial]`	An iterable object of trials.	required

Source code in pygx/tuning/_early_stopping.py

def recover(self, history: Iterable[Trial]) -> None:
    """Recover states by replaying the trial history.

    Subclass can override.

    NOTE: `recover` will always be called before the first
    `should_stop_early` is called. It could be called multiple times if
    there are multiple sources of history, e.g: trials from a previous
    study and existing trials from current study.

    The default behavior is to replay `should_stop_early` on every trial
    in the history whose status is 'COMPLETED', 'PENDING' or 'STOPPING'.

    Args:
      history: An iterable object of trials.
    """
    for trial in history:
        if trial.status in ['COMPLETED', 'PENDING', 'STOPPING']:
            # TODO(daiyip): Stopped trials do not need to be fed back.
            self.should_stop_early(trial)

Feedback ¶

Feedback(metrics_to_optimize: Sequence[str])

Interface for the feedback object for a trial.

Feedback object is an agent to communicate to the search algorithm and other workers based on current trial, which includes:

Information about current example:
id: The ID of current example, starting from 1.
dna: The DNA for current example.
Methods to communicate with the search algorithm:
add_measurement: Add a measurement for current example. Multiple measurements can be added as progressive evaluation of the example, which can be used by the early stopping policy to suggest whether current evaluation can be stopped early.
done: Mark evaluation on current example as done, use the reward from the latest measurement to feedback to the algorithm, and move to the next example.
__call__: A shortcut method that calls add_measurement and done in sequence.
skip: Mark evaluation on current example as done, and move to the next example without providing feedback to the algorithm.
should_stop_early: Tell if progressive evaluation on current example can be stopped early.
end_loop: Mark the loop as done. All workers will get out of the loop after they finish evaluating their current examples.
Methods to publish information associated with current trial:
set_metadata: Set persistent metadata by key.
get_metadata: Get persistent metadata by key.
add_link: Add a related link by key.

Source code in pygx/tuning/_protocols.py

def __init__(self, metrics_to_optimize: Sequence[str]):
    super().__init__()
    self._metrics_to_optimize = metrics_to_optimize
    self._sample_time = time.time()

id `abstractmethod` `property` ¶

id: int

Gets the ID of current trial.

dna `abstractmethod` `property` ¶

dna: DNA

Gets the DNA of the example used in current trial.

checkpoint_to_warm_start_from `abstractmethod` `property` ¶

checkpoint_to_warm_start_from: str | None

Gets checkpoint path to warm start from.

add_measurement ¶

add_measurement(
    reward: None | float | Sequence[float] = None,
    metrics: dict[str, float] | None = None,
    step: int = 0,
    checkpoint_path: str | None = None,
    elapse_secs: float | None = None,
) -> None

Add a measurement for current trial.

This method can be called multiple times on the same trial, e.g:

for model, feedback in pg.sample(...):
  accuracy = train_and_evaluate(model, step=10)
  feedback.add_measurement(accuracy, step=10)
  accuracy = train_and_evaluate(model, step=25)
  feedback.add_measurement(accuracy, step=25)
  feedback.done()

Parameters:

Name	Type	Description	Default
`reward`	`None \| float \| Sequence[float]`	An optional float value as the reward for single-objective optimization, or a sequence of float values for multiple objectives optimization. In multiple-objective scenario, the float sequence will be paired up with the `metrics_to_optimize` argument of `pg.sample`, thus their length must be equal. Another way for providing reward for multiple-objective reward is through the `metrics` argument, which is a dict using metric name as key and its measure as value (the key should match with an element of the `metrics_to_optimize` argument). When multi-objective reward is provided from both the `reward` argument (via a sequence of float) and the `metrics` argument, their value should agree with each other.	`None`
`metrics`	`dict[str, float] \| None`	An optional dictionary of string to float as metrics. It can be used to provide metrics for multi-objective optimization, and/or carry additional metrics for study analysis.	`None`
`step`	`int`	An optional integer as the step (e.g. step for model training), at which the measurement applies. When a trial is completed, the measurement at the largest step will be chosen as the final measurement to feed back to the controller.	`0`
`checkpoint_path`	`str \| None`	An optional string as the checkpoint path produced from the evaluation (e.g. training a model), which can be used in transfer learning.	`None`
`elapse_secs`	`float \| None`	Time spent on evaluating current example so far. If None, it will be automatically computed by the backend.	`None`

Source code in pygx/tuning/_protocols.py

def add_measurement(
    self,
    reward: None | float | Sequence[float] = None,
    metrics: dict[str, float] | None = None,
    step: int = 0,
    checkpoint_path: str | None = None,
    elapse_secs: float | None = None,
) -> None:
    """Add a measurement for current trial.

    This method can be called multiple times on the same trial, e.g:

    ```python
    for model, feedback in pg.sample(...):
      accuracy = train_and_evaluate(model, step=10)
      feedback.add_measurement(accuracy, step=10)
      accuracy = train_and_evaluate(model, step=25)
      feedback.add_measurement(accuracy, step=25)
      feedback.done()
    ```

    Args:
      reward: An optional float value as the reward for single-objective
        optimization, or a sequence of float values for multiple
        objectives optimization. In multiple-objective scenario, the float
        sequence will be paired up with the `metrics_to_optimize` argument
        of `pg.sample`, thus their length must be equal.
        Another way for providing reward for multiple-objective reward is
        through the `metrics` argument, which is a dict using metric name
        as key and its measure as value (the key should match with an
        element of the `metrics_to_optimize` argument). When
        multi-objective reward is provided from both the `reward` argument
        (via a sequence of float) and the `metrics` argument, their value
        should agree with each other.
      metrics: An optional dictionary of string to float as metrics. It can
        be used to provide metrics for multi-objective optimization, and/or
        carry additional metrics for study analysis.
      step: An optional integer as the step (e.g. step for model training),
        at which the measurement applies. When a trial is completed, the
        measurement at the largest step will be chosen as the final
        measurement to feed back to the controller.
      checkpoint_path: An optional string as the checkpoint path produced
        from the evaluation (e.g. training a model), which can be used in
        transfer learning.
      elapse_secs: Time spent on evaluating current example so far. If None,
        it will be automatically computed by the backend.
    """
    metrics_to_optimize = self._metrics_to_optimize
    metrics = metrics or {}

    if isinstance(reward, (list, tuple)):
        rewards = reward
        if len(rewards) != len(metrics_to_optimize):
            raise ValueError(
                f'The number of items in the reward ({rewards!r}) computed by the '
                f'controller does not match with the number of metrics to '
                f'optimize ({metrics_to_optimize!r}).'
            )
        for k, v in zip(metrics_to_optimize, rewards):
            if k in metrics and metrics[k] != v:
                raise ValueError(
                    f"The value for metric {k} is provided from both the 'reward' "
                    f"and the 'metrics' arguments with different values: "
                    f'{[v, metrics[k]]!r}.'
                )
            metrics[k] = v
        reward = metrics.pop('reward', None)
    elif reward is not None:
        # `reward` is narrowed to non-Sequence here by the prior isinstance.
        reward = float(reward)  # pyright: ignore[reportArgumentType]

    for metric_name in metrics_to_optimize:
        if metric_name == 'reward':
            if reward is None:
                raise ValueError(
                    "'reward' must be provided as it is a goal to optimize."
                )
        elif metric_name in metrics:
            metrics[metric_name] = float(metrics[metric_name])
        else:
            raise ValueError(
                f'Metric {metric_name!r} must be provided '
                f'as it is a goal to optimize.'
            )

    if len(metrics_to_optimize) == 1 and metrics_to_optimize[0] != 'reward':
        if reward is None:
            reward = metrics[metrics_to_optimize[0]]
        else:
            raise ValueError(
                f"'reward' {reward!r} is provided while it is "
                f'not a goal to optimize.'
            )

    if elapse_secs is None:
        elapse_secs = time.time() - self._sample_time

    self._add_measurement(
        reward, metrics, step, checkpoint_path, elapse_secs
    )

get_trial `abstractmethod` ¶

get_trial() -> Trial

Gets current Trial.

Returns:

Type	Description
`Trial`	An up-to-date `Trial` object. A distributed tuning backend should
`Trial`	make sure the return value is up-to-date not only locally, but among
`Trial`	different workers.

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def get_trial(self) -> Trial:
    """Gets current Trial.

    Returns:
      An up-to-date `Trial` object. A distributed tuning backend should
      make sure the return value is up-to-date not only locally, but among
      different workers.
    """

set_metadata `abstractmethod` ¶

set_metadata(key: str, value: Any, per_trial: bool = True) -> None

Sets metadata for current trial or current sampling.

Metadata can be used in two use cases:

Worker processes that co-work on the same trial can use meta-data to communicate with each other.
Worker use metadata as a persistent store to save information for current trial, which can be retrieved via poll_result method later.

Parameters:

Name	Type	Description	Default
`key`	`str`	A string as key to metadata.	required
`value`	`Any`	A value that can be serialized by `pg.to_json_str`.	required
`per_trial`	`bool`	If True, the key is set per current trial. Otherwise, it is set per current sampling loop.	`True`

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def set_metadata(
    self, key: str, value: Any, per_trial: bool = True
) -> None:
    """Sets metadata for current trial or current sampling.

    Metadata can be used in two use cases:

     * Worker processes that co-work on the same trial can use meta-data to
       communicate with each other.
     * Worker use metadata as a persistent store to save information for
       current trial, which can be retrieved via `poll_result` method later.

    Args:
      key: A string as key to metadata.
      value: A value that can be serialized by `pg.to_json_str`.
      per_trial: If True, the key is set per current trial. Otherwise, it
        is set per current sampling loop.
    """

get_metadata `abstractmethod` ¶

get_metadata(key: str, per_trial: bool = True) -> Any | None

Gets metadata for current trial or current sampling.

Parameters:

Name	Type	Description	Default
`key`	`str`	A string as key to metadata.	required
`per_trial`	`bool`	If True, the key is retrieved per current trial. Otherwise, it is retrieved per current sampling.	`True`

Returns:

Type	Description
`Any \| None`	A value that can be deserialized by `pg.from_json_str`.

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def get_metadata(self, key: str, per_trial: bool = True) -> Any | None:
    """Gets metadata for current trial or current sampling.

    Args:
      key: A string as key to metadata.
      per_trial: If True, the key is retrieved per current trial.
        Otherwise, it is retrieved per current sampling.

    Returns:
      A value that can be deserialized by `pg.from_json_str`.
    """

add_link `abstractmethod` ¶

add_link(name: str, url: str) -> None

Adds a related link to current trial.

Added links can be retrieved from the Trial.related_links property via pg.poll_result.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name for the related link.	required
`url`	`str`	URL for this link.	required

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def add_link(self, name: str, url: str) -> None:
    """Adds a related link to current trial.

    Added links can be retrieved from the `Trial.related_links` property via
    `pg.poll_result`.

    Args:
      name: Name for the related link.
      url: URL for this link.
    """

done `abstractmethod` ¶

done(
    metadata: dict[str, Any] | None = None,
    related_links: dict[str, str] | None = None,
) -> None

Marks current trial as done.

Parameters:

Name	Type	Description	Default
`metadata`	`dict[str, Any] \| None`	Additional metadata to add to current trial.	`None`
`related_links`	`dict[str, str] \| None`	Additional links to add to current trial.	`None`

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def done(
    self,
    metadata: dict[str, Any] | None = None,
    related_links: dict[str, str] | None = None,
) -> None:
    """Marks current trial as done.

    Args:
      metadata: Additional metadata to add to current trial.
      related_links: Additional links to add to current trial.
    """

skip `abstractmethod` ¶

skip(reason: str | None = None) -> None

Moves to the next example without feeding back to the algorithm.

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def skip(self, reason: str | None = None) -> None:
    """Moves to the next example without feeding back to the algorithm."""

should_stop_early `abstractmethod` ¶

should_stop_early() -> bool

Whether progressive evaluation can stop early on current trial.

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def should_stop_early(self) -> bool:
    """Whether progressive evaluation can stop early on current trial."""

end_loop `abstractmethod` ¶

end_loop() -> None

Ends current sampling loop.

All workers will exit the pg.sample loop after finishing their current trials.

Source code in pygx/tuning/_protocols.py

@abc.abstractmethod
def end_loop(self) -> None:
    """Ends current sampling loop.

    All workers will exit the `pg.sample` loop after finishing their
    current trials.
    """

skip_on_exceptions ¶

skip_on_exceptions(
    exceptions: Sequence[type[Exception] | tuple[type[Exception], str]],
) -> AbstractContextManager[None]

Returns a context manager to skip trial on user-specified exceptions.

When a matching exception is raised inside the scope, the trial is skipped (feedback.skip) with the traceback as the skip reason, and the exception is not re-raised.

Example:

with feedback.skip_on_exceptions((ValueError, KeyError)):
  ...

with feedback.skip_on_exceptions(((ValueError, 'bad value for .*'),
                                  (ValueError, '.* invalid range'),
                                  TypeError)):
  ...

Parameters:

Name	Type	Description	Default
`exceptions`	`Sequence[type[Exception] \| tuple[type[Exception], str]]`	A sequence of (exception type, or exception type plus regular expression for error message).	required

Returns:

Type	Description
`AbstractContextManager[None]`	A context manager for skipping trials on user-specified exceptions.

Source code in pygx/tuning/_protocols.py

def skip_on_exceptions(
    self,
    exceptions: Sequence[type[Exception] | tuple[type[Exception], str]],
) -> contextlib.AbstractContextManager[None]:
    """Returns a context manager to skip trial on user-specified exceptions.

    When a matching exception is raised inside the scope, the trial is
    skipped (`feedback.skip`) with the traceback as the skip reason, and
    the exception is not re-raised.

    **Example:**

    ```python
    with feedback.skip_on_exceptions((ValueError, KeyError)):
      ...

    with feedback.skip_on_exceptions(((ValueError, 'bad value for .*'),
                                      (ValueError, '.* invalid range'),
                                      TypeError)):
      ...
    ```

    Args:
      exceptions: A sequence of (exception type, or exception type plus
        regular expression for error message).

    Returns:
      A context manager for skipping trials on user-specified exceptions.
    """

    def skip_on_exception(unused_error):
        error_stack = traceback.format_exc()
        logging.warning(
            'Skipping trial on unhandled exception: %s', error_stack
        )
        self.skip(error_stack)

    return error_handling.catch_errors(exceptions, skip_on_exception)

ignore_race_condition ¶

ignore_race_condition() -> Iterator[None]

Context manager for ignoring RaceConditionError within the scope.

Race condition may happen when multiple workers are working on the same trial (e.g. paired train/eval processes). Assuming there are two co-workers (X and Y), common race conditions are:

1) Both X and Y call feedback.done or feedback.skip to the same trial. 2) X calls feedback.done/feedback.skip, then Y calls feedback.add_measurement.

Users can use this context manager to simplify the code for handling multiple co-workers. (See the group argument of pg.sample)

Example:

import threading

feedback = ...
def thread_fun():
  with feedback.ignore_race_condition():
    feedback.add_measurement(0.1)

    # Multiple workers working on the same trial might trigger this code
    # from different processes.
    feedback.done()

x = threading.Thread(target=thread_fun)
x.start()
y = threading.Thread(target=thread_fun)
y.start()

Yields:

Type	Description
`None`	None.

Source code in pygx/tuning/_protocols.py

@contextlib.contextmanager
def ignore_race_condition(self) -> Iterator[None]:
    """Context manager for ignoring RaceConditionError within the scope.

    Race condition may happen when multiple workers are working on the
    same trial (e.g. paired train/eval processes). Assuming there are two
    co-workers (X and Y), common race conditions are:

    1) Both X and Y call `feedback.done` or `feedback.skip` to the same
       trial.
    2) X calls `feedback.done`/`feedback.skip`, then Y calls
     `feedback.add_measurement`.

    Users can use this context manager to simplify the code for handling
    multiple co-workers. (See the `group` argument of `pg.sample`)

    **Example:**

    ```python
    import threading

    feedback = ...
    def thread_fun():
      with feedback.ignore_race_condition():
        feedback.add_measurement(0.1)

        # Multiple workers working on the same trial might trigger this code
        # from different processes.
        feedback.done()

    x = threading.Thread(target=thread_fun)
    x.start()
    y = threading.Thread(target=thread_fun)
    y.start()
    ```

    Yields:
      None.
    """
    try:
        yield
    except RaceConditionError:
        pass

Measurement ¶

Measurement(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: _DataEntity

Measurement of a trial at certain step.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

RaceConditionError ¶

Bases: RuntimeError

Race condition error.

This error will be raised when the operations made to Feedback indicate a race condition. There are possible scenarios that may lead to such race conditions, which happen among multiple co-workers (taking X and Y for example) on the same trial:

X calls feedback.done/feedback.skip, then Y calls feedback.add_measurement.

It can be suppressed via Feedback.ignore_race_condition.

Result ¶

Bases: Formattable

Interface for tuning result.

metadata `abstractmethod` `property` ¶

metadata: dict[str, Any]

Returns the metadata of current sampling.

is_active `abstractmethod` `property` ¶

is_active: bool

Returns whether the tuner task is active.

last_updated `abstractmethod` `property` ¶

last_updated: datetime | None

Last updated time.

trials `abstractmethod` `property` ¶

trials: list[Trial]

Retrieve all trials.

best_trial `abstractmethod` `property` ¶

best_trial: Trial | None

Get best trial so far.

Trial ¶

Trial(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: _DataEntity

Metadata of a trial.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

get_reward_for_feedback ¶

get_reward_for_feedback(
    metric_names: Sequence[str] | None = None,
) -> None | float | tuple[float, ...]

Returns the reward(s) to feed back to the search algorithm.

Parameters:

Name	Type	Description	Default
`metric_names`	`Sequence[str] \| None`	Names of the metrics to return, where `'reward'` refers to the reward value itself. If None, the final measurement's reward is returned.	`None`

Returns:

Type	Description
`None \| float \| tuple[float, ...]`	None if the trial is not completed or is infeasible. Otherwise a
`None \| float \| tuple[float, ...]`	float (single metric) or a tuple of floats (multiple metrics) taken
`None \| float \| tuple[float, ...]`	from the final measurement.

Raises:

Type	Description
`ValueError`	If a requested metric does not exist in the final measurement.

Source code in pygx/tuning/_protocols.py

def get_reward_for_feedback(
    self, metric_names: Sequence[str] | None = None
) -> None | float | tuple[float, ...]:
    """Returns the reward(s) to feed back to the search algorithm.

    Args:
      metric_names: Names of the metrics to return, where `'reward'`
        refers to the reward value itself. If None, the final
        measurement's reward is returned.

    Returns:
      None if the trial is not completed or is infeasible. Otherwise a
      float (single metric) or a tuple of floats (multiple metrics) taken
      from the final measurement.

    Raises:
      ValueError: If a requested metric does not exist in the final
        measurement.
    """
    if self.status != 'COMPLETED' or self.infeasible:
        return None
    assert self.final_measurement is not None
    measurement = self.final_measurement
    if metric_names is None:
        return measurement.reward
    assert metric_names, metric_names
    metric_values = []
    for metric_name in metric_names:
        if metric_name == 'reward':
            v = measurement.reward
        else:
            v = measurement.metrics.get(metric_name, None)
        if v is None:
            raise ValueError(
                f'Metric {metric_name!r} does not exist in final '
                f'measurement {measurement!r} in trial {self.id}.'
            )
        metric_values.append(v)
    return (
        tuple(metric_values) if len(metric_values) > 1 else metric_values[0]
    )

add_backend ¶

add_backend(backend_name: str)

Class decorator that registers a Backend subclass under a name.

The registered name can then be passed as the backend argument of pg.sample / pg.poll_result, or to pg.tuning.set_default_backend. Registering an existing name replaces the previous backend.

Example:

@pg.tuning.add_backend('my-backend')
class MyBackend(pg.tuning.Backend):
  ...

Parameters:

Name	Type	Description	Default
`backend_name`	`str`	Name to register the decorated class under.	required

The returned class decorator raises TypeError if the decorated class is not a pg.tuning.Backend subclass.

Source code in pygx/tuning/_backend.py

def add_backend(backend_name: str):
    """Class decorator that registers a `Backend` subclass under a name.

    The registered name can then be passed as the `backend` argument of
    `pg.sample` / `pg.poll_result`, or to `pg.tuning.set_default_backend`.
    Registering an existing name replaces the previous backend.

    **Example:**

    ```python
    @pg.tuning.add_backend('my-backend')
    class MyBackend(pg.tuning.Backend):
      ...
    ```

    Args:
      backend_name: Name to register the decorated class under.

    The returned class decorator raises `TypeError` if the decorated class
    is not a `pg.tuning.Backend` subclass.
    """

    def _decorator(backend_cls):
        if not issubclass(backend_cls, Backend):
            raise TypeError(
                f'{backend_cls!r} is not a `pg.tuning.Backend` subclass.'
            )
        _backend_registry[backend_name] = backend_cls
        return backend_cls

    return _decorator

available_backends ¶

available_backends() -> list[str]

Gets available backend names.

Source code in pygx/tuning/_backend.py

def available_backends() -> list[str]:
    """Gets available backend names."""
    return list(_backend_registry.keys())

default_backend ¶

default_backend() -> str

Gets the default tuning backend name.

Source code in pygx/tuning/_backend.py

def default_backend() -> str:
    """Gets the default tuning backend name."""
    return _default_backend_name

poll_result ¶

poll_result(name: str, backend: str | None = None, **kwargs) -> Result

Gets the result of a sampling loop by its unique name.

pg.poll_result returns a live pg.tuning.Result view of the sampling identified by name (the name passed to pg.sample), exposing its trials, metadata and best trial so far. It can be called while the sampling is still running or after it completed. backend selects the tuning backend to poll (the default backend when None); extra keyword arguments are forwarded to the backend's poll_result classmethod.

Example:

import pygx as pg

for example, feedback in pg.sample(
    pg.Dict(x=pg.oneof([1, 2, 3])),
    pg.geno.Random(),
    num_examples=5,
    name='my_search'):
  feedback(reward=float(example.x))

result = pg.poll_result('my_search')
print(result.best_trial.id, result.best_trial.final_measurement.reward)

Returns:

Type	Description
`Result`	A `pg.tuning.Result` object for the sampling loop named `name`.

Raises:

Type	Description
`ValueError`	If the backend is unknown, or no sampling with `name` exists in the backend.

Source code in pygx/tuning/_backend.py

def poll_result(name: str, backend: str | None = None, **kwargs) -> Result:
    """Gets the result of a sampling loop by its unique name.

    `pg.poll_result` returns a live `pg.tuning.Result` view of the sampling
    identified by `name` (the `name` passed to `pg.sample`), exposing its
    trials, metadata and best trial so far. It can be called while the
    sampling is still running or after it completed. `backend` selects the
    tuning backend to poll (the default backend when None); extra keyword
    arguments are forwarded to the backend's `poll_result` classmethod.

    **Example:**

    ```python
    import pygx as pg

    for example, feedback in pg.sample(
        pg.Dict(x=pg.oneof([1, 2, 3])),
        pg.geno.Random(),
        num_examples=5,
        name='my_search'):
      feedback(reward=float(example.x))

    result = pg.poll_result('my_search')
    print(result.best_trial.id, result.best_trial.final_measurement.reward)
    ```

    Returns:
      A `pg.tuning.Result` object for the sampling loop named `name`.

    Raises:
      ValueError: If the backend is unknown, or no sampling with `name`
        exists in the backend.
    """
    return get_backend_cls(backend or default_backend()).poll_result(
        name, **kwargs
    )

set_default_backend ¶

set_default_backend(backend_name: str)

Sets the process-wide default tuning backend name.

The default backend (initially 'in-memory') is used by pg.sample and pg.poll_result when their backend argument is None.

Raises:

Type	Description
`ValueError`	If no backend is registered under `backend_name`.

Source code in pygx/tuning/_backend.py

def set_default_backend(backend_name: str):
    """Sets the process-wide default tuning backend name.

    The default backend (initially `'in-memory'`) is used by `pg.sample` and
    `pg.poll_result` when their `backend` argument is None.

    Raises:
      ValueError: If no backend is registered under `backend_name`.
    """
    if backend_name not in _backend_registry:
        raise ValueError(f'Backend {backend_name!r} does not exist.')
    global _default_backend_name  # pylint: disable=global-statement
    _default_backend_name = backend_name

sample ¶

sample(
    space: HyperValue | DynamicEvaluationContext | DNASpec,
    algorithm: DNAGenerator,
    num_examples: int | None = None,
    early_stopping_policy: EarlyStoppingPolicy | None = None,
    where: Callable[[HyperPrimitive], bool] | None = None,
    name: str | None = None,
    group: None | int | str = None,
    backend: str | None = None,
    metrics_to_optimize: Sequence[str] | None = None,
    **kwargs: Any
) -> Iterator[tuple[Any, Feedback]]

Yields an example and its feedback sampled from a hyper value.

Example 1: sample a search space defined by a symbolic hyper value:

import math
import pygx as pg

for example, feedback in pg.sample(
    pg.Dict(x=pg.floatv(-1, 1)),
    pg.geno.Random(),
    num_examples=10,
    name='my_search'):

  # We can access trial ID (starting from 1) and DNA from the feedback.
  print(feedback.id, feedback.dna)

  # We can report the reward computed on the example using
  # `feedback.add_measurement`, which can be called
  # multiple times to report the rewards incrementally.
  # Once a trial is done, we call `feedback.done` to mark evaluation on
  # current example as completed, or use `feedback.skip` to move to the
  # next sample without passing any feedback to the algorithm.
  # Without `feedback.done` or `feedback.skip`, the same trial will be
  # iterated over and over again for failover handling purpose.
  # Besides the reward and the step, metrics and checkpoint can be added
  # to each measurement. Additional meta-data and related links (URLs) can
  # be passed to `feedback.done` which can be retrieved via
  # `pg.poll_result` later.
  if example.x >= 0:
    # If we only want to add one measurement for each example, a
    # shortcut expression for the next two lines can be written as
    # follows:
    #   `feedback(reward=math.sqrt(example.x), step=1)`
    feedback.add_measurement(reward=math.sqrt(example.x), step=1)
    feedback.done()
  else:
    feedback.skip()

  # IMPORTANT: to stop the loop among all workers, we can call
  # `feedback.end_loop`. As a result, each worker will quit their loop
  # after current iteration, while using `break` in the for-loop is
  # only effective for the local process rather than remotely.
  if feedback.id > 1000:
    feedback.end_loop()

# At any time, we can poll the search result via `pg.poll_result`, please
# see `pg.tuning.Result` for more details.
result = pg.poll_result('my_search')
print(result.best_trial)

Example 2: sample a search space defined by pg.hyper.trace:

import pygx as pg

def fun():
  return pg.oneof([
    lambda: pg.oneof([1, 2, 3]),
    lambda: pg.floatv(0.1, 1.0),
    3]) + sum(pg.manyof(2, [1, 2, 3]))

for example, feedback in pg.sample(
    pg.hyper.trace(fun),
    pg.geno.Random(),
    num_examples=10,
    name='my_search'):
  # When space is a `pg.hyper.DynamicEvaluationContext` object,
  # the `example` yielded at each iteration is a context manager under which
  # the hyper primitives (e.g. pg.oneof) will be materialized into concrete
  # values according to the controller decision.
  with example():
    reward = fun()
  feedback(reward)

Example 3: sample DNAs from an abstract search space represented by pg.DNASpec:

import pygx as pg

def evaluate_dna(dna):
  return sum(dna.to_numbers())

for dna, feedback in pg.sample(
    pg.dna_spec(pg.List([pg.oneof(range(3)) for _ in range(5)])),
    pg.geno.Random(),
    num_examples=10,
    name='my_search'):
  reward = evaluate_dna(dna)
  feedback(reward)

Using pg.sample in distributed environment

pg.sample is designed with distributed sampling in mind, in which multiple processes can work on the trials of the same sampling loop. While the default 'in-memory' backend works only within a single process without failover handling, other backends may support distributed computing environment with persistent state. Nevertheless, the pg.sample API is the same among different backends, users can switch the backend easily by passing a different value to the backend argument, or set the default value globally via pg.tuning.set_default_backend.

Identifying a sampling

To identify a distributed loop, a unique name is introduced, which will also be used to poll the latest sampling result via pg.poll_result.

Failover handling

In a distributed setup, worker processes may incidentally die and restart. Unless a trial is explicitly marked as done (via feedback(reward) or feedback.done()) or skipped (via feedback.skip()), a worker will try to resume its work on the trial from where it left off.

Workgroup

In a distributed setup, worker processes may or may not work on the same trials. Worker group is introduced to serve this purpose, which is identified by an integer or a string named group. If group is not specified or having different values among the workers, every worker will work on different trials of the loop. On the contrary, workers having the same group will co-work on the same trials. Group is useful when evaluation on one example can be parallelized - for example - an example in the outer loop of a nested search. However, feedback on the same example should be fed back to the search algorithm only once. Therefore, workers in the same group need to communicate with each other to avoid duplicated evaluation and feedbacks. To facilitate such communication, per-trial metadata is supported, and can be accessed via feedback.set_metadata/get_metadata methods. The consistency in reading and writing the metadata is defined by the backend used. For the 'in-memory' backend, all the trials and their metadata is stored in memory, thus will be lost if the process gets restarted. On the contrary, the backends built on distributed computing environment may store both the trials and metadata in a persistent storage with varying read/write QPS and read/write consistency guarantees.

Switch between backends

The backend argument of pg.sample lets users choose a backend used in current sampling loop. Users can use different backends in the same process to achieve a best performance trade-off, e.g., using in-memory backend when the communication cost outweighs the redundant evaluation cost upon worker failure.

Helper function pygx.tuning.set_default_backend is introduced to set the default tuning backend for the entire process.

Parameters:

Name	Type	Description	Default
`space`	`HyperValue \| DynamicEvaluationContext \| DNASpec`	One of (a hyper value, a `pg.hyper.DynamicEvaluationContext`, a `pg.DNASpec`) to sample from. A hyper value is an object with to-be-determined values specified by `pg.oneof`, `pg.manyof`, `pg.floatv` and etc, representing a search space. A `pg.hyper.DynamicEvaluationContext` object represents a search space that is traced via dynamic evaluation. A `pg.DNASpec` represents an abstract search space that emits DNAs.	required
`algorithm`	`DNAGenerator`	The search algorithm that samples the search space. For example: `pg.geno.Random()`, `pg.algo.evolution.regularized_evolution(...)`, and etc.	required
`num_examples`	`int \| None`	An optional integer as the max number of examples to sample. If None, sample will return an iterator of infinite examples.	`None`
`early_stopping_policy`	`EarlyStoppingPolicy \| None`	An optional early stopping policy for user to tell if incremental evaluation (which reports multiple measurements) on each example can be early short circuited. After each call to `feedback.add_measurement`, users can use method `feedback.should_stop_early` to check whether current example worth further evaluation or not.	`None`
`where`	`Callable[[HyperPrimitive], bool] \| None`	Function to filter the hyper values. If None, all decision points from the `space` will be included for the algorithm to make decisions. Otherwise only the decision points on which 'where' returns True will be included. The remaining decision points are passed through in the example intact, so each example is itself a sub-space of `space`. `where` is usually used in nested search flows. See `pg.hyper.template` for details.	`None`
`name`	`str \| None`	A string as a unique identifier for current sampling. Two separate calls to `pg.sample` with the same `name` (also the same algorithm) will share the same sampling queue, whose examples are proposed by the same search algorithm.	`None`
`group`	`None \| int \| str`	A string or integer as the group ID of current process in distributed sampling, which will be used to group different workers into co-worker groups. Workers with the same group id will work on the same trial. On the contrary, workers in different groups will always be working with different trials. If not specified, each worker in current sampling will be in different groups. `group` is usually used in the outer loops of nested search, in order to allow workers to work on the same higher-order item.	`None`
`backend`	`str \| None`	An optional string to specify the backend name for sampling. if None, the default backend set by `pg.tuning.set_default_backend` will be used.	`None`
`metrics_to_optimize`	`Sequence[str] \| None`	A sequence of string as the names of the metrics to be optimized by the algorithm, which is ['reward'] by default. When specified, it should have only 1 item for single-objective algorithm and can have multiple items for algorithms that support multi-objective optimization.	`None`
`**kwargs`	`Any`	Arguments passed to the `Backend` subclass registered with the requested backend.	`{}`

Yields:

Type	Description
`Any`	Tuples of (example, feedback), with examples sampled from the search
`Feedback`	space defined by `space` through `algorithm`.

Raises:

Type	Description
`ValueError`	If `space` is a fixed value, or the requested `backend` is not available.

Source code in pygx/tuning/_sample.py

def sample(
    space: HyperValue | hyper.DynamicEvaluationContext | geno.DNASpec,
    algorithm: geno.DNAGenerator,
    num_examples: int | None = None,
    early_stopping_policy: EarlyStoppingPolicy | None = None,
    where: Callable[[hyper.HyperPrimitive], bool] | None = None,
    name: str | None = None,
    group: None | int | str = None,
    backend: str | None = None,
    metrics_to_optimize: Sequence[str] | None = None,
    **kwargs: Any,
) -> Iterator[tuple[Any, Feedback]]:
    """Yields an example and its feedback sampled from a hyper value.

    Example 1: sample a search space defined by a symbolic hyper value:

    ```python
    import math
    import pygx as pg

    for example, feedback in pg.sample(
        pg.Dict(x=pg.floatv(-1, 1)),
        pg.geno.Random(),
        num_examples=10,
        name='my_search'):

      # We can access trial ID (starting from 1) and DNA from the feedback.
      print(feedback.id, feedback.dna)

      # We can report the reward computed on the example using
      # `feedback.add_measurement`, which can be called
      # multiple times to report the rewards incrementally.
      # Once a trial is done, we call `feedback.done` to mark evaluation on
      # current example as completed, or use `feedback.skip` to move to the
      # next sample without passing any feedback to the algorithm.
      # Without `feedback.done` or `feedback.skip`, the same trial will be
      # iterated over and over again for failover handling purpose.
      # Besides the reward and the step, metrics and checkpoint can be added
      # to each measurement. Additional meta-data and related links (URLs) can
      # be passed to `feedback.done` which can be retrieved via
      # `pg.poll_result` later.
      if example.x >= 0:
        # If we only want to add one measurement for each example, a
        # shortcut expression for the next two lines can be written as
        # follows:
        #   `feedback(reward=math.sqrt(example.x), step=1)`
        feedback.add_measurement(reward=math.sqrt(example.x), step=1)
        feedback.done()
      else:
        feedback.skip()

      # IMPORTANT: to stop the loop among all workers, we can call
      # `feedback.end_loop`. As a result, each worker will quit their loop
      # after current iteration, while using `break` in the for-loop is
      # only effective for the local process rather than remotely.
      if feedback.id > 1000:
        feedback.end_loop()

    # At any time, we can poll the search result via `pg.poll_result`, please
    # see `pg.tuning.Result` for more details.
    result = pg.poll_result('my_search')
    print(result.best_trial)
    ```

    Example 2: sample a search space defined by `pg.hyper.trace`:

    ```python
    import pygx as pg

    def fun():
      return pg.oneof([
        lambda: pg.oneof([1, 2, 3]),
        lambda: pg.floatv(0.1, 1.0),
        3]) + sum(pg.manyof(2, [1, 2, 3]))

    for example, feedback in pg.sample(
        pg.hyper.trace(fun),
        pg.geno.Random(),
        num_examples=10,
        name='my_search'):
      # When space is a `pg.hyper.DynamicEvaluationContext` object,
      # the `example` yielded at each iteration is a context manager under which
      # the hyper primitives (e.g. pg.oneof) will be materialized into concrete
      # values according to the controller decision.
      with example():
        reward = fun()
      feedback(reward)
    ```

    Example 3: sample DNAs from an abstract search space represented by
    `pg.DNASpec`:

    ```python
    import pygx as pg

    def evaluate_dna(dna):
      return sum(dna.to_numbers())

    for dna, feedback in pg.sample(
        pg.dna_spec(pg.List([pg.oneof(range(3)) for _ in range(5)])),
        pg.geno.Random(),
        num_examples=10,
        name='my_search'):
      reward = evaluate_dna(dna)
      feedback(reward)
    ```

    **Using `pg.sample` in distributed environment**

    `pg.sample` is designed with distributed sampling in mind, in which
    multiple processes can work on the trials of the same sampling loop. While
    the default 'in-memory' backend works only within a single process without
    failover handling, other backends may support distributed computing
    environment with persistent state. Nevertheless, the `pg.sample` API is
    the same among different backends, users can switch the backend easily by
    passing a different value to the `backend` argument, or set the default
    value globally via `pg.tuning.set_default_backend`.


    **Identifying a sampling**

    To identify a distributed loop, a unique `name` is introduced, which will
    also be used to poll the latest sampling result via `pg.poll_result`.

    **Failover handling**

    In a distributed setup, worker processes may incidentally die and restart.
    Unless a trial is explicitly marked as done (via `feedback(reward)` or
    `feedback.done()`) or skipped (via `feedback.skip()`), a worker will try to
    resume its work on the trial from where it left off.

    **Workgroup**

    In a distributed setup, worker processes may or may not work on the same
    trials. Worker group is introduced to serve this purpose, which is
    identified by an integer or a string named `group`. If `group` is not
    specified or having different values among the workers, every worker will
    work on different trials of the loop. On the contrary, workers having the
    same `group` will co-work on the same trials. Group is useful when
    evaluation on one example can be parallelized - for example - an example
    in the outer loop of a nested search. However, feedback on the same
    example should be fed back to the search algorithm only once. Therefore,
    workers in the same group need to communicate with each other to avoid
    duplicated evaluation and feedbacks. To facilitate such communication,
    per-trial metadata is supported, and can be accessed via
    `feedback.set_metadata/get_metadata` methods. The consistency in
    reading and writing the metadata is defined by the backend used. For the
    'in-memory' backend, all the trials and their metadata is stored in memory,
    thus will be lost if the process gets restarted. On the contrary, the
    backends built on distributed computing environment may store both the
    trials and metadata in a persistent storage with varying read/write QPS
    and read/write consistency guarantees.

    **Switch between backends**

    The `backend` argument of `pg.sample` lets users choose a backend used in
    current sampling loop. Users can use different backends in the same
    process to achieve a best performance trade-off, e.g., using `in-memory`
    backend when the communication cost outweighs the redundant evaluation
    cost upon worker failure.

    Helper function
    [`pygx.tuning.set_default_backend`][pygx.tuning.set_default_backend]
    is introduced to set the default tuning backend for the entire process.

    Args:
      space: One of (a hyper value, a `pg.hyper.DynamicEvaluationContext`,
        a `pg.DNASpec`) to sample from.
        A hyper value is an object with to-be-determined values specified by
        `pg.oneof`, `pg.manyof`, `pg.floatv` and etc, representing a search
        space.
        A `pg.hyper.DynamicEvaluationContext` object represents a search space
        that is traced via dynamic evaluation.
        A `pg.DNASpec` represents an abstract search space that emits DNAs.
      algorithm: The search algorithm that samples the search space. For
        example: `pg.geno.Random()`,
        `pg.algo.evolution.regularized_evolution(...)`, and etc.
      num_examples: An optional integer as the max number of examples to
        sample. If None, sample will return an iterator of infinite examples.
      early_stopping_policy: An optional early stopping policy for user to tell
        if incremental evaluation (which reports multiple measurements) on each
        example can be early short circuited.
        After each call to `feedback.add_measurement`, users can use method
        `feedback.should_stop_early` to check whether current example worth
        further evaluation or not.
      where: Function to filter the hyper values. If None, all decision points
        from the `space` will be included for the algorithm to make
        decisions. Otherwise only the decision points on which 'where' returns
        True will be included. The remaining decision points are passed
        through in the example intact, so each example is itself a sub-space
        of `space`. `where` is usually used in nested search flows. See
        `pg.hyper.template` for details.
      name: A string as a unique identifier for current sampling. Two separate
        calls to `pg.sample` with the same `name` (also the same algorithm) will
        share the same sampling queue, whose examples are proposed by the same
        search algorithm.
      group: A string or integer as the group ID of current process in
        distributed sampling, which will be used to group different workers into
        co-worker groups. Workers with the same group id will work on the same
        trial. On the contrary, workers in different groups will always be
        working with different trials. If not specified, each worker in
        current sampling will be in different groups. `group` is usually used
        in the outer
        loops of nested search, in order to allow workers to work on the same
        higher-order item.
      backend: An optional string to specify the backend name for sampling.
        if None, the default backend set by `pg.tuning.set_default_backend`
        will be used.
      metrics_to_optimize: A sequence of string as the names of the metrics
        to be optimized by the algorithm, which is ['reward'] by default.
        When specified, it should have only 1 item for single-objective
        algorithm and can have multiple items for algorithms that support
        multi-objective optimization.
      **kwargs: Arguments passed to the `Backend` subclass registered with
        the requested backend.

    Yields:
      Tuples of (example, feedback), with examples sampled from the search
      space defined by `space` through `algorithm`.

    Raises:
      ValueError: If `space` is a fixed value, or the requested `backend`
        is not available.
    """
    # Placeholder for Google-internal usage instrumentation.

    # Create template based on the hyper value.
    if isinstance(space, hyper.DynamicEvaluationContext):
        dynamic_evaluation_context = space
        dna_spec = space.dna_spec
        template = None
    elif isinstance(space, geno.DNASpec):
        dynamic_evaluation_context = None
        dna_spec = space
        template = None
    else:
        if symbolic.is_deterministic(space):
            raise ValueError(f"'space' is a constant value: {space!r}.")
        template = hyper.template(space, where)
        dna_spec = template.dna_spec()
        dynamic_evaluation_context = None

    # Create and set up the backend.
    metrics_to_optimize = metrics_to_optimize or ['reward']
    backend_instance = backend_lib.get_backend_cls(
        backend or backend_lib.default_backend()
    ).create(
        name,
        group,
        dna_spec,
        algorithm,
        metrics_to_optimize,
        early_stopping_policy,
        num_examples,
        **kwargs,
    )

    while True:
        try:
            feedback = backend_instance.next()
            dna = feedback.dna
            reward = dna.metadata.get('reward')
            if reward is None:
                # Decode and return current example to client code for evaluation.
                if template is not None:
                    value = template.decode(dna)
                elif dynamic_evaluation_context is not None:
                    # pylint: disable-next=unnecessary-lambda-assignment
                    value = lambda: dynamic_evaluation_context.apply(dna)
                else:
                    value = dna
                yield (value, feedback)
            else:
                # Reward may be computed at the controller side, we can
                # short circuit the client-side evaluation for the current item.
                # Also, there can be multiple co-workers working on the same trial,
                # we ignore errors triggered by racing feedbacks, which will be
                # considered as no-op.
                with feedback.ignore_race_condition():
                    feedback(
                        reward, metadata=dict(client_evaluation_skipped=True)
                    )
        except StopIteration:
            return

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

pygx.tuning¶

tuning ¶

Backend ¶

create classmethod ¶

poll_result abstractmethod classmethod ¶

next abstractmethod ¶

EarlyStoppingPolicy ¶

dna_spec property ¶

setup ¶

should_stop_early abstractmethod ¶

recover ¶

Feedback ¶

id abstractmethod property ¶

dna abstractmethod property ¶

checkpoint_to_warm_start_from abstractmethod property ¶

add_measurement ¶

get_trial abstractmethod ¶

set_metadata abstractmethod ¶

get_metadata abstractmethod ¶

add_link abstractmethod ¶

done abstractmethod ¶

skip abstractmethod ¶

should_stop_early abstractmethod ¶

end_loop abstractmethod ¶

skip_on_exceptions ¶

ignore_race_condition ¶

Measurement ¶

RaceConditionError ¶

Result ¶

metadata abstractmethod property ¶

is_active abstractmethod property ¶

last_updated abstractmethod property ¶

trials abstractmethod property ¶

best_trial abstractmethod property ¶

Trial ¶

get_reward_for_feedback ¶

add_backend ¶

available_backends ¶

default_backend ¶

poll_result ¶

set_default_backend ¶

sample ¶

`pygx.tuning`¶

create `classmethod` ¶

poll_result `abstractmethod` `classmethod` ¶

next `abstractmethod` ¶

dna_spec `property` ¶

should_stop_early `abstractmethod` ¶

id `abstractmethod` `property` ¶

dna `abstractmethod` `property` ¶

checkpoint_to_warm_start_from `abstractmethod` `property` ¶

get_trial `abstractmethod` ¶

set_metadata `abstractmethod` ¶

get_metadata `abstractmethod` ¶

add_link `abstractmethod` ¶

done `abstractmethod` ¶

skip `abstractmethod` ¶

should_stop_early `abstractmethod` ¶

end_loop `abstractmethod` ¶

metadata `abstractmethod` `property` ¶

is_active `abstractmethod` `property` ¶

last_updated `abstractmethod` `property` ¶

trials `abstractmethod` `property` ¶

best_trial `abstractmethod` `property` ¶