Basic Definitions

Opaque Spaces and Policies

OpaqueSpace dataclass

Bases: Generic[P, T], Space[T]

A space defined by a mapping from the ambient inner policy to a search stream.

Opaque spaces can be defined from strategy instances (StrategyInstance) or from queries (Query) via the using method. Crucially, policies are unaware of how an opaque space was created, preserving abstraction.

Parameters:

Name	Type	Description	Default
P		Type parameter for the ambient inner policy type.	required
T		Type parameter for the element type.	required

Attributes:

Name	Type	Description
stream	Callable[[PolicyEnv, P], Stream[T]]	Maps the ambient inner policy to a search stream.

Source code in src/delphyne/stdlib/opaque.py

@dataclass(frozen=True)
class OpaqueSpace(Generic[P, T], dp.Space[T]):
    """
    A space defined by a mapping from the ambient inner policy to a
    search stream.

    Opaque spaces can be defined from strategy instances
    (`StrategyInstance`) or from queries (`Query`) via the `using`
    method. Crucially, policies are unaware of how an opaque space was
    created, preserving abstraction.

    Parameters:
        P: Type parameter for the ambient inner policy type.
        T: Type parameter for the element type.

    Attributes:
        stream: Maps the ambient inner policy to a search stream.
    """

    stream: Callable[[PolicyEnv, P], Stream[T]]
    _source: dp.NestedTree[Any, Any, T] | dp.AttachedQuery[T]
    _tags: Sequence[dp.Tag]

    @override
    def source(self) -> dp.NestedTree[Any, Any, T] | dp.AttachedQuery[T]:
        return self._source

    @override
    def tags(self) -> Sequence[dp.Tag]:
        return self._tags

    @staticmethod
    def from_query[P1, T1](
        query: dp.AbstractQuery[T1],
        get_policy: Callable[[P1, Sequence[dp.Tag]], PromptingPolicy],
    ) -> "dp.SpaceBuilder[OpaqueSpace[P1, T1]]":
        """
        Create an opaque space from a query.

        The `Query.using` method is a more ergonomic wrapper.
        """

        def build(
            spawner: QuerySpawner, tags: Sequence[dp.Tag]
        ) -> OpaqueSpace[P1, T1]:
            attached = spawner(query)
            return OpaqueSpace(
                stream=(lambda env, pol: get_policy(pol, tags)(attached, env)),
                _source=attached,
                _tags=tags,
            )

        return dp.SpaceBuilder(
            build=lambda _, spawner, tags: build(spawner, tags),
            tags=query.default_tags(),
        )

    @staticmethod
    def from_strategy[N: dp.Node, P1, P2, T1](
        strategy: dp.StrategyComp[N, P2, T1],
        get_policy: Callable[[P1, Sequence[dp.Tag]], Policy[N, P2]],
    ) -> "dp.SpaceBuilder[OpaqueSpace[P1, T1]]":
        """
        Create an opaque space from a strategy instance.

        The `StrategyInstance.using` method is a more ergonomic wrapper.
        """

        def build(
            spawner: NestedTreeSpawner, tags: Sequence[dp.Tag]
        ) -> OpaqueSpace[P1, T1]:
            nested = spawner(strategy)

            def stream(env: PolicyEnv, policy: P1) -> Stream[T1]:
                tree = nested.spawn_tree()
                sub = get_policy(policy, tags)
                return sub.search(tree, env, sub.inner)

            return OpaqueSpace(stream, nested, tags)

        return dp.SpaceBuilder(
            build=lambda spawner, _, tags: build(spawner, tags),
            tags=strategy.default_tags(),
        )

from_query staticmethod

from_query(
    query: AbstractQuery[T1], get_policy: Callable[[P1, Sequence[Tag]], PromptingPolicy]
) -> SpaceBuilder[OpaqueSpace[P1, T1]]

Create an opaque space from a query.

The Query.using method is a more ergonomic wrapper.

Source code in src/delphyne/stdlib/opaque.py

@staticmethod
def from_query[P1, T1](
    query: dp.AbstractQuery[T1],
    get_policy: Callable[[P1, Sequence[dp.Tag]], PromptingPolicy],
) -> "dp.SpaceBuilder[OpaqueSpace[P1, T1]]":
    """
    Create an opaque space from a query.

    The `Query.using` method is a more ergonomic wrapper.
    """

    def build(
        spawner: QuerySpawner, tags: Sequence[dp.Tag]
    ) -> OpaqueSpace[P1, T1]:
        attached = spawner(query)
        return OpaqueSpace(
            stream=(lambda env, pol: get_policy(pol, tags)(attached, env)),
            _source=attached,
            _tags=tags,
        )

    return dp.SpaceBuilder(
        build=lambda _, spawner, tags: build(spawner, tags),
        tags=query.default_tags(),
    )

