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 (aka. 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	An 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	Arguments passed to the BackendFactory subclass registered with the requested backend.	{}

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: An 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: Arguments passed to the `BackendFactory` subclass registered
        with the requested backend.

    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 policy.

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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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)

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 source 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 all trials that contain all intermediate measurements.

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 source
    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 all trials that
    contain all intermediate measurements.

    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, started 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=15) 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::

      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=15)
        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 make
Trial	sure the return value is up-to-date not only locally, but among different
Trial	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 curent 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 curent 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

Move to next example without providing the feedback to the algorithm.

Source code in pygx/tuning/_protocols.py

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

should_stop_early abstractmethod

should_stop_early() -> bool

Whether progressive evaluation can be stopped early on current trial.

Source code in pygx/tuning/_protocols.py

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

end_loop abstractmethod

end_loop() -> None

Ends current sapling loop.

Source code in pygx/tuning/_protocols.py

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

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.

Usages::

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.

    Usages::

      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 B 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)

Usages:: 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 B 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`)

    Usages::
      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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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 raisen when the operations made to Feedback indicates 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 B calls feedback.add_measurement.

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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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]

Get reward for feedback.

Source code in pygx/tuning/_protocols.py

def get_reward_for_feedback(
    self, metric_names: Sequence[str] | None = None
) -> None | float | tuple[float]:
    """Get reward for feedback."""
    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)

Decorator to register a backend factory with name.

Source code in pygx/tuning/_backend.py

def add_backend(backend_name: str):
    """Decorator to register a backend factory with name."""

    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 tuning result by name.

Source code in pygx/tuning/_backend.py

def poll_result(name: str, backend: str | None = None, **kwargs) -> Result:
    """Gets tuning result by name."""
    return get_backend_cls(backend or default_backend()).poll_result(
        name, **kwargs
    )

set_default_backend

set_default_backend(backend_name: str)

Sets the default tuning backend name.

Source code in pygx/tuning/_backend.py

def set_default_backend(backend_name: str):
    """Sets the default tuning 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::

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 (staring 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 session.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::

def fun(): return pg.oneof([ lambda: pg.oneof([1, 2, 3]), lambda: pg.float(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::

for dna, feedback in pg.sample( pg.List([pg.oneof(range(3))] * 5).dna_spec(), 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.

Workroup

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 get restarted. On the contrary, the backends built on distributed computing environment may store both the trials and metadata in a persistent storage with varing read/write QPS and read/write consistency guarentees.

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 overweighs the redudant 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 rest decision points will be passed through in the example, intact, which is a sub-space of the search space represented by the space. where is usually used in nested search flows. Please see 'hyper.Template' docstr 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	An 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 BackendFactory subclass registered with the requested backend.	{}

Yields:

Type	Description
Any	An iterator of tuples (example, feedback) as examples sampled from the
Feedback	search space defined by the 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::

      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 (staring 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 session.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`::

      def fun():
        return pg.oneof([
          lambda: pg.oneof([1, 2, 3]),
          lambda: pg.float(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`::

      for dna, feedback in pg.sample(
          pg.List([pg.oneof(range(3))] * 5).dna_spec(),
          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.

    **Workroup**

    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 get restarted. On the contrary, the backends
    built on distributed computing environment may store both the trials and
    metadata in a persistent storage with varing read/write QPS and read/write
    consistency guarentees.

    **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 overweighs the redudant 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 rest decision points will be passed through
        in the example, intact, which is a sub-space of the search space
        represented by the `space`. `where` is usually used in nested search
        flows. Please see 'hyper.Template' docstr 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: An 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 `BackendFactory` subclass registered with
        the requested backend.

    Yields:
      An iterator of tuples (example, feedback) as examples sampled from the
      search space defined by the `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