Skip to content

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.

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 
 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
 99
100
101
102
103
@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.

    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(
            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(tree, env)

            return OpaqueSpace(stream, nested, tags)

        return dp.SpaceBuilder(
            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 
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
@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(
        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 
 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
@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(tree, env)

        return OpaqueSpace(stream, nested, tags)

    return dp.SpaceBuilder(
        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], StandardPolicy[N, P], SupportsStreamCombinators

A policy maps a tree with a given signature (contravariant parameter N) and inner policy type (covariant parameter P) to a search stream.

Values of this type can be built by combining a search policy and an inner policy using the & operator defined on type SearchPolicy.

Source code in src/delphyne/stdlib/policies.py 
 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
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@dataclass(frozen=True)
class Policy(
    Generic[N, P],
    StandardPolicy[N, P],
    st.SupportsStreamCombinators,
):
    """
    A policy maps a tree with a given signature (contravariant parameter
    N) and inner policy type (covariant parameter P) to a search stream.

    Values of this type can be built by combining a search policy and an
    inner policy using the `&` operator defined on type `SearchPolicy`.
    """

    _fn: "_PolicyFn[N, P]"

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

    def __rmatmul__(self, other: StreamTransformer) -> "Policy[N, P]":
        """
        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 or_else[M: Node, Q](
        self: "Policy[M, Q]", other: "Policy[M, Q]"
    ) -> "Policy[M, Q]":
        """
        Combine two policies into one that tries the first one,
        and if it yields no solution, tries the second one.
        """

        def policy[T](
            tree: dp.Tree[M, Q, T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            yield from self(tree, env).or_else(other(tree, env))

        return Policy(policy)

    @classmethod
    def with_env[M: Node, Q](
        cls, f: Callable[[PolicyEnv], "Policy[M, Q]"], /
    ) -> "Policy[M, Q]":
        """
        Create a policy that depends on the global policy environment.
        """

        def aux[T](tree: dp.Tree[M, Q, T], env: PolicyEnv) -> dp.StreamGen[T]:
            policy = f(env)
            yield from policy(tree, env)

        return Policy(aux)

    @classmethod
    def sequence[M: Node, Q](
        cls, policies: Iterable["Policy[M, Q]"], *, stop_on_reject: bool = True
    ) -> "Policy[M, Q]":
        def aux[T](tree: dp.Tree[M, Q, T], env: PolicyEnv) -> dp.StreamGen[T]:
            yield from Stream.sequence(
                (p(tree, env) for p in policies),
                stop_on_reject=stop_on_reject,
            )

        return Policy(aux)

    @classmethod
    def parallel[M: Node, Q](
        cls, policies: Sequence["Policy[M, Q]"]
    ) -> "Policy[M, Q]":
        def aux[T](tree: dp.Tree[M, Q, T], env: PolicyEnv) -> dp.StreamGen[T]:
            yield from Stream.parallel([p(tree, env) for p in policies])

        return Policy(aux)

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

        return Policy(policy)

__rmatmul__ 

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

Compose a search policy with a stream transformer.

Source code in src/delphyne/stdlib/policies.py 
60
61
62
63
64
65
66
def __rmatmul__(self, other: StreamTransformer) -> "Policy[N, P]":
    """
    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)

or_else 

or_else(other: Policy[M, Q]) -> Policy[M, Q]

Combine two policies into one that tries the first one, and if it yields no solution, tries the second one.

Source code in src/delphyne/stdlib/policies.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def or_else[M: Node, Q](
    self: "Policy[M, Q]", other: "Policy[M, Q]"
) -> "Policy[M, Q]":
    """
    Combine two policies into one that tries the first one,
    and if it yields no solution, tries the second one.
    """

    def policy[T](
        tree: dp.Tree[M, Q, T], env: PolicyEnv
    ) -> dp.StreamGen[T]:
        yield from self(tree, env).or_else(other(tree, env))

    return Policy(policy)

with_env classmethod 

with_env(f: Callable[[PolicyEnv], Policy[M, Q]]) -> Policy[M, Q]

Create a policy that depends on the global policy environment.

Source code in src/delphyne/stdlib/policies.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
@classmethod
def with_env[M: Node, Q](
    cls, f: Callable[[PolicyEnv], "Policy[M, Q]"], /
) -> "Policy[M, Q]":
    """
    Create a policy that depends on the global policy environment.
    """

    def aux[T](tree: dp.Tree[M, Q, T], env: PolicyEnv) -> dp.StreamGen[T]:
        policy = f(env)
        yield from policy(tree, env)

    return Policy(aux)

SearchPolicy dataclass

Bases: AbstractSearchPolicy[PolicyEnv, N], SupportsStreamCombinators

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 
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
229
230
231
232
233
234
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
263
@dataclass(frozen=True)
class SearchPolicy[N: Node](
    dp.AbstractSearchPolicy[PolicyEnv, N],
    st.SupportsStreamCombinators,
):
    """
    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.
        """

        def policy[T](
            tree: dp.Tree[N, P, T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            return self._fn(tree, env, other)

        return Policy(policy)

    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 iter(trans(self(tree, env, policy), env))

        return SearchPolicy(policy)

    def or_else[M: Node](
        self: "SearchPolicy[M]", other: "SearchPolicy[M]"
    ) -> "SearchPolicy[M]":
        """
        Combine two search policies into one that tries the first one,
        and if it yields no solution, tries the second one.
        """

        def policy[T](
            tree: dp.Tree[M, Any, T], env: PolicyEnv, policy: Any
        ) -> dp.StreamGen[T]:
            yield from self(tree, env, policy).or_else(
                other(tree, env, policy)
            )

        return SearchPolicy(policy)

    @classmethod
    def with_env[M: Node](
        cls, f: Callable[[PolicyEnv], "SearchPolicy[M]"], /
    ) -> "SearchPolicy[M]":
        """
        Create a search policy that depends on the global policy
        environment.
        """

        def aux[P, T](
            tree: dp.Tree[M, P, T], env: PolicyEnv, policy: P
        ) -> dp.StreamGen[T]:
            search_policy = f(env)
            yield from search_policy(tree, env, policy)

        return SearchPolicy(aux)

    @classmethod
    def sequence[M: Node](
        cls,
        policies: Iterable["SearchPolicy[M]"],
        *,
        stop_on_reject: bool = True,
    ) -> "SearchPolicy[M]":
        def aux[P, T](
            tree: dp.Tree[M, P, T], env: PolicyEnv, policy: P
        ) -> dp.StreamGen[T]:
            yield from Stream.sequence(
                (p(tree, env, policy) for p in policies),
                stop_on_reject=stop_on_reject,
            )

        return SearchPolicy(aux)

    @classmethod
    def parallel[M: Node](
        cls, policies: Sequence["SearchPolicy[M]"]
    ) -> "SearchPolicy[M]":
        def aux[P, T](
            tree: dp.Tree[M, P, T], env: PolicyEnv, policy: P
        ) -> dp.StreamGen[T]:
            yield from Stream.parallel(
                [p(tree, env, policy) for p in policies]
            )

        return SearchPolicy(aux)

__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 
170
171
172
173
174
175
176
177
178
179
180
def __and__[P](self, other: P) -> "Policy[N, P]":
    """
    Pair a search policy with an inner policy to form a policy.
    """

    def policy[T](
        tree: dp.Tree[N, P, T], env: PolicyEnv
    ) -> dp.StreamGen[T]:
        return self._fn(tree, env, other)

    return Policy(policy)

__rmatmul__ 

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

Compose a search policy with a stream transformer.

Source code in src/delphyne/stdlib/policies.py 
182
183
184
185
186
187
188
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)

or_else 

or_else(other: SearchPolicy[M]) -> SearchPolicy[M]

Combine two search policies into one that tries the first one, and if it yields no solution, tries the second one.

Source code in src/delphyne/stdlib/policies.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def or_else[M: Node](
    self: "SearchPolicy[M]", other: "SearchPolicy[M]"
) -> "SearchPolicy[M]":
    """
    Combine two search policies into one that tries the first one,
    and if it yields no solution, tries the second one.
    """

    def policy[T](
        tree: dp.Tree[M, Any, T], env: PolicyEnv, policy: Any
    ) -> dp.StreamGen[T]:
        yield from self(tree, env, policy).or_else(
            other(tree, env, policy)
        )

    return SearchPolicy(policy)

with_env classmethod 

with_env(f: Callable[[PolicyEnv], SearchPolicy[M]]) -> SearchPolicy[M]

Create a search policy that depends on the global policy environment.

Source code in src/delphyne/stdlib/policies.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
@classmethod
def with_env[M: Node](
    cls, f: Callable[[PolicyEnv], "SearchPolicy[M]"], /
) -> "SearchPolicy[M]":
    """
    Create a search policy that depends on the global policy
    environment.
    """

    def aux[P, T](
        tree: dp.Tree[M, P, T], env: PolicyEnv, policy: P
    ) -> dp.StreamGen[T]:
        search_policy = f(env)
        yield from search_policy(tree, env, policy)

    return SearchPolicy(aux)

PromptingPolicy dataclass

Bases: AbstractPromptingPolicy[PolicyEnv], SupportsStreamCombinators

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 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
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
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
@dataclass(frozen=True)
class PromptingPolicy(
    dp.AbstractPromptingPolicy[PolicyEnv],
    st.SupportsStreamCombinators,
):
    """
    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 or_else(self, other: "PromptingPolicy") -> "PromptingPolicy":
        """
        Combine two prompting policies into one that tries the first one,
        and if it yields no solution, tries the second one.
        """

        def policy[T](
            query: dp.AttachedQuery[T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            yield from self(query, env).or_else(other(query, env))

        return PromptingPolicy(policy)

    @classmethod
    def with_env(
        cls, f: Callable[[PolicyEnv], "PromptingPolicy"], /
    ) -> "PromptingPolicy":
        """
        Create a prompting policy that depends on the global policy
        environment.
        """

        def aux[T](
            query: dp.AttachedQuery[T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            prompting_policy = f(env)
            yield from prompting_policy(query, env)

        return PromptingPolicy(aux)

    @classmethod
    def sequence(
        cls,
        policies: Iterable["PromptingPolicy"],
        *,
        stop_on_reject: bool = True,
    ) -> "PromptingPolicy":
        def aux[T](
            query: dp.AttachedQuery[T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            yield from Stream.sequence(
                (p(query, env) for p in policies),
                stop_on_reject=stop_on_reject,
            )

        return PromptingPolicy(aux)

    @classmethod
    def parallel(
        cls, policies: Sequence["PromptingPolicy"]
    ) -> "PromptingPolicy":
        def aux[T](
            query: dp.AttachedQuery[T], env: PolicyEnv
        ) -> dp.StreamGen[T]:
            yield from Stream.parallel([p(query, env) for p in policies])

        return PromptingPolicy(aux)

    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 iter(trans(self(query, env), env))

        return PromptingPolicy(policy)

or_else 

or_else(other: PromptingPolicy) -> PromptingPolicy

Combine two prompting policies into one that tries the first one, and if it yields no solution, tries the second one.

Source code in src/delphyne/stdlib/policies.py 
361
362
363
364
365
366
367
368
369
370
371
372
def or_else(self, other: "PromptingPolicy") -> "PromptingPolicy":
    """
    Combine two prompting policies into one that tries the first one,
    and if it yields no solution, tries the second one.
    """

    def policy[T](
        query: dp.AttachedQuery[T], env: PolicyEnv
    ) -> dp.StreamGen[T]:
        yield from self(query, env).or_else(other(query, env))

    return PromptingPolicy(policy)

with_env classmethod 

with_env(f: Callable[[PolicyEnv], PromptingPolicy]) -> PromptingPolicy

Create a prompting policy that depends on the global policy environment.

Source code in src/delphyne/stdlib/policies.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@classmethod
def with_env(
    cls, f: Callable[[PolicyEnv], "PromptingPolicy"], /
) -> "PromptingPolicy":
    """
    Create a prompting policy that depends on the global policy
    environment.
    """

    def aux[T](
        query: dp.AttachedQuery[T], env: PolicyEnv
    ) -> dp.StreamGen[T]:
        prompting_policy = f(env)
        yield from prompting_policy(query, env)

    return PromptingPolicy(aux)

__rmatmul__ 

__rmatmul__(other: StreamTransformer) -> PromptingPolicy

Compose a prompting policy with a stream transformer.

Source code in src/delphyne/stdlib/policies.py 
419
420
421
422
423
424
425
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 
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
@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 | B | 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 iter(search_policy(new_tree, env, policy))

        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 
548
549
550
551
552
553
554
555
556
557
558
559
560
@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 | B | N]

Compose a contextual tree transformer with a search policy.

Source code in src/delphyne/stdlib/policies.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
def __rmatmul__[N: Node](
    self, search_policy: "SearchPolicy[B | N]"
) -> "SearchPolicy[A | B | 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 iter(search_policy(new_tree, env, policy))

    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 
292
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
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 
266
267
268
269
270
271
272
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 
275
276
277
278
279
280
281
282
283
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 
286
287
288
289
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 
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
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 
439
440
441
442
443
444
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 
447
448
449
450
451
452
453
454
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 
457
458
459
460
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 
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
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 
498
499
500
501
502
503
504
505
506
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 
509
510
511
512
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 
515
516
517
518
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(tree, env)

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(tree, env)

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