from_strategy staticmethod

from_strategy(
    strategy: StrategyComp[N, P2, T1],
    get_policy: Callable[[P1, Sequence[Tag]], Policy[N, P2]],
) -> SpaceBuilder[OpaqueSpace[P1, T1]]

Create an opaque space from a strategy instance.

The StrategyInstance.using method is a more ergonomic wrapper.

Source code in src/delphyne/stdlib/opaque.py

@staticmethod
def from_strategy[N: dp.Node, P1, P2, T1](
    strategy: dp.StrategyComp[N, P2, T1],
    get_policy: Callable[[P1, Sequence[dp.Tag]], Policy[N, P2]],
) -> "dp.SpaceBuilder[OpaqueSpace[P1, T1]]":
    """
    Create an opaque space from a strategy instance.

    The `StrategyInstance.using` method is a more ergonomic wrapper.
    """

    def build(
        spawner: NestedTreeSpawner, tags: Sequence[dp.Tag]
    ) -> OpaqueSpace[P1, T1]:
        nested = spawner(strategy)

        def stream(env: PolicyEnv, policy: P1) -> Stream[T1]:
            tree = nested.spawn_tree()
            sub = get_policy(policy, tags)
            return sub.search(tree, env, sub.inner)

        return OpaqueSpace(stream, nested, tags)

    return dp.SpaceBuilder(
        build=lambda spawner, _, tags: build(spawner, tags),
        tags=strategy.default_tags(),
    )

Opaque

Opaque: SpaceBuilder[OpaqueSpace[P, T]]

A convenience type alias for an opaque space builder.

IPDict

IPDict: Mapping[str, Policy[Any, Any] | PromptingPolicy]

Type of an Inner-Policy Dictionary.

Inner-Policy dictionaries allow to define strategies in a more concise way in exchange for less static type safety.

Normally, an inner policy type must be defined for every strategy, and opaque spaces are created from queries or strategy by passing the using method a mapping from the ambient inner policy to a proper sub-policy, often in the form of an anonymous function:

@dataclass class MyInnerPolicy:
    foo: PromptingPolicy # etc

def my_strategy() -> Strategy[Branch, MyInnerPolicy, str]:
    x = yield from branch(Foo().using(lambda p: p.foo)) # etc

As an alternative, one can have a strategy use an inner policy dictionary, by passing ellipses (...) to the using method:

def my_strategy() -> Strategy[Branch, IPDict, str]:
    x = yield from branch(Foo().using(...)) # etc

When doing so, a simple Python dictionary can be used as an inner policy, whose keys are space tags (the same tags can be referenced in demonstration tests). In the example above, and since a spaces induced by a query inherits its name as a tag by default, one can define an inner policy for my_strategy as:

{"Foo": foo_prompting_policy, ...}

A conjunction of tags can also be specified, separated by & (without spaces). For example, {"tag1&tag2": pp, ...} associates prompting policies pp to spaces with both tags tag1 and tag2. New tags can be added to a space builder using the SpaceBuilder.tagged method.

Info

If several entries of the inner policy dictionary apply for a given instance of .using(...), a runtime error is raised.

See tests/example_strategies:generate_number for another example.

Policy dataclass

Bases: Generic[N, P], AbstractPolicy[PolicyEnv, N, P]

A pair of a search policy and of an inner policy.

More preciely, a policy for trees with effects N (contravariant) gathers a search policy handling N along with an inner policy object of type P (covariant).

Values of this type can be built concisely using the & operator defined on type SearchPolicy.

Source code in src/delphyne/stdlib/policies.py

@dataclass(frozen=True)
class Policy(Generic[N, P], dp.AbstractPolicy[PolicyEnv, N, P]):
    """
    A pair of a search policy and of an inner policy.

    More preciely, a policy for trees with effects `N` (contravariant)
    gathers a search policy handling `N` along with an inner policy
    object of type `P` (covariant).

    Values of this type can be built concisely using the `&` operator
    defined on type `SearchPolicy`.
    """

    _search: "SearchPolicy[N]"
    _inner: P

    @property
    def search(self) -> "SearchPolicy[N]":
        return self._search

    @property
    def inner(self) -> P:
        return self._inner

SearchPolicy dataclass

Bases: AbstractSearchPolicy[PolicyEnv, N]

A search policy takes as arguments a tree with a given signature (covariant type parameter N), a global policy environment, and an inner policy with appropriate type, and returns a search stream.

SearchPolicy is a subclass of AbstractSearchPolicy, which provides convenience features such as support for the @ composition operator (for composing search policies with stream transformers and tree transformers) and the & operator for pairing a search policy with an inner policy.

