Skip to content

Basic Definitions

Opaque Spaces and Policies

OpaqueSpace dataclass

Bases: 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.

Class Type Parameters:

Name Bound or Constraints 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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@dataclass(frozen=True)
class OpaqueSpace[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.

    Type 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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@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 precisely, 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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@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 precisely, 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.

Search policies can be conveniently defined using the search_policy decorator. See dfs for an example.

Source code in src/delphyne/stdlib/policies.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@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.

    Search policies can be conveniently defined using the
    `search_policy` decorator. See `dfs` for an example.
    """

    _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 
80
81
82
83
84
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 
86
87
88
89
90
91
92
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).

Prompting policies can be conveniently defined using the prompting_policy decorator. See the definition of few_shot for an example.

Source code in src/delphyne/stdlib/policies.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@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).

    Prompting policies can be conveniently defined using the
    `prompting_policy` decorator. See the definition of `few_shot` for
    an example.
    """

    _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 
191
192
193
194
195
196
197
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

Wrapper for a function that maps trees to trees, possibly changing their signature. Can depend on the global policy environment (hence the contextual aspect).

Contextual tree transformers can be composed with search policies to modify their accepted signature. They can be convniently defined using the contextual_tree_transformer decorator. See elim_compute and elim_messages for examples.

Class Type Parameters:

Name Bound or Constraints Description Default
A Node

The type of nodes that the transformer removes from search policy signature.

 required
B Node

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 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
@dataclass
class ContextualTreeTransformer[A: Node, B: Node]:
    """
    Wrapper for a function that maps trees to trees, possibly
    changing their signature. Can depend on the global policy
    environment (hence the *contextual* aspect).

    Contextual tree transformers can be composed with search policies to
    modify their accepted signature. They can be convniently defined
    using the `contextual_tree_transformer` decorator. See
    `elim_compute` and `elim_messages` for examples.

    Type 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 
320
321
322
323
324
325
326
327
328
329
330
331
332
@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 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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 
106
107
108
109
110
111
112
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 
115
116
117
118
119
120
121
122
123
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 
126
127
128
129
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 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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 
211
212
213
214
215
216
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 
219
220
221
222
223
224
225
226
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 
229
230
231
232
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 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
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 
270
271
272
273
274
275
276
277
278
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 
281
282
283
284
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 
287
288
289
290
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.

Class Type Parameters:

Name Bound or Constraints Description Default
N Node

Signature of the strategy.

 required
P

Inner policy type associated with the strategy.

 required
T

Return type of the strategy.

 required
Source code in src/delphyne/stdlib/strategies.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@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.

    Type Parameters:
        N: Signature of the strategy.
        P: Inner policy type associated with the strategy.
        T: Return type of the strategy.
    """

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

        Type Parameters:
            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.

Type Parameters:

Name Bound or Constraints Description Default
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 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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.

    Type Parameters:
        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 
87
88
89
90
91
92
93
94
95
96
97
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(
    f: Callable[..., Any] | None = None,
    /,
    *,
    name: str | None = None,
    ret: TypeAnnot[Any] | NoTypeInfo = NoTypeInfo(),
    inherit_tags: Callable[..., Sequence[SpaceBuilder[Any]]] | None = None,
) -> 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.

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

 NoTypeInfo()
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.

 None

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 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def strategy(
    f: Callable[..., Any] | None = None,
    /,
    *,
    name: str | None = None,
    ret: TypeAnnot[Any] | NoTypeInfo = NoTypeInfo(),
    inherit_tags: Callable[..., Sequence[dp.SpaceBuilder[Any]]] | None = None,
) -> 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.

    # @strategy(f) case
    if f is not None:
        assert name is None
        assert isinstance(ret, NoTypeInfo)
        assert inherit_tags is None
        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
    else:

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

            return functools.wraps(f)(wrapped)

        return decorator

_StrategyDecorator

Bases: Protocol

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

Source code in src/delphyne/stdlib/strategies.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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]]: ...