Source code in src/delphyne/stdlib/policies.py

@dataclass(frozen=True)
class SearchPolicy[N: Node](dp.AbstractSearchPolicy[PolicyEnv, N]):
    """
    A search policy takes as arguments a tree with a given signature
    (covariant type parameter `N`), a global policy environment, and an
    inner policy with appropriate type, and returns a search stream.

    `SearchPolicy` is a subclass of `AbstractSearchPolicy`, which
    provides convenience features such as support for the `@`
    composition operator (for composing search policies with stream
    transformers and tree transformers) and the `&` operator for pairing
    a search policy with an inner policy.
    """

    _fn: "_SearchPolicyFn[N]"

    def __call__[P, T](
        self,
        tree: "dp.Tree[N, P, T]",
        env: PolicyEnv,
        policy: P,
    ) -> Stream[T]:
        return Stream(lambda: self._fn(tree, env, policy))

    def __and__[P](self, other: P) -> "Policy[N, P]":
        """
        Pair a search policy with an inner policy to form a policy.
        """
        return Policy(self, other)

    def __rmatmul__(self, other: StreamTransformer) -> "SearchPolicy[N]":
        """
        Compose a search policy with a stream transformer.
        """
        if not isinstance(other, StreamTransformer):  # pyright: ignore[reportUnnecessaryIsInstance]
            return NotImplemented
        return self._compose_with_stream_transformer(other)

    def _compose_with_stream_transformer(
        self,
        trans: StreamTransformer,
    ) -> "SearchPolicy[N]":
        def policy[P, T](
            tree: dp.Tree[N, P, T], env: PolicyEnv, policy: P
        ) -> dp.StreamGen[T]:
            return trans(self(tree, env, policy), env).gen()

        return SearchPolicy(policy)

__and__

__and__(other: P) -> Policy[N, P]

Pair a search policy with an inner policy to form a policy.

Source code in src/delphyne/stdlib/policies.py

def __and__[P](self, other: P) -> "Policy[N, P]":
    """
    Pair a search policy with an inner policy to form a policy.
    """
    return Policy(self, other)

__rmatmul__

__rmatmul__(other: StreamTransformer) -> SearchPolicy[N]

Compose a search policy with a stream transformer.

Source code in src/delphyne/stdlib/policies.py

def __rmatmul__(self, other: StreamTransformer) -> "SearchPolicy[N]":
    """
    Compose a search policy with a stream transformer.
    """
    if not isinstance(other, StreamTransformer):  # pyright: ignore[reportUnnecessaryIsInstance]
        return NotImplemented
    return self._compose_with_stream_transformer(other)

PromptingPolicy dataclass

Bases: AbstractPromptingPolicy[PolicyEnv]

A prompting policy takes as arguments a query (attached to a specific node) and a global policy environment, and returns a search stream (SearchStream).

PromptingPolicy is a subclass of AbstractPromptingPolicy, which provides convenience features such as support for the @ composition operator (for composing prompting policies with stream transformers).

Source code in src/delphyne/stdlib/policies.py

@dataclass(frozen=True)
class PromptingPolicy(dp.AbstractPromptingPolicy[PolicyEnv]):
    """
    A prompting policy takes as arguments a query (attached to a
    specific node) and a global policy environment, and returns a search
    stream (`SearchStream`).

    `PromptingPolicy` is a subclass of `AbstractPromptingPolicy`, which
    provides convenience features such as support for the `@`
    composition operator (for composing prompting policies with stream
    transformers).
    """

    _fn: "_PromptingPolicyFn"

    def __call__[T](
        self, query: dp.AttachedQuery[T], env: PolicyEnv
    ) -> Stream[T]:
        return Stream(lambda: self._fn(query, env))

    def __rmatmul__(self, other: StreamTransformer) -> "PromptingPolicy":
        """
        Compose a prompting policy with a stream transformer.
        """
        if not isinstance(other, StreamTransformer):  # pyright: ignore[reportUnnecessaryIsInstance]
            return NotImplemented
        return self._compose_with_stream_transformer(other)

    def _compose_with_stream_transformer(
        self,
        trans: StreamTransformer,
    ) -> "PromptingPolicy":
        def policy[T](
            query: dp.AttachedQuery[T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            return trans(self(query, env), env).gen()

        return PromptingPolicy(policy)

__rmatmul__

__rmatmul__(other: StreamTransformer) -> PromptingPolicy

Compose a prompting policy with a stream transformer.

Source code in src/delphyne/stdlib/policies.py

def __rmatmul__(self, other: StreamTransformer) -> "PromptingPolicy":
    """
    Compose a prompting policy with a stream transformer.
    """
    if not isinstance(other, StreamTransformer):  # pyright: ignore[reportUnnecessaryIsInstance]
        return NotImplemented
    return self._compose_with_stream_transformer(other)

ContextualTreeTransformer dataclass

Contextual tree transformer, which can be composed with search policies to modify their accepted signature.

Parameters:

Name	Type	Description	Default
A		The type of nodes that the transformer removes from search policy signature.	required
B		The type of nodes that the transformer adds to search policy signature (or the bottom type Never if no types are added).	required

Attributes:

Name	Type	Description
fn	_ContextualTreeTransformerFn[A, B]	A function that takes a policy environment and an inner policy as arguments (hence the contextual aspect) and returns a pure tree transformer (PureTreeTransformerFn)

Source code in src/delphyne/stdlib/policies.py

@dataclass
class ContextualTreeTransformer[A: Node, B: Node]:
    """
    Contextual tree transformer, which can be composed with search
    policies to modify their accepted signature.

    Parameters:
        A: The type of nodes that the transformer removes from search
            policy signature.
        B: The type of nodes that the transformer adds to search policy
            signature (or the bottom type `Never` if no types are
            added).

    Attributes:
        fn: A function that takes a policy environment and an inner
            policy as arguments (hence the *contextual* aspect) and
            returns a pure tree transformer (`PureTreeTransformerFn`)
    """

    fn: _ContextualTreeTransformerFn[A, B]

    @staticmethod
    def pure(
        fn: PureTreeTransformerFn[A, B],
    ) -> "ContextualTreeTransformer[A, B]":
        """
        Create a contextual tree transformer from a pure tree
        transformer.
        """

        def contextual(env: PolicyEnv, policy: Any):
            return fn

        return ContextualTreeTransformer(contextual)

    def __rmatmul__[N: Node](
        self, search_policy: "SearchPolicy[B | N]"
    ) -> "SearchPolicy[A | N]":
        """
        Compose a contextual tree transformer with a search policy.
        """
        if not isinstance(search_policy, SearchPolicy):  # pyright: ignore[reportUnnecessaryIsInstance]
            return NotImplemented

        def new_search_policy[P, T](
            tree: dp.Tree[A | N, P, T],
            env: PolicyEnv,
            policy: P,
        ) -> dp.StreamGen[T]:
            new_tree = self.fn(env, policy)(tree)
            return search_policy(new_tree, env, policy).gen()

        return SearchPolicy(new_search_policy)

pure staticmethod

pure(fn: PureTreeTransformerFn[A, B]) -> ContextualTreeTransformer[A, B]

Create a contextual tree transformer from a pure tree transformer.

Source code in src/delphyne/stdlib/policies.py

@staticmethod
def pure(
    fn: PureTreeTransformerFn[A, B],
) -> "ContextualTreeTransformer[A, B]":
    """
    Create a contextual tree transformer from a pure tree
    transformer.
    """

    def contextual(env: PolicyEnv, policy: Any):
        return fn

    return ContextualTreeTransformer(contextual)

__rmatmul__

__rmatmul__(search_policy: SearchPolicy[B | N]) -> SearchPolicy[A | N]

Compose a contextual tree transformer with a search policy.

Source code in src/delphyne/stdlib/policies.py

def __rmatmul__[N: Node](
    self, search_policy: "SearchPolicy[B | N]"
) -> "SearchPolicy[A | N]":
    """
    Compose a contextual tree transformer with a search policy.
    """
    if not isinstance(search_policy, SearchPolicy):  # pyright: ignore[reportUnnecessaryIsInstance]
        return NotImplemented

    def new_search_policy[P, T](
        tree: dp.Tree[A | N, P, T],
        env: PolicyEnv,
        policy: P,
    ) -> dp.StreamGen[T]:
        new_tree = self.fn(env, policy)(tree)
        return search_policy(new_tree, env, policy).gen()

    return SearchPolicy(new_search_policy)

Convenience Decorators

search_policy

search_policy(fn: _ParametricSearchPolicyFn[N, A]) -> _ParametricSearchPolicy[N, A]

Convenience decorator for creating parametric search policies (i.e., functions that return search policies).

See dfs for an example.

Attributes:

Name	Type	Description
fn		A function that takes a tree, a policy environment, an inner policy, and additional parameters as arguments and returns a search stream generator (SearchStreamGen).

Returns:

Type	Description
_ParametricSearchPolicy[N, A]	A function that takes the additional parameters of fn as
_ParametricSearchPolicy[N, A]	arguments and returns a search policy (SearchPolicy).

Source code in src/delphyne/stdlib/policies.py

def search_policy[N: Node, **A](
    fn: _ParametricSearchPolicyFn[N, A],
) -> _ParametricSearchPolicy[N, A]:
    """
    Convenience decorator for creating parametric search policies (i.e.,
    functions that return search policies).

    See `dfs` for an example.

    Attributes:
        fn: A function that takes a tree, a policy environment, an inner
            policy, and additional parameters as arguments and returns a
            search stream generator (`SearchStreamGen`).

    Returns:
        A function that takes the additional parameters of `fn` as
        arguments and returns a search policy (`SearchPolicy`).
    """

    def parametric(*args: A.args, **kwargs: A.kwargs) -> SearchPolicy[N]:
        def policy[T](
            tree: dp.Tree[N, Any, T], env: PolicyEnv, policy: Any
        ) -> dp.StreamGen[T]:
            return fn(tree, env, policy, *args, **kwargs)

        return SearchPolicy(policy)

    return parametric

_SearchPolicyFn

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _SearchPolicyFn[N: Node](Protocol):
    def __call__[P, T](
        self,
        tree: dp.Tree[N, P, T],
        env: PolicyEnv,
        policy: P,
    ) -> dp.StreamGen[T]: ...

_ParametricSearchPolicyFn

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _ParametricSearchPolicyFn[N: Node, **A](Protocol):
    def __call__[P, T](
        self,
        tree: dp.Tree[N, P, T],
        env: PolicyEnv,
        policy: P,
        *args: A.args,
        **kwargs: A.kwargs,
    ) -> dp.StreamGen[T]: ...

_ParametricSearchPolicy

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _ParametricSearchPolicy[N: Node, **A](Protocol):
    def __call__(
        self, *args: A.args, **kwargs: A.kwargs
    ) -> SearchPolicy[N]: ...

prompting_policy

prompting_policy(
    fn: _ParametricPromptingPolicyFn[A],
) -> _ParametricPromptingPolicy[A]

Convenience decorator for creating parametric prompting policies (i.e., functions that return prompting policies).

See the definition of few_shot for an example.

Attributes:

Name	Type	Description
fn		A function that takes an attached query, a policy environment, and additional parameters as arguments and returns a search stream generator (SearchStreamGen).

Returns:

Type	Description
_ParametricPromptingPolicy[A]	A function that takes the additional parameters of fn as
_ParametricPromptingPolicy[A]	arguments and returns a prompting policy (PromptingPolicy).

Source code in src/delphyne/stdlib/policies.py

def prompting_policy[**A](
    fn: _ParametricPromptingPolicyFn[A],
) -> _ParametricPromptingPolicy[A]:
    """
    Convenience decorator for creating parametric prompting policies
    (i.e., functions that return prompting policies).

    See the definition of `few_shot` for an example.

    Attributes:
        fn: A function that takes an attached query, a policy
            environment, and additional parameters as arguments and
            returns a search stream generator (`SearchStreamGen`).

    Returns:
        A function that takes the additional parameters of `fn` as
        arguments and returns a prompting policy (`PromptingPolicy`).
    """

    def parametric(*args: A.args, **kwargs: A.kwargs) -> PromptingPolicy:
        def policy[T](
            query: dp.AttachedQuery[T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            return fn(query, env, *args, **kwargs)

        return PromptingPolicy(policy)

    return parametric

_PromptingPolicyFn

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _PromptingPolicyFn(Protocol):
    def __call__[T](
        self,
        query: dp.AttachedQuery[T],
        env: PolicyEnv,
    ) -> dp.StreamGen[T]: ...

_ParametricPromptingPolicyFn

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _ParametricPromptingPolicyFn[**A](Protocol):
    def __call__[T](
        self,
        query: dp.AttachedQuery[T],
        env: PolicyEnv,
        *args: A.args,
        **kwargs: A.kwargs,
    ) -> dp.StreamGen[T]: ...

_ParametricPromptingPolicy

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _ParametricPromptingPolicy[**A](Protocol):
    def __call__(
        self, *args: A.args, **kwargs: A.kwargs
    ) -> PromptingPolicy: ...

contextual_tree_transformer

contextual_tree_transformer(
    f: _ParametricContextualTreeTransformerFn[A, B, C],
) -> Callable[C, ContextualTreeTransformer[A, B]]

A convenience decorator for defining contextual tree transformers.

See the implementation of elim_messages for an example.

Parameters:

Name	Type	Description	Default
f	_ParametricContextualTreeTransformerFn[A, B, C]	A function that takes a policy environment, an inner policy, and additional parameters as arguments and returns a pure tree transformer (PureTreeTransformerFn).	required

Returns:

Type	Description
Callable[C, ContextualTreeTransformer[A, B]]	A function that takes the additional parameters of f as
Callable[C, ContextualTreeTransformer[A, B]]	arguments and returns a contextual tree transformer
Callable[C, ContextualTreeTransformer[A, B]]	(ContextualTreeTransformer).

Source code in src/delphyne/stdlib/policies.py

def contextual_tree_transformer[A: Node, B: Node, **C](
    f: _ParametricContextualTreeTransformerFn[A, B, C], /
) -> Callable[C, ContextualTreeTransformer[A, B]]:
    """
    A convenience decorator for defining contextual tree transformers.

    See the implementation of `elim_messages` for an example.

    Arguments:
        f: A function that takes a policy environment, an inner policy,
            and additional parameters as arguments and returns a pure
            tree transformer (`PureTreeTransformerFn`).

    Returns:
        A function that takes the additional parameters of `f` as
        arguments and returns a contextual tree transformer
        (`ContextualTreeTransformer`).
    """

    def parametric(*args: C.args, **kwargs: C.kwargs):
        def contextual(env: PolicyEnv, policy: Any):
            return f(env, policy, *args, **kwargs)

        return ContextualTreeTransformer(contextual)

    return parametric

PureTreeTransformerFn

Bases: Protocol

A function that maps any tree with signature A | N to a tree with signature B | N, for all N.

Source code in src/delphyne/stdlib/policies.py

class PureTreeTransformerFn[A: Node, B: Node](Protocol):
    """
    A function that maps any tree with signature `A | N` to a tree with
    signature `B | N`, for all `N`.
    """

    def __call__[N: Node, P, T](
        self, tree: dp.Tree[A | N, P, T]
    ) -> dp.Tree[B | N, P, T]: ...

_ContextualTreeTransformerFn

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _ContextualTreeTransformerFn[A: Node, B: Node](Protocol):
    def __call__(
        self, env: PolicyEnv, policy: Any
    ) -> PureTreeTransformerFn[A, B]: ...

_ParametricContextualTreeTransformerFn

Bases: Protocol

Source code in src/delphyne/stdlib/policies.py

class _ParametricContextualTreeTransformerFn[A: Node, B: Node, **C](Protocol):
    def __call__(
        self, env: PolicyEnv, policy: Any, *args: C.args, **kwargs: C.kwargs
    ) -> PureTreeTransformerFn[A, B]: ...

Strategies

StrategyInstance dataclass

Bases: StrategyComp[N, P, T]

A strategy computation that can be reified into a search tree, obtained by instantiating a strategy function.

StrategyInstance is a subclass of StrategyComp that adds convenience features such as the using method for building opaque spaces. The strategy decorator can be used to wrap strategy functions so as to have them return StrategyInstance objects.

Source code in src/delphyne/stdlib/strategies.py

@dataclass(frozen=True)
class StrategyInstance[N: dp.Node, P, T](dp.StrategyComp[N, P, T]):
    """
    A *strategy computation* that can be reified into a search tree,
    obtained by instantiating a strategy function.

    `StrategyInstance` is a subclass of `StrategyComp` that adds
    convenience features such as the `using` method for building opaque
    spaces. The `strategy` decorator can be used to wrap strategy
    functions so as to have them return `StrategyInstance` objects.
    """

    @overload
    def using(self, get_policy: EllipsisType, /) -> Opaque[IPDict, T]: ...

    @overload
    def using[Pout](
        self,
        get_policy: Callable[[Pout], Policy[N, P]] | EllipsisType,
        /,
        inner_policy_type: type[Pout] | None = None,
    ) -> Opaque[Pout, T]: ...

    def using[Pout](
        self,
        get_policy: Callable[[Pout], Policy[N, P]] | EllipsisType,
        /,
        inner_policy_type: type[Pout] | None = None,
    ) -> Opaque[Pout, T]:
        """
        Turn a strategy instance into an opaque space by providing a
        mapping from the ambient inner policy to an appropriate policy.

        Attributes:
            get_policy: A function that maps the ambient inner policy to
                a policy (i.e., a pair of a search policy and of an
                inner policy) to use for answering the query.
                Alternatively, if the ellipsis value `...` is passed,
                the inner policy type is assumed to be `IPDict`, and
                sub-policies are automatically selected using tags (see
                `IPDict` documentation).
            inner_policy_type: Ambient inner policy type for the outer
                strategy from which the strategy is called. This
                information is not used at runtime but it can be
                provided to help type inference when necessary.

        Parameters:
            P: Inner policy type associated with the strategy.
            Pout: Ambient inner policy type associated with the outer
                strategy from which the strategy is called.
        """

        # Using operators such as `&` or `@` instead of using does not
        # work well since some type checkers (e.g., Pyright) perform
        # worse inference when using those instead of a standard method.
        if isinstance(get_policy, EllipsisType):
            return OpaqueSpace[Pout, T].from_strategy(
                self, cast(Any, pol.dict_subpolicy)
            )
        return OpaqueSpace[Pout, T].from_strategy(
            self, lambda p, _: get_policy(p)
        )

    def run_toplevel(
        self,
        env: PolicyEnv,
        policy: Policy[N, P],
        monitor: dp.TreeMonitor = dp.TreeMonitor(),
    ) -> Stream[T]:
        """
        Reify a strategy into a tree and run it using a given policy.
        """
        tree = dp.reify(self, monitor)
        return policy.search(tree, env, policy.inner)

using

using(get_policy: EllipsisType) -> Opaque[IPDict, T]

using(
    get_policy: Callable[[Pout], Policy[N, P]] | EllipsisType,
    /,
    inner_policy_type: type[Pout] | None = None,
) -> Opaque[Pout, T]

using(
    get_policy: Callable[[Pout], Policy[N, P]] | EllipsisType,
    /,
    inner_policy_type: type[Pout] | None = None,
) -> Opaque[Pout, T]

Turn a strategy instance into an opaque space by providing a mapping from the ambient inner policy to an appropriate policy.

Attributes:

Name	Type	Description
get_policy		A function that maps the ambient inner policy to a policy (i.e., a pair of a search policy and of an inner policy) to use for answering the query. Alternatively, if the ellipsis value ... is passed, the inner policy type is assumed to be IPDict, and sub-policies are automatically selected using tags (see IPDict documentation).
inner_policy_type		Ambient inner policy type for the outer strategy from which the strategy is called. This information is not used at runtime but it can be provided to help type inference when necessary.

Parameters:

Name	Type	Description	Default
P		Inner policy type associated with the strategy.	required
Pout		Ambient inner policy type associated with the outer strategy from which the strategy is called.	required

Source code in src/delphyne/stdlib/strategies.py

def using[Pout](
    self,
    get_policy: Callable[[Pout], Policy[N, P]] | EllipsisType,
    /,
    inner_policy_type: type[Pout] | None = None,
) -> Opaque[Pout, T]:
    """
    Turn a strategy instance into an opaque space by providing a
    mapping from the ambient inner policy to an appropriate policy.

    Attributes:
        get_policy: A function that maps the ambient inner policy to
            a policy (i.e., a pair of a search policy and of an
            inner policy) to use for answering the query.
            Alternatively, if the ellipsis value `...` is passed,
            the inner policy type is assumed to be `IPDict`, and
            sub-policies are automatically selected using tags (see
            `IPDict` documentation).
        inner_policy_type: Ambient inner policy type for the outer
            strategy from which the strategy is called. This
            information is not used at runtime but it can be
            provided to help type inference when necessary.

    Parameters:
        P: Inner policy type associated with the strategy.
        Pout: Ambient inner policy type associated with the outer
            strategy from which the strategy is called.
    """

    # Using operators such as `&` or `@` instead of using does not
    # work well since some type checkers (e.g., Pyright) perform
    # worse inference when using those instead of a standard method.
    if isinstance(get_policy, EllipsisType):
        return OpaqueSpace[Pout, T].from_strategy(
            self, cast(Any, pol.dict_subpolicy)
        )
    return OpaqueSpace[Pout, T].from_strategy(
        self, lambda p, _: get_policy(p)
    )

run_toplevel

run_toplevel(
    env: PolicyEnv, policy: Policy[N, P], monitor: TreeMonitor = TreeMonitor()
) -> Stream[T]

Reify a strategy into a tree and run it using a given policy.

Source code in src/delphyne/stdlib/strategies.py

def run_toplevel(
    self,
    env: PolicyEnv,
    policy: Policy[N, P],
    monitor: dp.TreeMonitor = dp.TreeMonitor(),
) -> Stream[T]:
    """
    Reify a strategy into a tree and run it using a given policy.
    """
    tree = dp.reify(self, monitor)
    return policy.search(tree, env, policy.inner)

strategy

strategy(
    f: Callable[A, Strategy[N, P, T]],
) -> Callable[A, StrategyInstance[N, P, T]]

strategy(
    *,
    name: str | None = None,
    ret: TypeAnnot[Any] | NoTypeInfo = NoTypeInfo(),
    inherit_tags: Callable[..., Sequence[SpaceBuilder[Any]]] | None = None,
) -> _StrategyDecorator

strategy(*dec_args: Any, **dec_kwargs: Any) -> Any

Standard parametric decorator for wrapping strategy functions into functions returning StrategyInstance objects.

Parameters:

Name	Type	Description	Default
name	optional	Name of the strategy. If not provided, the name attribute of the strategy function is used instead. The name of the strategy is used in defining default tags and when visualizing traces.	required
ret	optional	Return type of the strategy function. If not provided, it is obtained by inspecting type annotations. This information is used when visualizing traces and for serializing top-level success values when running commands.	required
inherit_tags	optional	A function that maps all arguments from the decorated strategy function to a sequence of space builders from which tags must be inherited. By default, nothing is inherited.	required

Info

strategy()(f) can be shortened as @strategy(f), hence the overloading of the type of strategy.

Runtime use of type annotations

The type annotations for the arguments and return type of a strategy function are leveraged at runtime in two ways:

• To improve the rendering of values when visualizing traces (e.g., using YAML serialization instead of pprint).
• To unserialize arguments for the top-level strategy when specified in JSON or YAML and serialize its return value.

In summary, type annotations are fully optional, except when trying to unserialize (resp. serialize) the arguments (resp. return type) of a top-level strategy involving custom data types (and not just JSON values such as integers, strings, dictionaries...).

Source code in src/delphyne/stdlib/strategies.py

def strategy(*dec_args: Any, **dec_kwargs: Any) -> Any:
    """
    Standard parametric decorator for wrapping strategy functions into
    functions returning `StrategyInstance` objects.

    Parameters:
        name (optional): Name of the strategy. If not provided, the
            __name__ attribute of the strategy function is used instead.
            The name of the strategy is used in defining default tags
            and when visualizing traces.
        ret (optional): Return type of the strategy function. If not
            provided, it is obtained by inspecting type annotations.
            This information is used when visualizing traces and for
            serializing top-level success values when running commands.
        inherit_tags (optional): A function that maps all arguments from
            the decorated strategy function to a sequence of space
            builders from which tags must be inherited. By default,
            nothing is inherited.

    !!! info
        `strategy()(f)` can be shortened as `@strategy(f)`, hence the
        overloading of the type of `strategy`.

    ??? info "Runtime use of type annotations"
        The type annotations for the arguments and return type of a
        strategy function are leveraged at runtime in two ways:

        - To improve the rendering of values when visualizing traces
          (e.g., using YAML serialization instead of `pprint`).
        - To unserialize arguments for the top-level strategy when
          specified in JSON or YAML and serialize its return value.

        In summary, type annotations are fully optional, except when
        trying to unserialize (resp. serialize) the arguments (resp.
        return type) of a top-level strategy involving custom data types
        (and not just JSON values such as integers, strings,
        dictionaries...).
    """

    # Using functools.wraps is important so that the object loader can
    # get the type hints to properly instantiate arguments.
    error_msg = "Invalid use of the @strategy decorator."
    # If no positional argument is provided, we have a call of the form
    # `@strategy(...)(f)`. Otherwise, we have a call of the form
    # `@strategy(f)`.
    assert len(dec_args) in [0, 1], error_msg

    # @strategy(f) case
    if not dec_kwargs and len(dec_args) == 1 and callable(dec_args[0]):
        f: Any = dec_args[0]
        name = inspect.function_name(f)
        tags = (name,) if name else ()

        def wrapped(*args: Any, **kwargs: Any):
            return StrategyInstance(
                f,
                args,
                kwargs,
                _name=None,
                _return_type=NoTypeInfo(),
                _tags=tags,
            )

        return functools.wraps(f)(wrapped)

    # @strategy(...)(f) case
    elif not dec_args:

        def decorator(f: Any):
            def wrapped(*args: Any, **kwargs: Any):
                name = dec_kwargs.get("name")
                if name is None:
                    name = inspect.function_name(f)
                tags = (name,) if name else ()
                # Inherit tags from space arguments if needed.
                inherited_fun = dec_kwargs.get("inherit_tags", None)
                if inherited_fun is not None:
                    inherited = inherited_fun(*args, **kwargs)
                    for space in inherited:
                        assert isinstance(space, dp.SpaceBuilder)
                        tags = (*tags, *space.tags)
                return StrategyInstance(
                    f,
                    args,
                    kwargs,
                    _name=name,
                    _return_type=dec_kwargs.get("ret", NoTypeInfo()),
                    _tags=tags,
                )

            return functools.wraps(f)(wrapped)

        return decorator

    assert False, error_msg

_StrategyDecorator

Bases: Protocol

Type of the strategy decorator, after is optional arguments are instantiated.

Source code in src/delphyne/stdlib/strategies.py

class _StrategyDecorator(Protocol):
    """
    Type of the `strategy` decorator, after is optional arguments are
    instantiated.
    """

    # Note that this definition cannot be inlined in the return type of
    # `@strategy`, without which variables such as `P` and `T` would not
    # be correctly scoped and inferred (type checkers such as Pyright
    # would set them to `Unknown` if they cannot be inferred from `args`
    # in `@strategy(*args)(f)`).

    def __call__[**A, N: dp.Node, P, T](
        self,
        f: Callable[A, dp.Strategy[N, P, T]],
    ) -> Callable[A, StrategyInstance[N, P, T]]: ...